mirror of https://github.com/n00mkrad/flowframes
147 lines
5.0 KiB
Python
147 lines
5.0 KiB
Python
import email.message
|
|
import logging
|
|
from typing import 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 NormalizedName, 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,
|
|
Wheel,
|
|
)
|
|
|
|
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, wheel: Wheel, name: str) -> "Distribution":
|
|
with wheel.as_zipfile() as zf:
|
|
dist = pkg_resources_distribution_for_wheel(zf, name, wheel.location)
|
|
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 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)
|
|
|
|
def iter_provided_extras(self) -> Iterable[str]:
|
|
return self._dist.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)
|