This commit is contained in:
lz_db
2026-10-16 22:48:30 +08:00
parent 190de72571
commit 0b36fd3383
3161 changed files with 12787 additions and 77274 deletions
@@ -17,55 +17,65 @@ functions among several configuration file formats.
**PRIVATE MODULE**: API reserved for setuptools internal usage only.
"""
from __future__ import annotations
import ast
import importlib
import io
import os
import pathlib
import sys
from collections.abc import Iterable, Iterator, Mapping
from configparser import ConfigParser
import warnings
from glob import iglob
from importlib.machinery import ModuleSpec, all_suffixes
from configparser import ConfigParser
from importlib.machinery import ModuleSpec
from itertools import chain
from typing import (
TYPE_CHECKING,
Callable,
Dict,
Iterable,
Iterator,
List,
Mapping,
Optional,
Tuple,
TypeVar,
Union,
cast
)
from pathlib import Path
from types import ModuleType, TracebackType
from typing import TYPE_CHECKING, Any, Callable, TypeVar
from .. import _static
from .._path import StrPath, same_path as _same_path
from ..discovery import find_package_path
from ..warnings import SetuptoolsWarning
from types import ModuleType
from distutils.errors import DistutilsOptionError
from .._path import same_path as _same_path
if TYPE_CHECKING:
from typing_extensions import Self
from setuptools.dist import Distribution
from setuptools.dist import Distribution # noqa
from setuptools.discovery import ConfigDiscovery # noqa
from distutils.dist import DistributionMetadata # noqa
chain_iter = chain.from_iterable
_Path = Union[str, os.PathLike]
_K = TypeVar("_K")
_V_co = TypeVar("_V_co", covariant=True)
_V = TypeVar("_V", covariant=True)
class StaticModule:
"""Proxy to a module object that avoids executing arbitrary code."""
def __init__(self, name: str, spec: ModuleSpec) -> None:
module = ast.parse(pathlib.Path(spec.origin).read_bytes()) # type: ignore[arg-type] # Let it raise an error on None
def __init__(self, name: str, spec: ModuleSpec):
module = ast.parse(pathlib.Path(spec.origin).read_bytes())
vars(self).update(locals())
del self.self
def _find_assignments(self) -> Iterator[tuple[ast.AST, ast.AST]]:
def _find_assignments(self) -> Iterator[Tuple[ast.AST, ast.AST]]:
for statement in self.module.body:
if isinstance(statement, ast.Assign):
yield from ((target, statement.value) for target in statement.targets)
elif isinstance(statement, ast.AnnAssign) and statement.value:
yield (statement.target, statement.value)
def __getattr__(self, attr: str):
def __getattr__(self, attr):
"""Attempt to load an attribute "statically", via :func:`ast.literal_eval`."""
try:
return next(
@@ -78,8 +88,8 @@ class StaticModule:
def glob_relative(
patterns: Iterable[str], root_dir: StrPath | None = None
) -> list[str]:
patterns: Iterable[str], root_dir: Optional[_Path] = None
) -> List[str]:
"""Expand the list of glob patterns, but preserving relative paths.
:param list[str] patterns: List of glob patterns
@@ -91,16 +101,14 @@ def glob_relative(
expanded_values = []
root_dir = root_dir or os.getcwd()
for value in patterns:
# Has globby characters?
if any(char in value for char in glob_characters):
# then expand the glob pattern while keeping paths *relative*:
glob_path = os.path.abspath(os.path.join(root_dir, value))
expanded_values.extend(
sorted(
os.path.relpath(path, root_dir).replace(os.sep, "/")
for path in iglob(glob_path, recursive=True)
)
)
expanded_values.extend(sorted(
os.path.relpath(path, root_dir).replace(os.sep, "/")
for path in iglob(glob_path, recursive=True)))
else:
# take the value as-is
@@ -110,16 +118,14 @@ def glob_relative(
return expanded_values
def read_files(
filepaths: StrPath | Iterable[StrPath], root_dir: StrPath | None = None
) -> str:
def read_files(filepaths: Union[str, bytes, Iterable[_Path]], root_dir=None) -> str:
"""Return the content of the files concatenated using ``\n`` as str
This function is sandboxed and won't reach anything outside ``root_dir``
(By default ``root_dir`` is the current directory).
"""
from more_itertools import always_iterable
from setuptools.extern.more_itertools import always_iterable
root_dir = os.path.abspath(root_dir or os.getcwd())
_filepaths = (os.path.join(root_dir, path) for path in always_iterable(filepaths))
@@ -130,20 +136,20 @@ def read_files(
)
def _filter_existing_files(filepaths: Iterable[StrPath]) -> Iterator[StrPath]:
def _filter_existing_files(filepaths: Iterable[_Path]) -> Iterator[_Path]:
for path in filepaths:
if os.path.isfile(path):
yield path
else:
SetuptoolsWarning.emit(f"File {path!r} cannot be found")
warnings.warn(f"File {path!r} cannot be found")
def _read_file(filepath: bytes | StrPath) -> str:
with open(filepath, encoding='utf-8') as f:
def _read_file(filepath: Union[bytes, _Path]) -> str:
with io.open(filepath, encoding='utf-8') as f:
return f.read()
def _assert_local(filepath: StrPath, root_dir: str):
def _assert_local(filepath: _Path, root_dir: str):
if Path(os.path.abspath(root_dir)) not in Path(os.path.abspath(filepath)).parents:
msg = f"Cannot access {filepath!r} (or anything outside {root_dir!r})"
raise DistutilsOptionError(msg)
@@ -153,9 +159,9 @@ def _assert_local(filepath: StrPath, root_dir: str):
def read_attr(
attr_desc: str,
package_dir: Mapping[str, str] | None = None,
root_dir: StrPath | None = None,
) -> Any:
package_dir: Optional[Mapping[str, str]] = None,
root_dir: Optional[_Path] = None
):
"""Reads the value of an attribute from a module.
This function will try to read the attributed statically first
@@ -178,20 +184,18 @@ def read_attr(
attr_name = attrs_path.pop()
module_name = '.'.join(attrs_path)
module_name = module_name or '__init__'
path = _find_module(module_name, package_dir, root_dir)
_parent_path, path, module_name = _find_module(module_name, package_dir, root_dir)
spec = _find_spec(module_name, path)
try:
value = getattr(StaticModule(module_name, spec), attr_name)
# XXX: Is marking as static contents coming from modules too optimistic?
return _static.attempt_conversion(value)
return getattr(StaticModule(module_name, spec), attr_name)
except Exception:
# fallback to evaluate module
module = _load_spec(spec, module_name)
return getattr(module, attr_name)
def _find_spec(module_name: str, module_path: StrPath | None) -> ModuleSpec:
def _find_spec(module_name: str, module_path: Optional[_Path]) -> ModuleSpec:
spec = importlib.util.spec_from_file_location(module_name, module_path)
spec = spec or importlib.util.find_spec(module_name)
@@ -207,38 +211,48 @@ def _load_spec(spec: ModuleSpec, module_name: str) -> ModuleType:
return sys.modules[name]
module = importlib.util.module_from_spec(spec)
sys.modules[name] = module # cache (it also ensures `==` works on loaded items)
assert spec.loader is not None
spec.loader.exec_module(module)
spec.loader.exec_module(module) # type: ignore
return module
def _find_module(
module_name: str, package_dir: Mapping[str, str] | None, root_dir: StrPath
) -> str | None:
"""Find the path to the module named ``module_name``,
considering the ``package_dir`` in the build configuration and ``root_dir``.
>>> tmp = getfixture('tmpdir')
>>> _ = tmp.ensure("a/b/c.py")
>>> _ = tmp.ensure("a/b/d/__init__.py")
>>> r = lambda x: x.replace(str(tmp), "tmp").replace(os.sep, "/")
>>> r(_find_module("a.b.c", None, tmp))
'tmp/a/b/c.py'
>>> r(_find_module("f.g.h", {"": "1", "f": "2", "f.g": "3", "f.g.h": "a/b/d"}, tmp))
'tmp/a/b/d/__init__.py'
module_name: str, package_dir: Optional[Mapping[str, str]], root_dir: _Path
) -> Tuple[_Path, Optional[str], str]:
"""Given a module (that could normally be imported by ``module_name``
after the build is complete), find the path to the parent directory where
it is contained and the canonical name that could be used to import it
considering the ``package_dir`` in the build configuration and ``root_dir``
"""
path_start = find_package_path(module_name, package_dir or {}, root_dir)
candidates = chain.from_iterable(
(f"{path_start}{ext}", os.path.join(path_start, f"__init__{ext}"))
for ext in all_suffixes()
parent_path = root_dir
module_parts = module_name.split('.')
if package_dir:
if module_parts[0] in package_dir:
# A custom path was specified for the module we want to import
custom_path = package_dir[module_parts[0]]
parts = custom_path.rsplit('/', 1)
if len(parts) > 1:
parent_path = os.path.join(root_dir, parts[0])
parent_module = parts[1]
else:
parent_module = custom_path
module_name = ".".join([parent_module, *module_parts[1:]])
elif '' in package_dir:
# A custom parent directory was specified for all root modules
parent_path = os.path.join(root_dir, package_dir[''])
path_start = os.path.join(parent_path, *module_name.split("."))
candidates = chain(
(f"{path_start}.py", os.path.join(path_start, "__init__.py")),
iglob(f"{path_start}.*")
)
return next((x for x in candidates if os.path.isfile(x)), None)
module_path = next((x for x in candidates if os.path.isfile(x)), None)
return parent_path, module_path, module_name
def resolve_class(
qualified_class_name: str,
package_dir: Mapping[str, str] | None = None,
root_dir: StrPath | None = None,
package_dir: Optional[Mapping[str, str]] = None,
root_dir: Optional[_Path] = None
) -> Callable:
"""Given a qualified class name, return the associated class object"""
root_dir = root_dir or os.getcwd()
@@ -246,16 +260,16 @@ def resolve_class(
class_name = qualified_class_name[idx + 1 :]
pkg_name = qualified_class_name[:idx]
path = _find_module(pkg_name, package_dir, root_dir)
module = _load_spec(_find_spec(pkg_name, path), pkg_name)
_parent_path, path, module_name = _find_module(pkg_name, package_dir, root_dir)
module = _load_spec(_find_spec(module_name, path), module_name)
return getattr(module, class_name)
def cmdclass(
values: dict[str, str],
package_dir: Mapping[str, str] | None = None,
root_dir: StrPath | None = None,
) -> dict[str, Callable]:
values: Dict[str, str],
package_dir: Optional[Mapping[str, str]] = None,
root_dir: Optional[_Path] = None
) -> Dict[str, Callable]:
"""Given a dictionary mapping command names to strings for qualified class
names, apply :func:`resolve_class` to the dict values.
"""
@@ -265,10 +279,10 @@ def cmdclass(
def find_packages(
*,
namespaces=True,
fill_package_dir: dict[str, str] | None = None,
root_dir: StrPath | None = None,
**kwargs,
) -> list[str]:
fill_package_dir: Optional[Dict[str, str]] = None,
root_dir: Optional[_Path] = None,
**kwargs
) -> List[str]:
"""Works similarly to :func:`setuptools.find_packages`, but with all
arguments given as keyword arguments. Moreover, ``where`` can be given
as a list (the results will be simply concatenated).
@@ -286,19 +300,17 @@ def find_packages(
:rtype: list
"""
from more_itertools import always_iterable, unique_everseen
from setuptools.discovery import construct_package_dir
from setuptools.extern.more_itertools import unique_everseen, always_iterable
# check "not namespaces" first due to python/mypy#6232
if not namespaces:
from setuptools.discovery import PackageFinder
else:
if namespaces:
from setuptools.discovery import PEP420PackageFinder as PackageFinder
else:
from setuptools.discovery import PackageFinder # type: ignore
root_dir = root_dir or os.curdir
where = kwargs.pop('where', ['.'])
packages: list[str] = []
packages: List[str] = []
fill_package_dir = {} if fill_package_dir is None else fill_package_dir
search = list(unique_everseen(always_iterable(where)))
@@ -310,29 +322,35 @@ def find_packages(
pkgs = PackageFinder.find(package_path, **kwargs)
packages.extend(pkgs)
if pkgs and not (
fill_package_dir.get("") == path or os.path.samefile(package_path, root_dir)
fill_package_dir.get("") == path
or os.path.samefile(package_path, root_dir)
):
fill_package_dir.update(construct_package_dir(pkgs, path))
return packages
def _nest_path(parent: StrPath, path: StrPath) -> str:
def _nest_path(parent: _Path, path: _Path) -> str:
path = parent if path in {".", ""} else os.path.join(parent, path)
return os.path.normpath(path)
def version(value: Callable | Iterable[str | int] | str) -> str:
def version(value: Union[Callable, Iterable[Union[str, int]], str]) -> str:
"""When getting the version directly from an attribute,
it should be normalised to string.
"""
_value = value() if callable(value) else value
if callable(value):
value = value()
if isinstance(_value, str):
return _value
if hasattr(_value, '__iter__'):
return '.'.join(map(str, _value))
return f'{_value}'
value = cast(Iterable[Union[str, int]], value)
if not isinstance(value, str):
if hasattr(value, '__iter__'):
value = '.'.join(map(str, value))
else:
value = '%s' % value
return value
def canonic_package_data(package_data: dict) -> dict:
@@ -342,8 +360,8 @@ def canonic_package_data(package_data: dict) -> dict:
def canonic_data_files(
data_files: list | dict, root_dir: StrPath | None = None
) -> list[tuple[str, list[str]]]:
data_files: Union[list, dict], root_dir: Optional[_Path] = None
) -> List[Tuple[str, List[str]]]:
"""For compatibility with ``setup.py``, ``data_files`` should be a list
of pairs instead of a dict.
@@ -358,17 +376,14 @@ def canonic_data_files(
]
def entry_points(
text: str, text_source: str = "entry-points"
) -> dict[str, dict[str, str]]:
def entry_points(text: str, text_source="entry-points") -> Dict[str, dict]:
"""Given the contents of entry-points file,
process it into a 2-level dictionary (``dict[str, dict[str, str]]``).
The first level keys are entry-point groups, the second level keys are
entry-point names, and the second level values are references to objects
(that correspond to the entry-point value).
"""
# Using undocumented behaviour, see python/typeshed#12700
parser = ConfigParser(default_section=None, delimiters=("=",)) # type: ignore[call-overload]
parser = ConfigParser(default_section=None, delimiters=("=",)) # type: ignore
parser.optionxform = str # case sensitive
parser.read_string(text, text_source)
groups = {k: dict(v.items()) for k, v in parser.items()}
@@ -386,7 +401,7 @@ class EnsurePackagesDiscovered:
and those might not have been processed yet.
"""
def __init__(self, distribution: Distribution) -> None:
def __init__(self, distribution: "Distribution"):
self._dist = distribution
self._called = False
@@ -396,15 +411,10 @@ class EnsurePackagesDiscovered:
self._called = True
self._dist.set_defaults(name=False) # Skip name, we can still be parsing
def __enter__(self) -> Self:
def __enter__(self):
return self
def __exit__(
self,
exc_type: type[BaseException] | None,
exc_value: BaseException | None,
traceback: TracebackType | None,
):
def __exit__(self, _exc_type, _exc_value, _traceback):
if self._called:
self._dist.set_defaults.analyse_name() # Now we can set a default name
@@ -419,7 +429,7 @@ class EnsurePackagesDiscovered:
return LazyMappingProxy(self._get_package_dir)
class LazyMappingProxy(Mapping[_K, _V_co]):
class LazyMappingProxy(Mapping[_K, _V]):
"""Mapping proxy that delays resolving the target object, until really needed.
>>> def obtain_mapping():
@@ -433,16 +443,16 @@ class LazyMappingProxy(Mapping[_K, _V_co]):
'other value'
"""
def __init__(self, obtain_mapping_value: Callable[[], Mapping[_K, _V_co]]) -> None:
def __init__(self, obtain_mapping_value: Callable[[], Mapping[_K, _V]]):
self._obtain = obtain_mapping_value
self._value: Mapping[_K, _V_co] | None = None
self._value: Optional[Mapping[_K, _V]] = None
def _target(self) -> Mapping[_K, _V_co]:
def _target(self) -> Mapping[_K, _V]:
if self._value is None:
self._value = self._obtain()
return self._value
def __getitem__(self, key: _K) -> _V_co:
def __getitem__(self, key: _K) -> _V:
return self._target()[key]
def __len__(self) -> int: