Skip to content

Commit

Permalink
move export code into export plugin
Browse files Browse the repository at this point in the history
  • Loading branch information
@radoering
radoering committed Aug 7, 2022
1 parent 601dc78 commit 3f1d53b
Showing 1 changed file with 0 additions and 178 deletions.
178 changes: 0 additions & 178 deletions src/poetry/packages/locker.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,6 @@
import os
import re

from copy import deepcopy
from hashlib import sha256
from pathlib import Path
from typing import TYPE_CHECKING
Expand All @@ -27,20 +26,12 @@
from tomlkit.exceptions import TOMLKitError
from tomlkit.items import Array

from poetry.packages import DependencyPackage
from poetry.utils.extras import get_extra_package_names


if TYPE_CHECKING:
from collections.abc import Iterable
from collections.abc import Iterator
from collections.abc import Sequence

from poetry.core.packages.directory_dependency import DirectoryDependency
from poetry.core.packages.file_dependency import FileDependency
from poetry.core.packages.url_dependency import URLDependency
from poetry.core.packages.vcs_dependency import VCSDependency
from poetry.core.version.markers import BaseMarker
from tomlkit.items import Table
from tomlkit.toml_document import TOMLDocument

Expand Down Expand Up @@ -205,175 +196,6 @@ def locked_repository(self) -> Repository:

return packages

@staticmethod
def __get_locked_package(
dependency: Dependency,
packages_by_name: dict[str, list[Package]],
decided: dict[Package, Dependency] | None = None,
) -> Package | None:
"""
Internal helper to identify corresponding locked package using dependency
version constraints.
"""
decided = decided or {}

# Get the packages that are consistent with this dependency.
packages = [
package
for package in packages_by_name.get(dependency.name, [])
if package.python_constraint.allows_all(dependency.python_constraint)
and dependency.constraint.allows(package.version)
]

# If we've previously made a choice that is compatible with the current
# requirement, stick with it.
for package in packages:
old_decision = decided.get(package)
if (
old_decision is not None
and not old_decision.marker.intersect(dependency.marker).is_empty()
):
return package

return next(iter(packages), None)

@classmethod
def __walk_dependencies(
cls,
dependencies: list[Dependency],
packages_by_name: dict[str, list[Package]],
) -> dict[Package, Dependency]:
nested_dependencies: dict[Package, Dependency] = {}

visited: set[tuple[Dependency, BaseMarker]] = set()
while dependencies:
requirement = dependencies.pop(0)
if (requirement, requirement.marker) in visited:
continue
visited.add((requirement, requirement.marker))

locked_package = cls.__get_locked_package(
requirement, packages_by_name, nested_dependencies
)

if not locked_package:
raise RuntimeError(f"Dependency walk failed at {requirement}")

if requirement.extras:
locked_package = locked_package.with_features(requirement.extras)

# create dependency from locked package to retain dependency metadata
# if this is not done, we can end-up with incorrect nested dependencies
constraint = requirement.constraint
marker = requirement.marker
requirement = locked_package.to_dependency()
requirement.marker = requirement.marker.intersect(marker)

requirement.constraint = constraint

for require in locked_package.requires:
if require.is_optional() and not any(
require in locked_package.extras[feature]
for feature in locked_package.features
):
continue

require = deepcopy(require)
require.marker = require.marker.intersect(
requirement.marker.without_extras()
)
if not require.marker.is_empty():
dependencies.append(require)

key = locked_package
if key not in nested_dependencies:
nested_dependencies[key] = requirement
else:
nested_dependencies[key].marker = nested_dependencies[key].marker.union(
requirement.marker
)

return nested_dependencies

@classmethod
def get_project_dependencies(
cls,
project_requires: list[Dependency],
locked_packages: list[Package],
) -> Iterable[tuple[Package, Dependency]]:
# group packages entries by name, this is required because requirement might use
# different constraints.
packages_by_name: dict[str, list[Package]] = {}
for pkg in locked_packages:
if pkg.name not in packages_by_name:
packages_by_name[pkg.name] = []
packages_by_name[pkg.name].append(pkg)

# Put higher versions first so that we prefer them.
for packages in packages_by_name.values():
packages.sort(
key=lambda package: package.version,
reverse=True,
)

nested_dependencies = cls.__walk_dependencies(
dependencies=project_requires,
packages_by_name=packages_by_name,
)

return nested_dependencies.items()

def get_project_dependency_packages(
self,
project_requires: list[Dependency],
project_python_marker: BaseMarker | None = None,
extras: bool | Sequence[str] | None = None,
) -> Iterator[DependencyPackage]:
# Apply the project python marker to all requirements.
if project_python_marker is not None:
marked_requires: list[Dependency] = []
for require in project_requires:
require = deepcopy(require)
require.marker = require.marker.intersect(project_python_marker)
marked_requires.append(require)
project_requires = marked_requires

repository = self.locked_repository()

# Build a set of all packages required by our selected extras
extra_package_names: set[str] | None = None

if extras is not True:
extra_package_names = set(
get_extra_package_names(
repository.packages,
self.lock_data.get("extras", {}),
extras or (),
)
)

# If a package is optional and we haven't opted in to it, do not select
selected = []
for dependency in project_requires:
try:
package = repository.find_packages(dependency=dependency)[0]
except IndexError:
continue

if extra_package_names is not None and (
package.optional and package.name not in extra_package_names
):
# a package is locked as optional, but is not activated via extras
continue

selected.append(dependency)

for package, dependency in self.get_project_dependencies(
project_requires=selected,
locked_packages=repository.packages,
):
yield DependencyPackage(dependency=dependency, package=package)

def set_lock_data(self, root: Package, packages: list[Package]) -> bool:
files: dict[str, Any] = table()
package_specs = self._lock_packages(packages)
Expand Down

0 comments on commit 3f1d53b

Please sign in to comment.