Skip to content

P
python-poetry
Commit fdbe6e3a by Randy Döring
Options

refactor(repositories): unify find_packages of different repository types 

Co-authored-by: David Hotham <david.hotham@blueyonder.co.uk>
parent d4e55fd3
Showing with 105 additions and 133 deletions
src/poetry/repositories/legacy_repository.py
...@@ -14,8 +14,8 @@ from poetry.repositories.link_sources.html import SimpleRepositoryPage ...@@ -14,8 +14,8 @@ from poetry.repositories.link_sources.html import SimpleRepositoryPage
if TYPE_CHECKING: if TYPE_CHECKING:
from poetry.core.packages.dependency import Dependency
from poetry.core.packages.utils.link import Link from poetry.core.packages.utils.link import Link
from poetry.core.semver.version_constraint import VersionConstraint
from poetry.config.config import Config from poetry.config.config import Config
...@@ -33,61 +33,6 @@ class LegacyRepository(HTTPRepository): ...@@ -33,61 +33,6 @@ class LegacyRepository(HTTPRepository):
super().__init__(name, url.rstrip("/"), config, disable_cache) super().__init__(name, url.rstrip("/"), config, disable_cache)
def find_packages(self, dependency: Dependency) -> list[Package]:
packages = []
constraint, allow_prereleases = self._get_constraints_from_dependency(
dependency
)
key = dependency.name
if not constraint.is_any():
key = f"{key}:{constraint!s}"
ignored_pre_release_versions = []
if self._cache.store("matches").has(key):
versions = self._cache.store("matches").get(key)
else:
page = self._get_page(f"/{dependency.name}/")
if page is None:
return []
versions = []
for version in page.versions(dependency.name):
if version.is_unstable() and not allow_prereleases:
if constraint.is_any():
# we need this when all versions of the package are pre-releases
ignored_pre_release_versions.append(version)
continue
if constraint.allows(version):
versions.append(version)
self._cache.store("matches").put(key, versions, 5)
for package_versions in (versions, ignored_pre_release_versions):
for version in package_versions:
package = Package(
dependency.name,
version,
source_type="legacy",
source_reference=self.name,
source_url=self._url,
)
packages.append(package)
self._log(
f"{len(packages)} packages found for {dependency.name} {constraint!s}",
level="debug",
)
if packages or not constraint.is_any():
# we have matching packages, or constraint is not (*)
break
return packages
def package( def package(
self, name: str, version: str, extras: list[str] | None = None self, name: str, version: str, extras: list[str] | None = None
) -> Package: ) -> Package:
...@@ -121,6 +66,43 @@ class LegacyRepository(HTTPRepository): ...@@ -121,6 +66,43 @@ class LegacyRepository(HTTPRepository):
return list(page.links_for_version(package.name, package.version)) return list(page.links_for_version(package.name, package.version))
def _find_packages(self, name: str, constraint: VersionConstraint) -> list[Package]:
"""
Find packages on the remote server.
"""
versions: list[Version]
key = name
if not constraint.is_any():
key = f"{key}:{constraint!s}"
if self._cache.store("matches").has(key):
versions = self._cache.store("matches").get(key)
else:
page = self._get_page(f"/{name}/")
if page is None:
self._log(
f"No packages found for {name}",
level="debug",
)
return []
versions = [
version for version in page.versions(name) if constraint.allows(version)
]
self._cache.store("matches").put(key, versions, 5)
return [
Package(
name,
version,
source_type="legacy",
source_reference=self.name,
source_url=self._url,
)
for version in versions
]
def _get_release_info(self, name: str, version: str) -> dict[str, Any]: def _get_release_info(self, name: str, version: str) -> dict[str, Any]:
page = self._get_page(f"/{canonicalize_name(name)}/") page = self._get_page(f"/{canonicalize_name(name)}/")
if page is None: if page is None:
......
src/poetry/repositories/pypi_repository.py
...@@ -12,6 +12,7 @@ from cachecontrol.controller import logger as cache_control_logger ...@@ -12,6 +12,7 @@ from cachecontrol.controller import logger as cache_control_logger
from html5lib.html5parser import parse from html5lib.html5parser import parse
from poetry.core.packages.package import Package from poetry.core.packages.package import Package
from poetry.core.packages.utils.link import Link from poetry.core.packages.utils.link import Link
from poetry.core.semver.version import Version
from poetry.core.version.exceptions import InvalidVersion from poetry.core.version.exceptions import InvalidVersion
from poetry.repositories.exceptions import PackageNotFound from poetry.repositories.exceptions import PackageNotFound
...@@ -26,7 +27,7 @@ logger = logging.getLogger(__name__) ...@@ -26,7 +27,7 @@ logger = logging.getLogger(__name__)
if TYPE_CHECKING: if TYPE_CHECKING:
from poetry.core.packages.dependency import Dependency from poetry.core.semver.version_constraint import VersionConstraint
class PyPiRepository(HTTPRepository): class PyPiRepository(HTTPRepository):
...@@ -43,62 +44,6 @@ class PyPiRepository(HTTPRepository): ...@@ -43,62 +44,6 @@ class PyPiRepository(HTTPRepository):
self._base_url = url self._base_url = url
self._fallback = fallback self._fallback = fallback
def find_packages(self, dependency: Dependency) -> list[Package]:
"""
Find packages on the remote server.
"""
constraint, allow_prereleases = self._get_constraints_from_dependency(
dependency
)
try:
info = self.get_package_info(dependency.name)
except PackageNotFound:
self._log(
f"No packages found for {dependency.name} {constraint!s}",
level="debug",
)
return []
packages = []
ignored_pre_release_packages = []
for version, release in info["releases"].items():
if not release:
# Bad release
self._log(
f"No release information found for {dependency.name}-{version},"
" skipping",
level="debug",
)
continue
try:
package = Package(info["info"]["name"], version)
except InvalidVersion:
self._log(
f'Unable to parse version "{version}" for the'
f" {dependency.name} package, skipping",
level="debug",
)
continue
if package.is_prerelease() and not allow_prereleases:
if constraint.is_any():
# we need this when all versions of the package are pre-releases
ignored_pre_release_packages.append(package)
continue
if constraint.allows(package.version):
packages.append(package)
self._log(
f"{len(packages)} packages found for {dependency.name} {constraint!s}",
level="debug",
)
return packages or ignored_pre_release_packages
def search(self, query: str) -> list[Package]: def search(self, query: str) -> list[Package]:
results = [] results = []
...@@ -160,6 +105,46 @@ class PyPiRepository(HTTPRepository): ...@@ -160,6 +105,46 @@ class PyPiRepository(HTTPRepository):
) )
return package_info return package_info
def _find_packages(self, name: str, constraint: VersionConstraint) -> list[Package]:
"""
Find packages on the remote server.
"""
try:
info = self.get_package_info(name)
except PackageNotFound:
self._log(
f"No packages found for {name} {constraint!s}",
level="debug",
)
return []
packages = []
for version_string, release in info["releases"].items():
if not release:
# Bad release
self._log(
f"No release information found for {name}-{version_string},"
" skipping",
level="debug",
)
continue
try:
version = Version.parse(version_string)
except InvalidVersion:
self._log(
f'Unable to parse version "{version_string}" for the'
f" {name} package, skipping",
level="debug",
)
continue
if constraint.allows(version):
packages.append(Package(info["info"]["name"], version))
return packages
def _get_package_info(self, name: str) -> dict[str, Any]: def _get_package_info(self, name: str) -> dict[str, Any]:
data = self._get(f"pypi/{name}/json") data = self._get(f"pypi/{name}/json")
if data is None: if data is None:
......
src/poetry/repositories/repository.py
...@@ -35,30 +35,28 @@ class Repository: ...@@ -35,30 +35,28 @@ class Repository:
def find_packages(self, dependency: Dependency) -> list[Package]: def find_packages(self, dependency: Dependency) -> list[Package]:
packages = [] packages = []
ignored_pre_release_packages = []
constraint, allow_prereleases = self._get_constraints_from_dependency( constraint, allow_prereleases = self._get_constraints_from_dependency(
dependency dependency
) )
ignored_pre_release_packages = []
for package in self.packages: for package in self._find_packages(dependency.name, constraint):
if dependency.name == package.name: if (
if ( package.is_prerelease()
package.is_prerelease() and not allow_prereleases
and not allow_prereleases and not package.is_direct_origin()
and not package.source_type ):
): if constraint.is_any():
# If prereleases are not allowed and the package is a prerelease # we need this when all versions of the package are pre-releases
# and is a standard package then we skip it ignored_pre_release_packages.append(package)
if constraint.is_any(): continue
# we need this when all versions of the package are pre-releases
ignored_pre_release_packages.append(package) packages.append(package)
continue
self._log(
if constraint.allows(package.version) or ( f"{len(packages)} packages found for {dependency.name} {constraint!s}",
package.is_prerelease() level="debug",
and constraint.allows(package.version.next_patch()) )
):
packages.append(package)
return packages or ignored_pre_release_packages return packages or ignored_pre_release_packages
...@@ -114,6 +112,13 @@ class Repository: ...@@ -114,6 +112,13 @@ class Repository:
return constraint, allow_prereleases return constraint, allow_prereleases
def _find_packages(self, name: str, constraint: VersionConstraint) -> list[Package]:
return [
package
for package in self._packages
if package.name == name and constraint.allows(package.version)
]
def _log(self, msg: str, level: str = "info") -> None: def _log(self, msg: str, level: str = "info") -> None:
logger = logging.getLogger(f"{__name__}.{self.__class__.__name__}") logger = logging.getLogger(f"{__name__}.{self.__class__.__name__}")
getattr(logger, level)(f"<c1>Source ({self.name}):</c1> {msg}") getattr(logger, level)(f"<c1>Source ({self.name}):</c1> {msg}")
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment