check.py 5.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163
  1. """Validation of dependencies of packages
  2. """
  3. # The following comment should be removed at some point in the future.
  4. # mypy: strict-optional=False
  5. # mypy: disallow-untyped-defs=False
  6. import logging
  7. from collections import namedtuple
  8. from pip._vendor.packaging.utils import canonicalize_name
  9. from pip._vendor.pkg_resources import RequirementParseError
  10. from pip._internal.distributions import (
  11. make_distribution_for_install_requirement,
  12. )
  13. from pip._internal.utils.misc import get_installed_distributions
  14. from pip._internal.utils.typing import MYPY_CHECK_RUNNING
  15. logger = logging.getLogger(__name__)
  16. if MYPY_CHECK_RUNNING:
  17. from pip._internal.req.req_install import InstallRequirement
  18. from typing import (
  19. Any, Callable, Dict, Optional, Set, Tuple, List
  20. )
  21. # Shorthands
  22. PackageSet = Dict[str, 'PackageDetails']
  23. Missing = Tuple[str, Any]
  24. Conflicting = Tuple[str, str, Any]
  25. MissingDict = Dict[str, List[Missing]]
  26. ConflictingDict = Dict[str, List[Conflicting]]
  27. CheckResult = Tuple[MissingDict, ConflictingDict]
  28. PackageDetails = namedtuple('PackageDetails', ['version', 'requires'])
  29. def create_package_set_from_installed(**kwargs):
  30. # type: (**Any) -> Tuple[PackageSet, bool]
  31. """Converts a list of distributions into a PackageSet.
  32. """
  33. # Default to using all packages installed on the system
  34. if kwargs == {}:
  35. kwargs = {"local_only": False, "skip": ()}
  36. package_set = {}
  37. problems = False
  38. for dist in get_installed_distributions(**kwargs):
  39. name = canonicalize_name(dist.project_name)
  40. try:
  41. package_set[name] = PackageDetails(dist.version, dist.requires())
  42. except RequirementParseError as e:
  43. # Don't crash on broken metadata
  44. logger.warning("Error parsing requirements for %s: %s", name, e)
  45. problems = True
  46. return package_set, problems
  47. def check_package_set(package_set, should_ignore=None):
  48. # type: (PackageSet, Optional[Callable[[str], bool]]) -> CheckResult
  49. """Check if a package set is consistent
  50. If should_ignore is passed, it should be a callable that takes a
  51. package name and returns a boolean.
  52. """
  53. if should_ignore is None:
  54. def should_ignore(name):
  55. return False
  56. missing = {}
  57. conflicting = {}
  58. for package_name in package_set:
  59. # Info about dependencies of package_name
  60. missing_deps = set() # type: Set[Missing]
  61. conflicting_deps = set() # type: Set[Conflicting]
  62. if should_ignore(package_name):
  63. continue
  64. for req in package_set[package_name].requires:
  65. name = canonicalize_name(req.project_name) # type: str
  66. # Check if it's missing
  67. if name not in package_set:
  68. missed = True
  69. if req.marker is not None:
  70. missed = req.marker.evaluate()
  71. if missed:
  72. missing_deps.add((name, req))
  73. continue
  74. # Check if there's a conflict
  75. version = package_set[name].version # type: str
  76. if not req.specifier.contains(version, prereleases=True):
  77. conflicting_deps.add((name, version, req))
  78. if missing_deps:
  79. missing[package_name] = sorted(missing_deps, key=str)
  80. if conflicting_deps:
  81. conflicting[package_name] = sorted(conflicting_deps, key=str)
  82. return missing, conflicting
  83. def check_install_conflicts(to_install):
  84. # type: (List[InstallRequirement]) -> Tuple[PackageSet, CheckResult]
  85. """For checking if the dependency graph would be consistent after \
  86. installing given requirements
  87. """
  88. # Start from the current state
  89. package_set, _ = create_package_set_from_installed()
  90. # Install packages
  91. would_be_installed = _simulate_installation_of(to_install, package_set)
  92. # Only warn about directly-dependent packages; create a whitelist of them
  93. whitelist = _create_whitelist(would_be_installed, package_set)
  94. return (
  95. package_set,
  96. check_package_set(
  97. package_set, should_ignore=lambda name: name not in whitelist
  98. )
  99. )
  100. def _simulate_installation_of(to_install, package_set):
  101. # type: (List[InstallRequirement], PackageSet) -> Set[str]
  102. """Computes the version of packages after installing to_install.
  103. """
  104. # Keep track of packages that were installed
  105. installed = set()
  106. # Modify it as installing requirement_set would (assuming no errors)
  107. for inst_req in to_install:
  108. abstract_dist = make_distribution_for_install_requirement(inst_req)
  109. dist = abstract_dist.get_pkg_resources_distribution()
  110. name = canonicalize_name(dist.key)
  111. package_set[name] = PackageDetails(dist.version, dist.requires())
  112. installed.add(name)
  113. return installed
  114. def _create_whitelist(would_be_installed, package_set):
  115. # type: (Set[str], PackageSet) -> Set[str]
  116. packages_affected = set(would_be_installed)
  117. for package_name in package_set:
  118. if package_name in packages_affected:
  119. continue
  120. for req in package_set[package_name].requires:
  121. if canonicalize_name(req.name) in packages_affected:
  122. packages_affected.add(package_name)
  123. break
  124. return packages_affected