-
-
Notifications
You must be signed in to change notification settings - Fork 34.3k
Expand file tree
/
Copy path__init__.py
More file actions
1206 lines (968 loc) · 37.3 KB
/
__init__.py
File metadata and controls
1206 lines (968 loc) · 37.3 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
"""
APIs exposing metadata from third-party Python packages.
This codebase is shared between importlib.metadata in the stdlib
and importlib_metadata in PyPI. See
https://github.com/python/importlib_metadata/wiki/Development-Methodology
for more detail.
"""
from __future__ import annotations
import abc
import collections
import email
import functools
import itertools
import operator
import os
import pathlib
import posixpath
import re
import sys
import textwrap
import types
from collections.abc import Iterable, Mapping
from contextlib import suppress
from importlib import import_module
from importlib.abc import MetaPathFinder
from itertools import starmap
from typing import Any
from . import _meta
from ._collections import FreezableDefaultDict, Pair
from ._context import ExceptionTrap
from ._functools import method_cache, noop, pass_none, passthrough
from ._itertools import always_iterable, bucket, unique_everseen
from ._meta import PackageMetadata, SimplePath
from ._typing import md_none
__all__ = [
'Distribution',
'DistributionFinder',
'PackageMetadata',
'PackageNotFoundError',
'PackagePath',
'MetadataNotFound',
'SimplePath',
'distribution',
'distributions',
'entry_points',
'files',
'metadata',
'packages_distributions',
'requires',
'version',
]
class PackageNotFoundError(ModuleNotFoundError):
"""The package was not found."""
def __str__(self) -> str:
return f"No package metadata was found for {self.name}"
@property
def name(self) -> str: # type: ignore[override] # make readonly
(name,) = self.args
return name
class MetadataNotFound(FileNotFoundError):
"""No metadata file is present in the distribution."""
class Sectioned:
"""
A simple entry point config parser for performance
>>> for item in Sectioned.read(Sectioned._sample):
... print(item)
Pair(name='sec1', value='# comments ignored')
Pair(name='sec1', value='a = 1')
Pair(name='sec1', value='b = 2')
Pair(name='sec2', value='a = 2')
>>> res = Sectioned.section_pairs(Sectioned._sample)
>>> item = next(res)
>>> item.name
'sec1'
>>> item.value
Pair(name='a', value='1')
>>> item = next(res)
>>> item.value
Pair(name='b', value='2')
>>> item = next(res)
>>> item.name
'sec2'
>>> item.value
Pair(name='a', value='2')
>>> list(res)
[]
"""
_sample = textwrap.dedent(
"""
[sec1]
# comments ignored
a = 1
b = 2
[sec2]
a = 2
"""
).lstrip()
@classmethod
def section_pairs(cls, text):
return (
section._replace(value=Pair.parse(section.value))
for section in cls.read(text, filter_=cls.valid)
if section.name is not None
)
@staticmethod
def read(text, filter_=None):
lines = filter(filter_, map(str.strip, text.splitlines()))
name = None
for value in lines:
section_match = value.startswith('[') and value.endswith(']')
if section_match:
name = value.strip('[]')
continue
yield Pair(name, value)
@staticmethod
def valid(line: str):
return line and not line.startswith('#')
class _EntryPointMatch(types.SimpleNamespace):
module: str
attr: str
extras: str
class EntryPoint:
"""An entry point as defined by Python packaging conventions.
See `the packaging docs on entry points
<https://packaging.python.org/specifications/entry-points/>`_
for more information.
>>> ep = EntryPoint(
... name=None, group=None, value='package.module:attr [extra1, extra2]')
>>> ep.module
'package.module'
>>> ep.attr
'attr'
>>> ep.extras
['extra1', 'extra2']
If the value package or module are not valid identifiers, a
ValueError is raised on access.
>>> EntryPoint(name=None, group=None, value='invalid-name').module
Traceback (most recent call last):
...
ValueError: ('Invalid object reference...invalid-name...
>>> EntryPoint(name=None, group=None, value='invalid-name').attr
Traceback (most recent call last):
...
ValueError: ('Invalid object reference...invalid-name...
>>> EntryPoint(name=None, group=None, value='invalid-name').extras
Traceback (most recent call last):
...
ValueError: ('Invalid object reference...invalid-name...
The same thing happens on construction.
>>> EntryPoint(name=None, group=None, value='invalid-name')
Traceback (most recent call last):
...
ValueError: ('Invalid object reference...invalid-name...
"""
pattern = re.compile(
r'(?P<module>[\w.]+)\s*'
r'(:\s*(?P<attr>[\w.]+)\s*)?'
r'((?P<extras>\[.*\])\s*)?$'
)
"""
A regular expression describing the syntax for an entry point,
which might look like:
- module
- package.module
- package.module:attribute
- package.module:object.attribute
- package.module:attr [extra1, extra2]
Other combinations are possible as well.
The expression is lenient about whitespace around the ':',
following the attr, and following any extras.
"""
name: str
value: str
group: str
dist: Distribution | None = None
def __init__(self, name: str, value: str, group: str) -> None:
vars(self).update(name=name, value=value, group=group)
self.module
def load(self) -> Any:
"""Load the entry point from its definition. If only a module
is indicated by the value, return that module. Otherwise,
return the named object.
"""
module = import_module(self.module)
attrs = filter(None, (self.attr or '').split('.'))
return functools.reduce(getattr, attrs, module)
@property
def module(self) -> str:
return self._match.module
@property
def attr(self) -> str:
return self._match.attr
@property
def extras(self) -> list[str]:
return re.findall(r'\w+', self._match.extras or '')
@functools.cached_property
def _match(self) -> _EntryPointMatch:
match = self.pattern.match(self.value)
if not match:
raise ValueError(
'Invalid object reference. '
'See https://packaging.python.org'
'/en/latest/specifications/entry-points/#data-model',
self.value,
)
return _EntryPointMatch(**match.groupdict())
def _for(self, dist):
vars(self).update(dist=dist)
return self
def matches(self, **params):
"""
EntryPoint matches the given parameters.
>>> ep = EntryPoint(group='foo', name='bar', value='bing:bong [extra1, extra2]')
>>> ep.matches(group='foo')
True
>>> ep.matches(name='bar', value='bing:bong [extra1, extra2]')
True
>>> ep.matches(group='foo', name='other')
False
>>> ep.matches()
True
>>> ep.matches(extras=['extra1', 'extra2'])
True
>>> ep.matches(module='bing')
True
>>> ep.matches(attr='bong')
True
"""
self._disallow_dist(params)
attrs = (getattr(self, param) for param in params)
return all(map(operator.eq, params.values(), attrs))
@staticmethod
def _disallow_dist(params):
"""
Querying by dist is not allowed (dist objects are not comparable).
>>> EntryPoint(name='fan', value='fav', group='fag').matches(dist='foo')
Traceback (most recent call last):
...
ValueError: "dist" is not suitable for matching...
"""
if "dist" in params:
raise ValueError(
'"dist" is not suitable for matching. '
"Instead, use Distribution.entry_points.select() on a "
"located distribution."
)
def _key(self):
return self.name, self.value, self.group
def __lt__(self, other):
return self._key() < other._key()
def __eq__(self, other):
return self._key() == other._key()
def __setattr__(self, name, value):
raise AttributeError("EntryPoint objects are immutable.")
def __repr__(self):
return (
f'EntryPoint(name={self.name!r}, value={self.value!r}, '
f'group={self.group!r})'
)
def __hash__(self) -> int:
return hash(self._key())
class EntryPoints(tuple):
"""
An immutable collection of selectable EntryPoint objects.
"""
__slots__ = ()
def __getitem__(self, name: str) -> EntryPoint: # type: ignore[override] # Work with str instead of int
"""
Get the EntryPoint in self matching name.
"""
try:
return next(iter(self.select(name=name)))
except StopIteration:
raise KeyError(name)
def __repr__(self):
"""
Repr with classname and tuple constructor to
signal that we deviate from regular tuple behavior.
"""
return '%s(%r)' % (self.__class__.__name__, tuple(self))
def select(self, **params) -> EntryPoints:
"""
Select entry points from self that match the
given parameters (typically group and/or name).
"""
return EntryPoints(ep for ep in self if ep.matches(**params))
@property
def names(self) -> set[str]:
"""
Return the set of all names of all entry points.
"""
return {ep.name for ep in self}
@property
def groups(self) -> set[str]:
"""
Return the set of all groups of all entry points.
"""
return {ep.group for ep in self}
@classmethod
def _from_text_for(cls, text, dist):
return cls(ep._for(dist) for ep in cls._from_text(text))
@staticmethod
def _from_text(text):
return (
EntryPoint(name=item.value.name, value=item.value.value, group=item.name)
for item in Sectioned.section_pairs(text or '')
)
class PackagePath(pathlib.PurePosixPath):
"""A reference to a path in a package"""
hash: FileHash | None
size: int
dist: Distribution
def read_text(self, encoding: str = 'utf-8') -> str:
return self.locate().read_text(encoding=encoding)
def read_binary(self) -> bytes:
return self.locate().read_bytes()
def locate(self) -> SimplePath:
"""Return a path-like object for this path"""
return self.dist.locate_file(self)
class FileHash:
def __init__(self, spec: str) -> None:
self.mode, _, self.value = spec.partition('=')
def __repr__(self) -> str:
return f'<FileHash mode: {self.mode} value: {self.value}>'
class Distribution(metaclass=abc.ABCMeta):
"""
An abstract Python distribution package.
Custom providers may derive from this class and define
the abstract methods to provide a concrete implementation
for their environment. Some providers may opt to override
the default implementation of some properties to bypass
the file-reading mechanism.
"""
@abc.abstractmethod
def read_text(self, filename) -> str | None:
"""Attempt to load metadata file given by the name.
Python distribution metadata is organized by blobs of text
typically represented as "files" in the metadata directory
(e.g. package-1.0.dist-info). These files include things
like:
- METADATA: The distribution metadata including fields
like Name and Version and Description.
- entry_points.txt: A series of entry points as defined in
`the entry points spec <https://packaging.python.org/en/latest/specifications/entry-points/#file-format>`_.
- RECORD: A record of files according to
`this recording spec <https://packaging.python.org/en/latest/specifications/recording-installed-packages/#the-record-file>`_.
A package may provide any set of files, including those
not listed here or none at all.
:param filename: The name of the file in the distribution info.
:return: The text if found, otherwise None.
"""
@abc.abstractmethod
def locate_file(self, path: str | os.PathLike[str]) -> SimplePath:
"""
Given a path to a file in this distribution, return a SimplePath
to it.
This method is used by callers of ``Distribution.files()`` to
locate files within the distribution. If it's possible for a
Distribution to represent files in the distribution as
``SimplePath`` objects, it should implement this method
to resolve such objects.
Some Distribution providers may elect not to resolve SimplePath
objects within the distribution by raising a
NotImplementedError, but consumers of such a Distribution would
be unable to invoke ``Distribution.files()``.
"""
@classmethod
def from_name(cls, name: str) -> Distribution:
"""Return the Distribution for the given package name.
:param name: The name of the distribution package to search for.
:return: The Distribution instance (or subclass thereof) for the named
package, if found.
:raises PackageNotFoundError: When the named package's distribution
metadata cannot be found.
:raises ValueError: When an invalid value is supplied for name.
"""
if not name:
raise ValueError("A distribution name is required.")
try:
return next(iter(cls._prefer_valid(cls.discover(name=name))))
except StopIteration:
raise PackageNotFoundError(name) from None
@classmethod
def discover(
cls, *, context: DistributionFinder.Context | None = None, **kwargs
) -> Iterable[Distribution]:
"""Return an iterable of Distribution objects for all packages.
Pass a ``context`` or pass keyword arguments for constructing
a context.
:context: A ``DistributionFinder.Context`` object.
:return: Iterable of Distribution objects for packages matching
the context.
"""
if context and kwargs:
raise ValueError("cannot accept context and kwargs")
context = context or DistributionFinder.Context(**kwargs)
return itertools.chain.from_iterable(
resolver(context) for resolver in cls._discover_resolvers()
)
@staticmethod
def _prefer_valid(dists: Iterable[Distribution]) -> Iterable[Distribution]:
"""
Prefer (move to the front) distributions that have metadata.
Ref python/importlib_resources#489.
"""
has_metadata = ExceptionTrap(MetadataNotFound).passes(
operator.attrgetter('metadata')
)
buckets = bucket(dists, has_metadata)
return itertools.chain(buckets[True], buckets[False])
@staticmethod
def at(path: str | os.PathLike[str]) -> Distribution:
"""Return a Distribution for the indicated metadata path.
:param path: a string or path-like object
:return: a concrete Distribution instance for the path
"""
return PathDistribution(pathlib.Path(path))
@staticmethod
def _discover_resolvers():
"""Search the meta_path for resolvers (MetadataPathFinders)."""
declared = (
getattr(finder, 'find_distributions', None) for finder in sys.meta_path
)
return filter(None, declared)
@property
def metadata(self) -> _meta.PackageMetadata:
"""Return the parsed metadata for this Distribution.
The returned object will have keys that name the various bits of
metadata per the
`Core metadata specifications <https://packaging.python.org/en/latest/specifications/core-metadata/#core-metadata>`_.
Custom providers may provide the METADATA file or override this
property.
:raises MetadataNotFound: If no metadata file is present.
"""
text = (
self.read_text('METADATA')
or self.read_text('PKG-INFO')
# This last clause is here to support old egg-info files. Its
# effect is to just end up using the PathDistribution's self._path
# (which points to the egg-info file) attribute unchanged.
or self.read_text('')
)
return self._assemble_message(self._ensure_metadata_present(text))
@staticmethod
def _assemble_message(text: str) -> _meta.PackageMetadata:
# deferred for performance (python/cpython#109829)
from . import _adapters
return _adapters.Message(email.message_from_string(text))
def _ensure_metadata_present(self, text: str | None) -> str:
if text is not None:
return text
raise MetadataNotFound('No package metadata was found.')
@property
def name(self) -> str:
"""Return the 'Name' metadata for the distribution package."""
return self.metadata['Name']
@property
def _normalized_name(self):
"""Return a normalized version of the name."""
return Prepared.normalize(self.name)
@property
def version(self) -> str:
"""Return the 'Version' metadata for the distribution package."""
return self.metadata['Version']
@property
def entry_points(self) -> EntryPoints:
"""
Return EntryPoints for this distribution.
Custom providers may provide the ``entry_points.txt`` file
or override this property.
"""
return EntryPoints._from_text_for(self.read_text('entry_points.txt'), self)
@property
def files(self) -> list[PackagePath] | None:
"""Files in this distribution.
:return: List of PackagePath for this distribution or None
Result is `None` if the metadata file that enumerates files
(i.e. RECORD for dist-info, or installed-files.txt or
SOURCES.txt for egg-info) is missing.
Result may be empty if the metadata exists but is empty.
Custom providers are recommended to provide a "RECORD" file (in
``read_text``) or override this property to allow for callers to be
able to resolve filenames provided by the package.
"""
def make_file(name, hash=None, size_str=None):
result = PackagePath(name)
result.hash = FileHash(hash) if hash else None
result.size = int(size_str) if size_str else None
result.dist = self
return result
@pass_none
def make_files(lines):
# Delay csv import, since Distribution.files is not as widely used
# as other parts of importlib.metadata
import csv
return starmap(make_file, csv.reader(lines))
@pass_none
def skip_missing_files(package_paths):
return list(filter(lambda path: path.locate().exists(), package_paths))
return skip_missing_files(
make_files(
self._read_files_distinfo()
or self._read_files_egginfo_installed()
or self._read_files_egginfo_sources()
)
)
def _read_files_distinfo(self):
"""
Read the lines of RECORD.
"""
text = self.read_text('RECORD')
return text and text.splitlines()
def _read_files_egginfo_installed(self):
"""
Read installed-files.txt and return lines in a similar
CSV-parsable format as RECORD: each file must be placed
relative to the site-packages directory and must also be
quoted (since file names can contain literal commas).
This file is written when the package is installed by pip,
but it might not be written for other installation methods.
Assume the file is accurate if it exists.
"""
text = self.read_text('installed-files.txt')
# Prepend the .egg-info/ subdir to the lines in this file.
# But this subdir is only available from PathDistribution's
# self._path.
subdir = getattr(self, '_path', None)
if not text or not subdir:
return
paths = (
(subdir / name)
.resolve()
.relative_to(self.locate_file('').resolve(), walk_up=True)
.as_posix()
for name in text.splitlines()
)
return map('"{}"'.format, paths)
def _read_files_egginfo_sources(self):
"""
Read SOURCES.txt and return lines in a similar CSV-parsable
format as RECORD: each file name must be quoted (since it
might contain literal commas).
Note that SOURCES.txt is not a reliable source for what
files are installed by a package. This file is generated
for a source archive, and the files that are present
there (e.g. setup.py) may not correctly reflect the files
that are present after the package has been installed.
"""
text = self.read_text('SOURCES.txt')
return text and map('"{}"'.format, text.splitlines())
@property
def requires(self) -> list[str] | None:
"""Generated requirements specified for this Distribution"""
reqs = self._read_dist_info_reqs() or self._read_egg_info_reqs()
return reqs and list(reqs)
def _read_dist_info_reqs(self):
return self.metadata.get_all('Requires-Dist')
def _read_egg_info_reqs(self):
source = self.read_text('requires.txt')
return pass_none(self._deps_from_requires_text)(source)
@classmethod
def _deps_from_requires_text(cls, source):
return cls._convert_egg_info_reqs_to_simple_reqs(Sectioned.read(source))
@staticmethod
def _convert_egg_info_reqs_to_simple_reqs(sections):
"""
Historically, setuptools would solicit and store 'extra'
requirements, including those with environment markers,
in separate sections. More modern tools expect each
dependency to be defined separately, with any relevant
extras and environment markers attached directly to that
requirement. This method converts the former to the
latter. See _test_deps_from_requires_text for an example.
"""
def make_condition(name):
return name and f'extra == "{name}"'
def quoted_marker(section):
section = section or ''
extra, sep, markers = section.partition(':')
if extra and markers:
markers = f'({markers})'
conditions = list(filter(None, [markers, make_condition(extra)]))
return '; ' + ' and '.join(conditions) if conditions else ''
def url_req_space(req):
"""
PEP 508 requires a space between the url_spec and the quoted_marker.
Ref python/importlib_metadata#357.
"""
# '@' is uniquely indicative of a url_req.
return ' ' * ('@' in req)
for section in sections:
space = url_req_space(section.value)
yield section.value + space + quoted_marker(section.name)
@property
def origin(self):
return self._load_json('direct_url.json')
def _load_json(self, filename):
# Deferred for performance (python/importlib_metadata#503)
import json
return pass_none(json.loads)(
self.read_text(filename),
object_hook=lambda data: types.SimpleNamespace(**data),
)
class DistributionFinder(MetaPathFinder):
"""
A MetaPathFinder capable of discovering installed distributions.
Custom providers should implement this interface in order to
supply metadata.
"""
class Context:
"""
Keyword arguments presented by the caller to
``distributions()`` or ``Distribution.discover()``
to narrow the scope of a search for distributions
in all DistributionFinders.
Each DistributionFinder may expect any parameters
and should attempt to honor the canonical
parameters defined below when appropriate.
This mechanism gives a custom provider a means to
solicit additional details from the caller beyond
"name" and "path" when searching distributions.
For example, imagine a provider that exposes suites
of packages in either a "public" or "private" ``realm``.
A caller may wish to query only for distributions in
a particular realm and could call
``distributions(realm="private")`` to signal to the
custom provider to only include distributions from that
realm.
"""
name = None
"""
Specific name for which a distribution finder should match.
A name of ``None`` matches all distributions.
"""
def __init__(self, **kwargs):
vars(self).update(kwargs)
@property
def path(self) -> list[str]:
"""
The sequence of directory path that a distribution finder
should search.
Typically refers to Python installed package paths such as
"site-packages" directories and defaults to ``sys.path``.
"""
return vars(self).get('path', sys.path)
@abc.abstractmethod
def find_distributions(self, context=Context()) -> Iterable[Distribution]:
"""
Find distributions.
Return an iterable of all Distribution instances capable of
loading the metadata for packages matching the ``context``,
a DistributionFinder.Context instance.
"""
@passthrough
def _clear_after_fork(cached):
"""Ensure ``func`` clears cached state after ``fork`` when supported.
``FastPath`` caches zip-backed ``pathlib.Path`` objects that retain a
reference to the parent's open ``ZipFile`` handle. Re-using a cached
instance in a forked child can therefore resurrect invalid file pointers
and trigger ``BadZipFile``/``OSError`` failures (python/importlib_metadata#520).
Registering ``cache_clear`` with ``os.register_at_fork`` keeps each process
on its own cache.
"""
getattr(os, 'register_at_fork', noop)(after_in_child=cached.cache_clear)
class FastPath:
"""
Micro-optimized class for searching a root for children.
Root is a path on the file system that may contain metadata
directories either as natural directories or within a zip file.
>>> FastPath('').children()
['...']
FastPath objects are cached and recycled for any given root.
>>> FastPath('foobar') is FastPath('foobar')
True
"""
@_clear_after_fork # type: ignore[misc]
@functools.lru_cache()
def __new__(cls, root):
return super().__new__(cls)
def __init__(self, root):
self.root = root
def joinpath(self, child):
return pathlib.Path(self.root, child)
def children(self):
with suppress(Exception):
return os.listdir(self.root or '.')
with suppress(Exception):
return self.zip_children()
return []
def zip_children(self):
# deferred for performance (python/importlib_metadata#502)
import zipfile
zip_path = zipfile.Path(self.root)
names = zip_path.root.namelist()
self.joinpath = zip_path.joinpath
return dict.fromkeys(child.split(posixpath.sep, 1)[0] for child in names)
def search(self, name):
return self.lookup(self.mtime).search(name)
@property
def mtime(self):
with suppress(OSError):
return os.stat(self.root).st_mtime
self.lookup.cache_clear()
@method_cache
def lookup(self, mtime):
return Lookup(self)
class Lookup:
"""
A micro-optimized class for searching a (fast) path for metadata.
"""
def __init__(self, path: FastPath):
"""
Calculate all of the children representing metadata.
From the children in the path, calculate early all of the
children that appear to represent metadata (infos) or legacy
metadata (eggs).
"""
base = os.path.basename(path.root).lower()
base_is_egg = base.endswith(".egg")
self.infos = FreezableDefaultDict(list)
self.eggs = FreezableDefaultDict(list)
for child in path.children():
low = child.lower()
if low.endswith((".dist-info", ".egg-info")):
# rpartition is faster than splitext and suitable for this purpose.
name = low.rpartition(".")[0].partition("-")[0]
normalized = Prepared.normalize(name)
self.infos[normalized].append(path.joinpath(child))
elif base_is_egg and low == "egg-info":
name = base.rpartition(".")[0].partition("-")[0]
legacy_normalized = Prepared.legacy_normalize(name)
self.eggs[legacy_normalized].append(path.joinpath(child))
self.infos.freeze()
self.eggs.freeze()
def search(self, prepared: Prepared):
"""
Yield all infos and eggs matching the Prepared query.
"""
infos = (
self.infos[prepared.normalized]
if prepared
else itertools.chain.from_iterable(self.infos.values())
)
eggs = (
self.eggs[prepared.legacy_normalized]
if prepared
else itertools.chain.from_iterable(self.eggs.values())
)
return itertools.chain(infos, eggs)
class Prepared:
"""
A prepared search query for metadata on a possibly-named package.
Pre-calculates the normalization to prevent repeated operations.
>>> none = Prepared(None)
>>> none.normalized
>>> none.legacy_normalized
>>> bool(none)
False
>>> sample = Prepared('Sample__Pkg-name.foo')
>>> sample.normalized
'sample_pkg_name_foo'
>>> sample.legacy_normalized
'sample__pkg_name.foo'
>>> bool(sample)
True
"""
normalized = None
legacy_normalized = None
def __init__(self, name: str | None):
self.name = name
if name is None:
return
self.normalized = self.normalize(name)
self.legacy_normalized = self.legacy_normalize(name)
@staticmethod
def normalize(name):
"""
PEP 503 normalization plus dashes as underscores.
Specifically avoids ``re.sub`` as prescribed for performance
benefits (see python/cpython#143658).
"""
value = name.lower().replace("-", "_").replace(".", "_")
# Condense repeats
while "__" in value:
value = value.replace("__", "_")
return value
@staticmethod
def legacy_normalize(name):
"""
Normalize the package name as found in the convention in
older packaging tools versions and specs.
"""
return name.lower().replace('-', '_')
def __bool__(self):
return bool(self.name)
class MetadataPathFinder(DistributionFinder):
@classmethod
def find_distributions(
cls, context=DistributionFinder.Context()
) -> Iterable[PathDistribution]:
"""
Find distributions.
Return an iterable of all Distribution instances capable of
loading the metadata for packages matching ``context.name``
(or all names if ``None`` indicated) along the paths in the list
of directories ``context.path``.
"""
found = cls._search_paths(context.name, context.path)
return map(PathDistribution, found)
@classmethod
def _search_paths(cls, name, paths):