You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
153 lines
5.2 KiB
153 lines
5.2 KiB
"""Validation of dependencies of packages |
|
""" |
|
|
|
import logging |
|
from typing import TYPE_CHECKING, Callable, Dict, List, NamedTuple, Optional, Set, Tuple |
|
|
|
from pip._vendor.packaging.requirements import Requirement |
|
from pip._vendor.packaging.utils import canonicalize_name |
|
|
|
from pip._internal.distributions import make_distribution_for_install_requirement |
|
from pip._internal.metadata import get_default_environment |
|
from pip._internal.metadata.base import DistributionVersion |
|
from pip._internal.req.req_install import InstallRequirement |
|
|
|
if TYPE_CHECKING: |
|
from pip._vendor.packaging.utils import NormalizedName |
|
|
|
logger = logging.getLogger(__name__) |
|
|
|
|
|
class PackageDetails(NamedTuple): |
|
version: DistributionVersion |
|
dependencies: List[Requirement] |
|
|
|
|
|
# Shorthands |
|
PackageSet = Dict['NormalizedName', PackageDetails] |
|
Missing = Tuple['NormalizedName', Requirement] |
|
Conflicting = Tuple['NormalizedName', DistributionVersion, Requirement] |
|
|
|
MissingDict = Dict['NormalizedName', List[Missing]] |
|
ConflictingDict = Dict['NormalizedName', List[Conflicting]] |
|
CheckResult = Tuple[MissingDict, ConflictingDict] |
|
ConflictDetails = Tuple[PackageSet, CheckResult] |
|
|
|
|
|
def create_package_set_from_installed() -> Tuple[PackageSet, bool]: |
|
"""Converts a list of distributions into a PackageSet.""" |
|
package_set = {} |
|
problems = False |
|
env = get_default_environment() |
|
for dist in env.iter_installed_distributions(local_only=False, skip=()): |
|
name = dist.canonical_name |
|
try: |
|
dependencies = list(dist.iter_dependencies()) |
|
package_set[name] = PackageDetails(dist.version, dependencies) |
|
except (OSError, ValueError) as e: |
|
# Don't crash on unreadable or broken metadata. |
|
logger.warning("Error parsing requirements for %s: %s", name, e) |
|
problems = True |
|
return package_set, problems |
|
|
|
|
|
def check_package_set(package_set, should_ignore=None): |
|
# type: (PackageSet, Optional[Callable[[str], bool]]) -> CheckResult |
|
"""Check if a package set is consistent |
|
|
|
If should_ignore is passed, it should be a callable that takes a |
|
package name and returns a boolean. |
|
""" |
|
|
|
missing = {} |
|
conflicting = {} |
|
|
|
for package_name, package_detail in package_set.items(): |
|
# Info about dependencies of package_name |
|
missing_deps = set() # type: Set[Missing] |
|
conflicting_deps = set() # type: Set[Conflicting] |
|
|
|
if should_ignore and should_ignore(package_name): |
|
continue |
|
|
|
for req in package_detail.dependencies: |
|
name = canonicalize_name(req.name) |
|
|
|
# Check if it's missing |
|
if name not in package_set: |
|
missed = True |
|
if req.marker is not None: |
|
missed = req.marker.evaluate() |
|
if missed: |
|
missing_deps.add((name, req)) |
|
continue |
|
|
|
# Check if there's a conflict |
|
version = package_set[name].version |
|
if not req.specifier.contains(version, prereleases=True): |
|
conflicting_deps.add((name, version, req)) |
|
|
|
if missing_deps: |
|
missing[package_name] = sorted(missing_deps, key=str) |
|
if conflicting_deps: |
|
conflicting[package_name] = sorted(conflicting_deps, key=str) |
|
|
|
return missing, conflicting |
|
|
|
|
|
def check_install_conflicts(to_install): |
|
# type: (List[InstallRequirement]) -> ConflictDetails |
|
"""For checking if the dependency graph would be consistent after \ |
|
installing given requirements |
|
""" |
|
# Start from the current state |
|
package_set, _ = create_package_set_from_installed() |
|
# Install packages |
|
would_be_installed = _simulate_installation_of(to_install, package_set) |
|
|
|
# Only warn about directly-dependent packages; create a whitelist of them |
|
whitelist = _create_whitelist(would_be_installed, package_set) |
|
|
|
return ( |
|
package_set, |
|
check_package_set( |
|
package_set, should_ignore=lambda name: name not in whitelist |
|
) |
|
) |
|
|
|
|
|
def _simulate_installation_of(to_install, package_set): |
|
# type: (List[InstallRequirement], PackageSet) -> Set[NormalizedName] |
|
"""Computes the version of packages after installing to_install. |
|
""" |
|
# Keep track of packages that were installed |
|
installed = set() |
|
|
|
# Modify it as installing requirement_set would (assuming no errors) |
|
for inst_req in to_install: |
|
abstract_dist = make_distribution_for_install_requirement(inst_req) |
|
dist = abstract_dist.get_pkg_resources_distribution() |
|
|
|
assert dist is not None |
|
name = canonicalize_name(dist.project_name) |
|
package_set[name] = PackageDetails(dist.parsed_version, dist.requires()) |
|
|
|
installed.add(name) |
|
|
|
return installed |
|
|
|
|
|
def _create_whitelist(would_be_installed, package_set): |
|
# type: (Set[NormalizedName], PackageSet) -> Set[NormalizedName] |
|
packages_affected = set(would_be_installed) |
|
|
|
for package_name in package_set: |
|
if package_name in packages_affected: |
|
continue |
|
|
|
for req in package_set[package_name].dependencies: |
|
if canonicalize_name(req.name) in packages_affected: |
|
packages_affected.add(package_name) |
|
break |
|
|
|
return packages_affected
|
|
|