Starter Code

This commit is contained in:
2025-01-29 23:39:42 -05:00
parent d32023c882
commit b4fde9997e
1113 changed files with 210062 additions and 0 deletions

View File

@@ -0,0 +1,48 @@
from typing import List, Optional
from .base import BaseDistribution, BaseEnvironment
__all__ = [
"BaseDistribution",
"BaseEnvironment",
"get_default_environment",
"get_environment",
"get_wheel_distribution",
]
def get_default_environment() -> BaseEnvironment:
"""Get the default representation for the current environment.
This returns an Environment instance from the chosen backend. The default
Environment instance should be built from ``sys.path`` and may use caching
to share instance state accorss calls.
"""
from .pkg_resources import Environment
return Environment.default()
def get_environment(paths: Optional[List[str]]) -> BaseEnvironment:
"""Get a representation of the environment specified by ``paths``.
This returns an Environment instance from the chosen backend based on the
given import paths. The backend must build a fresh instance representing
the state of installed distributions when this function is called.
"""
from .pkg_resources import Environment
return Environment.from_paths(paths)
def get_wheel_distribution(wheel_path: str, canonical_name: str) -> BaseDistribution:
"""Get the representation of the specified wheel's distribution metadata.
This returns a Distribution instance from the chosen backend based on
the given wheel's ``.dist-info`` directory.
:param canonical_name: Normalized project name of the given wheel.
"""
from .pkg_resources import Distribution
return Distribution.from_wheel(wheel_path, canonical_name)

View File

@@ -0,0 +1,242 @@
import email.message
import json
import logging
import re
from typing import (
TYPE_CHECKING,
Collection,
Container,
Iterable,
Iterator,
List,
Optional,
Union,
)
from pip._vendor.packaging.requirements import Requirement
from pip._vendor.packaging.version import LegacyVersion, Version
from pip._internal.models.direct_url import (
DIRECT_URL_METADATA_NAME,
DirectUrl,
DirectUrlValidationError,
)
from pip._internal.utils.misc import stdlib_pkgs # TODO: Move definition here.
if TYPE_CHECKING:
from typing import Protocol
from pip._vendor.packaging.utils import NormalizedName
else:
Protocol = object
DistributionVersion = Union[LegacyVersion, Version]
logger = logging.getLogger(__name__)
class BaseEntryPoint(Protocol):
@property
def name(self) -> str:
raise NotImplementedError()
@property
def value(self) -> str:
raise NotImplementedError()
@property
def group(self) -> str:
raise NotImplementedError()
class BaseDistribution(Protocol):
@property
def location(self) -> Optional[str]:
"""Where the distribution is loaded from.
A string value is not necessarily a filesystem path, since distributions
can be loaded from other sources, e.g. arbitrary zip archives. ``None``
means the distribution is created in-memory.
Do not canonicalize this value with e.g. ``pathlib.Path.resolve()``. If
this is a symbolic link, we want to preserve the relative path between
it and files in the distribution.
"""
raise NotImplementedError()
@property
def info_directory(self) -> Optional[str]:
"""Location of the .[egg|dist]-info directory.
Similarly to ``location``, a string value is not necessarily a
filesystem path. ``None`` means the distribution is created in-memory.
For a modern .dist-info installation on disk, this should be something
like ``{location}/{raw_name}-{version}.dist-info``.
Do not canonicalize this value with e.g. ``pathlib.Path.resolve()``. If
this is a symbolic link, we want to preserve the relative path between
it and other files in the distribution.
"""
raise NotImplementedError()
@property
def canonical_name(self) -> "NormalizedName":
raise NotImplementedError()
@property
def version(self) -> DistributionVersion:
raise NotImplementedError()
@property
def direct_url(self) -> Optional[DirectUrl]:
"""Obtain a DirectUrl from this distribution.
Returns None if the distribution has no `direct_url.json` metadata,
or if `direct_url.json` is invalid.
"""
try:
content = self.read_text(DIRECT_URL_METADATA_NAME)
except FileNotFoundError:
return None
try:
return DirectUrl.from_json(content)
except (
UnicodeDecodeError,
json.JSONDecodeError,
DirectUrlValidationError,
) as e:
logger.warning(
"Error parsing %s for %s: %s",
DIRECT_URL_METADATA_NAME,
self.canonical_name,
e,
)
return None
@property
def installer(self) -> str:
raise NotImplementedError()
@property
def editable(self) -> bool:
raise NotImplementedError()
@property
def local(self) -> bool:
raise NotImplementedError()
@property
def in_usersite(self) -> bool:
raise NotImplementedError()
@property
def in_site_packages(self) -> bool:
raise NotImplementedError()
def read_text(self, name: str) -> str:
"""Read a file in the .dist-info (or .egg-info) directory.
Should raise ``FileNotFoundError`` if ``name`` does not exist in the
metadata directory.
"""
raise NotImplementedError()
def iter_entry_points(self) -> Iterable[BaseEntryPoint]:
raise NotImplementedError()
@property
def metadata(self) -> email.message.Message:
"""Metadata of distribution parsed from e.g. METADATA or PKG-INFO."""
raise NotImplementedError()
@property
def metadata_version(self) -> Optional[str]:
"""Value of "Metadata-Version:" in distribution metadata, if available."""
return self.metadata.get("Metadata-Version")
@property
def raw_name(self) -> str:
"""Value of "Name:" in distribution metadata."""
# The metadata should NEVER be missing the Name: key, but if it somehow
# does not, fall back to the known canonical name.
return self.metadata.get("Name", self.canonical_name)
def iter_dependencies(self, extras: Collection[str] = ()) -> Iterable[Requirement]:
raise NotImplementedError()
class BaseEnvironment:
"""An environment containing distributions to introspect."""
@classmethod
def default(cls) -> "BaseEnvironment":
raise NotImplementedError()
@classmethod
def from_paths(cls, paths: Optional[List[str]]) -> "BaseEnvironment":
raise NotImplementedError()
def get_distribution(self, name: str) -> Optional["BaseDistribution"]:
"""Given a requirement name, return the installed distributions."""
raise NotImplementedError()
def _iter_distributions(self) -> Iterator["BaseDistribution"]:
"""Iterate through installed distributions.
This function should be implemented by subclass, but never called
directly. Use the public ``iter_distribution()`` instead, which
implements additional logic to make sure the distributions are valid.
"""
raise NotImplementedError()
def iter_distributions(self) -> Iterator["BaseDistribution"]:
"""Iterate through installed distributions."""
for dist in self._iter_distributions():
# Make sure the distribution actually comes from a valid Python
# packaging distribution. Pip's AdjacentTempDirectory leaves folders
# e.g. ``~atplotlib.dist-info`` if cleanup was interrupted. The
# valid project name pattern is taken from PEP 508.
project_name_valid = re.match(
r"^([A-Z0-9]|[A-Z0-9][A-Z0-9._-]*[A-Z0-9])$",
dist.canonical_name,
flags=re.IGNORECASE,
)
if not project_name_valid:
logger.warning(
"Ignoring invalid distribution %s (%s)",
dist.canonical_name,
dist.location,
)
continue
yield dist
def iter_installed_distributions(
self,
local_only: bool = True,
skip: Container[str] = stdlib_pkgs,
include_editables: bool = True,
editables_only: bool = False,
user_only: bool = False,
) -> Iterator[BaseDistribution]:
"""Return a list of installed distributions.
:param local_only: If True (default), only return installations
local to the current virtualenv, if in a virtualenv.
:param skip: An iterable of canonicalized project names to ignore;
defaults to ``stdlib_pkgs``.
:param include_editables: If False, don't report editables.
:param editables_only: If True, only report editables.
:param user_only: If True, only report installations in the user
site directory.
"""
it = self.iter_distributions()
if local_only:
it = (d for d in it if d.local)
if not include_editables:
it = (d for d in it if not d.editable)
if editables_only:
it = (d for d in it if d.editable)
if user_only:
it = (d for d in it if d.in_usersite)
return (d for d in it if d.canonical_name not in skip)

View File

@@ -0,0 +1,153 @@
import email.message
import logging
import zipfile
from typing import (
TYPE_CHECKING,
Collection,
Iterable,
Iterator,
List,
NamedTuple,
Optional,
)
from pip._vendor import pkg_resources
from pip._vendor.packaging.requirements import Requirement
from pip._vendor.packaging.utils import canonicalize_name
from pip._vendor.packaging.version import parse as parse_version
from pip._internal.utils import misc # TODO: Move definition here.
from pip._internal.utils.packaging import get_installer, get_metadata
from pip._internal.utils.wheel import pkg_resources_distribution_for_wheel
from .base import BaseDistribution, BaseEntryPoint, BaseEnvironment, DistributionVersion
if TYPE_CHECKING:
from pip._vendor.packaging.utils import NormalizedName
logger = logging.getLogger(__name__)
class EntryPoint(NamedTuple):
name: str
value: str
group: str
class Distribution(BaseDistribution):
def __init__(self, dist: pkg_resources.Distribution) -> None:
self._dist = dist
@classmethod
def from_wheel(cls, path: str, name: str) -> "Distribution":
with zipfile.ZipFile(path, allowZip64=True) as zf:
dist = pkg_resources_distribution_for_wheel(zf, name, path)
return cls(dist)
@property
def location(self) -> Optional[str]:
return self._dist.location
@property
def info_directory(self) -> Optional[str]:
return self._dist.egg_info
@property
def canonical_name(self) -> "NormalizedName":
return canonicalize_name(self._dist.project_name)
@property
def version(self) -> DistributionVersion:
return parse_version(self._dist.version)
@property
def installer(self) -> str:
return get_installer(self._dist)
@property
def editable(self) -> bool:
return misc.dist_is_editable(self._dist)
@property
def local(self) -> bool:
return misc.dist_is_local(self._dist)
@property
def in_usersite(self) -> bool:
return misc.dist_in_usersite(self._dist)
@property
def in_site_packages(self) -> bool:
return misc.dist_in_site_packages(self._dist)
def read_text(self, name: str) -> str:
if not self._dist.has_metadata(name):
raise FileNotFoundError(name)
return self._dist.get_metadata(name)
def iter_entry_points(self) -> Iterable[BaseEntryPoint]:
for group, entries in self._dist.get_entry_map().items():
for name, entry_point in entries.items():
name, _, value = str(entry_point).partition("=")
yield EntryPoint(name=name.strip(), value=value.strip(), group=group)
@property
def metadata(self) -> email.message.Message:
return get_metadata(self._dist)
def iter_dependencies(self, extras: Collection[str] = ()) -> Iterable[Requirement]:
if extras: # pkg_resources raises on invalid extras, so we sanitize.
extras = frozenset(extras).intersection(self._dist.extras)
return self._dist.requires(extras)
class Environment(BaseEnvironment):
def __init__(self, ws: pkg_resources.WorkingSet) -> None:
self._ws = ws
@classmethod
def default(cls) -> BaseEnvironment:
return cls(pkg_resources.working_set)
@classmethod
def from_paths(cls, paths: Optional[List[str]]) -> BaseEnvironment:
return cls(pkg_resources.WorkingSet(paths))
def _search_distribution(self, name: str) -> Optional[BaseDistribution]:
"""Find a distribution matching the ``name`` in the environment.
This searches from *all* distributions available in the environment, to
match the behavior of ``pkg_resources.get_distribution()``.
"""
canonical_name = canonicalize_name(name)
for dist in self.iter_distributions():
if dist.canonical_name == canonical_name:
return dist
return None
def get_distribution(self, name: str) -> Optional[BaseDistribution]:
# Search the distribution by looking through the working set.
dist = self._search_distribution(name)
if dist:
return dist
# If distribution could not be found, call working_set.require to
# update the working set, and try to find the distribution again.
# This might happen for e.g. when you install a package twice, once
# using setup.py develop and again using setup.py install. Now when
# running pip uninstall twice, the package gets removed from the
# working set in the first uninstall, so we have to populate the
# working set again so that pip knows about it and the packages gets
# picked up and is successfully uninstalled the second time too.
try:
# We didn't pass in any version specifiers, so this can never
# raise pkg_resources.VersionConflict.
self._ws.require(name)
except pkg_resources.DistributionNotFound:
return None
return self._search_distribution(name)
def _iter_distributions(self) -> Iterator[BaseDistribution]:
for dist in self._ws:
yield Distribution(dist)