1from __future__ import annotations
2
3import importlib.metadata
4import logging
5import os
6import pathlib
7import sys
8import zipfile
9from collections.abc import Iterator, Sequence
10
11from pip._vendor.packaging.utils import (
12 InvalidWheelFilename,
13 NormalizedName,
14 canonicalize_name,
15 parse_wheel_filename,
16)
17
18from pip._internal.metadata.base import BaseDistribution, BaseEnvironment
19from pip._internal.utils.filetypes import WHEEL_EXTENSION
20
21from ._compat import BadMetadata, BasePath, get_dist_canonical_name, get_info_location
22from ._dists import Distribution
23
24logger = logging.getLogger(__name__)
25
26# Used to avoid emitting duplicate invalid distribution metadata warnings for
27# the same dist-info directory.
28_warned_bad_metadata: set[BasePath | None] = set()
29
30
31def _looks_like_wheel(location: str) -> bool:
32 if not location.endswith(WHEEL_EXTENSION):
33 return False
34 if not os.path.isfile(location):
35 return False
36 try:
37 parse_wheel_filename(os.path.basename(location))
38 except InvalidWheelFilename:
39 return False
40 return zipfile.is_zipfile(location)
41
42
43class _DistributionFinder:
44 """Finder to locate distributions.
45
46 The main purpose of this class is to memoize found distributions' names, so
47 only one distribution is returned for each package name. At lot of pip code
48 assumes this (because it is setuptools's behavior), and not doing the same
49 can potentially cause a distribution in lower precedence path to override a
50 higher precedence one if the caller is not careful.
51
52 Eventually we probably want to make it possible to see lower precedence
53 installations as well. It's useful feature, after all.
54 """
55
56 FoundResult = tuple[importlib.metadata.Distribution, BasePath | None]
57
58 def __init__(self) -> None:
59 self._found_names: set[NormalizedName] = set()
60
61 def _find_impl(self, location: str) -> Iterator[FoundResult]:
62 """Find distributions in a location."""
63 # Skip looking inside a wheel. Since a package inside a wheel is not
64 # always valid (due to .data directories etc.), its .dist-info entry
65 # should not be considered an installed distribution.
66 if _looks_like_wheel(location):
67 return
68 # To know exactly where we find a distribution, we have to feed in the
69 # paths one by one, instead of dumping the list to importlib.metadata.
70 for dist in importlib.metadata.distributions(path=[location]):
71 info_location = get_info_location(dist)
72 try:
73 name = get_dist_canonical_name(dist)
74 except BadMetadata as e:
75 if info_location not in _warned_bad_metadata:
76 logger.warning("Skipping %s due to %s", info_location, e.reason)
77 _warned_bad_metadata.add(info_location)
78 continue
79 if name in self._found_names:
80 continue
81 self._found_names.add(name)
82 yield dist, info_location
83
84 def find(self, location: str) -> Iterator[BaseDistribution]:
85 """Find distributions in a location.
86
87 The path can be either a directory, or a ZIP archive.
88 """
89 for dist, info_location in self._find_impl(location):
90 if info_location is None:
91 installed_location: BasePath | None = None
92 else:
93 installed_location = info_location.parent
94 yield Distribution(dist, info_location, installed_location)
95
96 def find_legacy_editables(self, location: str) -> Iterator[BaseDistribution]:
97 """Read location in egg-link files and return distributions in there.
98
99 The path should be a directory; otherwise this returns nothing. This
100 follows how setuptools does this for compatibility. The first non-empty
101 line in the egg-link is read as a path (resolved against the egg-link's
102 containing directory if relative). Distributions found at that linked
103 location are returned.
104 """
105 path = pathlib.Path(location)
106 if not path.is_dir():
107 return
108 for child in path.iterdir():
109 if child.suffix != ".egg-link":
110 continue
111 with child.open() as f:
112 lines = (line.strip() for line in f)
113 target_rel = next((line for line in lines if line), "")
114 if not target_rel:
115 continue
116 target_location = str(path.joinpath(target_rel))
117 for dist, info_location in self._find_impl(target_location):
118 yield Distribution(dist, info_location, path)
119
120
121class Environment(BaseEnvironment):
122 def __init__(self, paths: Sequence[str]) -> None:
123 self._paths = paths
124
125 @classmethod
126 def default(cls) -> BaseEnvironment:
127 return cls(sys.path)
128
129 @classmethod
130 def from_paths(cls, paths: list[str] | None) -> BaseEnvironment:
131 if paths is None:
132 return cls(sys.path)
133 return cls(paths)
134
135 def _iter_distributions(self) -> Iterator[BaseDistribution]:
136 finder = _DistributionFinder()
137 for location in self._paths:
138 yield from finder.find(location)
139 yield from finder.find_legacy_editables(location)
140
141 def get_distribution(self, name: str) -> BaseDistribution | None:
142 canonical_name = canonicalize_name(name)
143 matches = (
144 distribution
145 for distribution in self.iter_all_distributions()
146 if distribution.canonical_name == canonical_name
147 )
148 return next(matches, None)