Загрузить файлы в «venv/Lib/site-packages/pip/_vendor/packaking»
This commit is contained in:
388
venv/Lib/site-packages/pip/_vendor/packaking/markers.py
Normal file
388
venv/Lib/site-packages/pip/_vendor/packaking/markers.py
Normal file
@@ -0,0 +1,388 @@
|
||||
# This file is dual licensed under the terms of the Apache License, Version
|
||||
# 2.0, and the BSD License. See the LICENSE file in the root of this repository
|
||||
# for complete details.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import operator
|
||||
import os
|
||||
import platform
|
||||
import sys
|
||||
from typing import AbstractSet, Callable, Literal, Mapping, TypedDict, Union, cast
|
||||
|
||||
from ._parser import MarkerAtom, MarkerList, Op, Value, Variable
|
||||
from ._parser import parse_marker as _parse_marker
|
||||
from ._tokenizer import ParserSyntaxError
|
||||
from .specifiers import InvalidSpecifier, Specifier
|
||||
from .utils import canonicalize_name
|
||||
|
||||
__all__ = [
|
||||
"Environment",
|
||||
"EvaluateContext",
|
||||
"InvalidMarker",
|
||||
"Marker",
|
||||
"UndefinedComparison",
|
||||
"UndefinedEnvironmentName",
|
||||
"default_environment",
|
||||
]
|
||||
|
||||
Operator = Callable[[str, Union[str, AbstractSet[str]]], bool]
|
||||
EvaluateContext = Literal["metadata", "lock_file", "requirement"]
|
||||
MARKERS_ALLOWING_SET = {"extras", "dependency_groups"}
|
||||
MARKERS_REQUIRING_VERSION = {
|
||||
"implementation_version",
|
||||
"platform_release",
|
||||
"python_full_version",
|
||||
"python_version",
|
||||
}
|
||||
|
||||
|
||||
class InvalidMarker(ValueError):
|
||||
"""
|
||||
An invalid marker was found, users should refer to PEP 508.
|
||||
"""
|
||||
|
||||
|
||||
class UndefinedComparison(ValueError):
|
||||
"""
|
||||
An invalid operation was attempted on a value that doesn't support it.
|
||||
"""
|
||||
|
||||
|
||||
class UndefinedEnvironmentName(ValueError):
|
||||
"""
|
||||
A name was attempted to be used that does not exist inside of the
|
||||
environment.
|
||||
"""
|
||||
|
||||
|
||||
class Environment(TypedDict):
|
||||
implementation_name: str
|
||||
"""The implementation's identifier, e.g. ``'cpython'``."""
|
||||
|
||||
implementation_version: str
|
||||
"""
|
||||
The implementation's version, e.g. ``'3.13.0a2'`` for CPython 3.13.0a2, or
|
||||
``'7.3.13'`` for PyPy3.10 v7.3.13.
|
||||
"""
|
||||
|
||||
os_name: str
|
||||
"""
|
||||
The value of :py:data:`os.name`. The name of the operating system dependent module
|
||||
imported, e.g. ``'posix'``.
|
||||
"""
|
||||
|
||||
platform_machine: str
|
||||
"""
|
||||
Returns the machine type, e.g. ``'i386'``.
|
||||
|
||||
An empty string if the value cannot be determined.
|
||||
"""
|
||||
|
||||
platform_release: str
|
||||
"""
|
||||
The system's release, e.g. ``'2.2.0'`` or ``'NT'``.
|
||||
|
||||
An empty string if the value cannot be determined.
|
||||
"""
|
||||
|
||||
platform_system: str
|
||||
"""
|
||||
The system/OS name, e.g. ``'Linux'``, ``'Windows'`` or ``'Java'``.
|
||||
|
||||
An empty string if the value cannot be determined.
|
||||
"""
|
||||
|
||||
platform_version: str
|
||||
"""
|
||||
The system's release version, e.g. ``'#3 on degas'``.
|
||||
|
||||
An empty string if the value cannot be determined.
|
||||
"""
|
||||
|
||||
python_full_version: str
|
||||
"""
|
||||
The Python version as string ``'major.minor.patchlevel'``.
|
||||
|
||||
Note that unlike the Python :py:data:`sys.version`, this value will always include
|
||||
the patchlevel (it defaults to 0).
|
||||
"""
|
||||
|
||||
platform_python_implementation: str
|
||||
"""
|
||||
A string identifying the Python implementation, e.g. ``'CPython'``.
|
||||
"""
|
||||
|
||||
python_version: str
|
||||
"""The Python version as string ``'major.minor'``."""
|
||||
|
||||
sys_platform: str
|
||||
"""
|
||||
This string contains a platform identifier that can be used to append
|
||||
platform-specific components to :py:data:`sys.path`, for instance.
|
||||
|
||||
For Unix systems, except on Linux and AIX, this is the lowercased OS name as
|
||||
returned by ``uname -s`` with the first part of the version as returned by
|
||||
``uname -r`` appended, e.g. ``'sunos5'`` or ``'freebsd8'``, at the time when Python
|
||||
was built.
|
||||
"""
|
||||
|
||||
|
||||
def _normalize_extras(
|
||||
result: MarkerList | MarkerAtom | str,
|
||||
) -> MarkerList | MarkerAtom | str:
|
||||
if not isinstance(result, tuple):
|
||||
return result
|
||||
|
||||
lhs, op, rhs = result
|
||||
if isinstance(lhs, Variable) and lhs.value == "extra":
|
||||
normalized_extra = canonicalize_name(rhs.value)
|
||||
rhs = Value(normalized_extra)
|
||||
elif isinstance(rhs, Variable) and rhs.value == "extra":
|
||||
normalized_extra = canonicalize_name(lhs.value)
|
||||
lhs = Value(normalized_extra)
|
||||
return lhs, op, rhs
|
||||
|
||||
|
||||
def _normalize_extra_values(results: MarkerList) -> MarkerList:
|
||||
"""
|
||||
Normalize extra values.
|
||||
"""
|
||||
|
||||
return [_normalize_extras(r) for r in results]
|
||||
|
||||
|
||||
def _format_marker(
|
||||
marker: list[str] | MarkerAtom | str, first: bool | None = True
|
||||
) -> str:
|
||||
assert isinstance(marker, (list, tuple, str))
|
||||
|
||||
# Sometimes we have a structure like [[...]] which is a single item list
|
||||
# where the single item is itself it's own list. In that case we want skip
|
||||
# the rest of this function so that we don't get extraneous () on the
|
||||
# outside.
|
||||
if (
|
||||
isinstance(marker, list)
|
||||
and len(marker) == 1
|
||||
and isinstance(marker[0], (list, tuple))
|
||||
):
|
||||
return _format_marker(marker[0])
|
||||
|
||||
if isinstance(marker, list):
|
||||
inner = (_format_marker(m, first=False) for m in marker)
|
||||
if first:
|
||||
return " ".join(inner)
|
||||
else:
|
||||
return "(" + " ".join(inner) + ")"
|
||||
elif isinstance(marker, tuple):
|
||||
return " ".join([m.serialize() for m in marker])
|
||||
else:
|
||||
return marker
|
||||
|
||||
|
||||
_operators: dict[str, Operator] = {
|
||||
"in": lambda lhs, rhs: lhs in rhs,
|
||||
"not in": lambda lhs, rhs: lhs not in rhs,
|
||||
"<": lambda _lhs, _rhs: False,
|
||||
"<=": operator.eq,
|
||||
"==": operator.eq,
|
||||
"!=": operator.ne,
|
||||
">=": operator.eq,
|
||||
">": lambda _lhs, _rhs: False,
|
||||
}
|
||||
|
||||
|
||||
def _eval_op(lhs: str, op: Op, rhs: str | AbstractSet[str], *, key: str) -> bool:
|
||||
op_str = op.serialize()
|
||||
if key in MARKERS_REQUIRING_VERSION:
|
||||
try:
|
||||
spec = Specifier(f"{op_str}{rhs}")
|
||||
except InvalidSpecifier:
|
||||
pass
|
||||
else:
|
||||
return spec.contains(lhs, prereleases=True)
|
||||
|
||||
oper: Operator | None = _operators.get(op_str)
|
||||
if oper is None:
|
||||
raise UndefinedComparison(f"Undefined {op!r} on {lhs!r} and {rhs!r}.")
|
||||
|
||||
return oper(lhs, rhs)
|
||||
|
||||
|
||||
def _normalize(
|
||||
lhs: str, rhs: str | AbstractSet[str], key: str
|
||||
) -> tuple[str, str | AbstractSet[str]]:
|
||||
# PEP 685 - Comparison of extra names for optional distribution dependencies
|
||||
# https://peps.python.org/pep-0685/
|
||||
# > When comparing extra names, tools MUST normalize the names being
|
||||
# > compared using the semantics outlined in PEP 503 for names
|
||||
if key == "extra":
|
||||
assert isinstance(rhs, str), "extra value must be a string"
|
||||
# Both sides are normalized at this point already
|
||||
return (lhs, rhs)
|
||||
if key in MARKERS_ALLOWING_SET:
|
||||
if isinstance(rhs, str): # pragma: no cover
|
||||
return (canonicalize_name(lhs), canonicalize_name(rhs))
|
||||
else:
|
||||
return (canonicalize_name(lhs), {canonicalize_name(v) for v in rhs})
|
||||
|
||||
# other environment markers don't have such standards
|
||||
return lhs, rhs
|
||||
|
||||
|
||||
def _evaluate_markers(
|
||||
markers: MarkerList, environment: dict[str, str | AbstractSet[str]]
|
||||
) -> bool:
|
||||
groups: list[list[bool]] = [[]]
|
||||
|
||||
for marker in markers:
|
||||
if isinstance(marker, list):
|
||||
groups[-1].append(_evaluate_markers(marker, environment))
|
||||
elif isinstance(marker, tuple):
|
||||
lhs, op, rhs = marker
|
||||
|
||||
if isinstance(lhs, Variable):
|
||||
environment_key = lhs.value
|
||||
lhs_value = environment[environment_key]
|
||||
rhs_value = rhs.value
|
||||
else:
|
||||
lhs_value = lhs.value
|
||||
environment_key = rhs.value
|
||||
rhs_value = environment[environment_key]
|
||||
|
||||
assert isinstance(lhs_value, str), "lhs must be a string"
|
||||
lhs_value, rhs_value = _normalize(lhs_value, rhs_value, key=environment_key)
|
||||
groups[-1].append(_eval_op(lhs_value, op, rhs_value, key=environment_key))
|
||||
elif marker == "or":
|
||||
groups.append([])
|
||||
elif marker == "and":
|
||||
pass
|
||||
else: # pragma: nocover
|
||||
raise TypeError(f"Unexpected marker {marker!r}")
|
||||
|
||||
return any(all(item) for item in groups)
|
||||
|
||||
|
||||
def format_full_version(info: sys._version_info) -> str:
|
||||
version = f"{info.major}.{info.minor}.{info.micro}"
|
||||
kind = info.releaselevel
|
||||
if kind != "final":
|
||||
version += kind[0] + str(info.serial)
|
||||
return version
|
||||
|
||||
|
||||
def default_environment() -> Environment:
|
||||
iver = format_full_version(sys.implementation.version)
|
||||
implementation_name = sys.implementation.name
|
||||
return {
|
||||
"implementation_name": implementation_name,
|
||||
"implementation_version": iver,
|
||||
"os_name": os.name,
|
||||
"platform_machine": platform.machine(),
|
||||
"platform_release": platform.release(),
|
||||
"platform_system": platform.system(),
|
||||
"platform_version": platform.version(),
|
||||
"python_full_version": platform.python_version(),
|
||||
"platform_python_implementation": platform.python_implementation(),
|
||||
"python_version": ".".join(platform.python_version_tuple()[:2]),
|
||||
"sys_platform": sys.platform,
|
||||
}
|
||||
|
||||
|
||||
class Marker:
|
||||
def __init__(self, marker: str) -> None:
|
||||
# Note: We create a Marker object without calling this constructor in
|
||||
# packaging.requirements.Requirement. If any additional logic is
|
||||
# added here, make sure to mirror/adapt Requirement.
|
||||
|
||||
# If this fails and throws an error, the repr still expects _markers to
|
||||
# be defined.
|
||||
self._markers: MarkerList = []
|
||||
|
||||
try:
|
||||
self._markers = _normalize_extra_values(_parse_marker(marker))
|
||||
# The attribute `_markers` can be described in terms of a recursive type:
|
||||
# MarkerList = List[Union[Tuple[Node, ...], str, MarkerList]]
|
||||
#
|
||||
# For example, the following expression:
|
||||
# python_version > "3.6" or (python_version == "3.6" and os_name == "unix")
|
||||
#
|
||||
# is parsed into:
|
||||
# [
|
||||
# (<Variable('python_version')>, <Op('>')>, <Value('3.6')>),
|
||||
# 'and',
|
||||
# [
|
||||
# (<Variable('python_version')>, <Op('==')>, <Value('3.6')>),
|
||||
# 'or',
|
||||
# (<Variable('os_name')>, <Op('==')>, <Value('unix')>)
|
||||
# ]
|
||||
# ]
|
||||
except ParserSyntaxError as e:
|
||||
raise InvalidMarker(str(e)) from e
|
||||
|
||||
def __str__(self) -> str:
|
||||
return _format_marker(self._markers)
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"<{self.__class__.__name__}('{self}')>"
|
||||
|
||||
def __hash__(self) -> int:
|
||||
return hash(str(self))
|
||||
|
||||
def __eq__(self, other: object) -> bool:
|
||||
if not isinstance(other, Marker):
|
||||
return NotImplemented
|
||||
|
||||
return str(self) == str(other)
|
||||
|
||||
def evaluate(
|
||||
self,
|
||||
environment: Mapping[str, str | AbstractSet[str]] | None = None,
|
||||
context: EvaluateContext = "metadata",
|
||||
) -> bool:
|
||||
"""Evaluate a marker.
|
||||
|
||||
Return the boolean from evaluating the given marker against the
|
||||
environment. environment is an optional argument to override all or
|
||||
part of the determined environment. The *context* parameter specifies what
|
||||
context the markers are being evaluated for, which influences what markers
|
||||
are considered valid. Acceptable values are "metadata" (for core metadata;
|
||||
default), "lock_file", and "requirement" (i.e. all other situations).
|
||||
|
||||
The environment is determined from the current Python process.
|
||||
"""
|
||||
current_environment = cast(
|
||||
"dict[str, str | AbstractSet[str]]", default_environment()
|
||||
)
|
||||
if context == "lock_file":
|
||||
current_environment.update(
|
||||
extras=frozenset(), dependency_groups=frozenset()
|
||||
)
|
||||
elif context == "metadata":
|
||||
current_environment["extra"] = ""
|
||||
|
||||
if environment is not None:
|
||||
current_environment.update(environment)
|
||||
if "extra" in current_environment:
|
||||
# The API used to allow setting extra to None. We need to handle
|
||||
# this case for backwards compatibility. Also skip running
|
||||
# normalize name if extra is empty.
|
||||
extra = cast("str | None", current_environment["extra"])
|
||||
current_environment["extra"] = canonicalize_name(extra) if extra else ""
|
||||
|
||||
return _evaluate_markers(
|
||||
self._markers, _repair_python_full_version(current_environment)
|
||||
)
|
||||
|
||||
|
||||
def _repair_python_full_version(
|
||||
env: dict[str, str | AbstractSet[str]],
|
||||
) -> dict[str, str | AbstractSet[str]]:
|
||||
"""
|
||||
Work around platform.python_version() returning something that is not PEP 440
|
||||
compliant for non-tagged Python builds.
|
||||
"""
|
||||
python_full_version = cast("str", env["python_full_version"])
|
||||
if python_full_version.endswith("+"):
|
||||
env["python_full_version"] = f"{python_full_version}local"
|
||||
return env
|
||||
978
venv/Lib/site-packages/pip/_vendor/packaking/metadata.py
Normal file
978
venv/Lib/site-packages/pip/_vendor/packaking/metadata.py
Normal file
@@ -0,0 +1,978 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import email.feedparser
|
||||
import email.header
|
||||
import email.message
|
||||
import email.parser
|
||||
import email.policy
|
||||
import keyword
|
||||
import pathlib
|
||||
import sys
|
||||
import typing
|
||||
from typing import (
|
||||
Any,
|
||||
Callable,
|
||||
Generic,
|
||||
Literal,
|
||||
TypedDict,
|
||||
cast,
|
||||
)
|
||||
|
||||
from . import licenses, requirements, specifiers, utils
|
||||
from . import version as version_module
|
||||
|
||||
if typing.TYPE_CHECKING:
|
||||
from .licenses import NormalizedLicenseExpression
|
||||
|
||||
T = typing.TypeVar("T")
|
||||
|
||||
|
||||
if sys.version_info >= (3, 11): # pragma: no cover
|
||||
ExceptionGroup = ExceptionGroup # noqa: F821
|
||||
else: # pragma: no cover
|
||||
|
||||
class ExceptionGroup(Exception):
|
||||
"""A minimal implementation of :external:exc:`ExceptionGroup` from Python 3.11.
|
||||
|
||||
If :external:exc:`ExceptionGroup` is already defined by Python itself,
|
||||
that version is used instead.
|
||||
"""
|
||||
|
||||
message: str
|
||||
exceptions: list[Exception]
|
||||
|
||||
def __init__(self, message: str, exceptions: list[Exception]) -> None:
|
||||
self.message = message
|
||||
self.exceptions = exceptions
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"{self.__class__.__name__}({self.message!r}, {self.exceptions!r})"
|
||||
|
||||
|
||||
class InvalidMetadata(ValueError):
|
||||
"""A metadata field contains invalid data."""
|
||||
|
||||
field: str
|
||||
"""The name of the field that contains invalid data."""
|
||||
|
||||
def __init__(self, field: str, message: str) -> None:
|
||||
self.field = field
|
||||
super().__init__(message)
|
||||
|
||||
|
||||
# The RawMetadata class attempts to make as few assumptions about the underlying
|
||||
# serialization formats as possible. The idea is that as long as a serialization
|
||||
# formats offer some very basic primitives in *some* way then we can support
|
||||
# serializing to and from that format.
|
||||
class RawMetadata(TypedDict, total=False):
|
||||
"""A dictionary of raw core metadata.
|
||||
|
||||
Each field in core metadata maps to a key of this dictionary (when data is
|
||||
provided). The key is lower-case and underscores are used instead of dashes
|
||||
compared to the equivalent core metadata field. Any core metadata field that
|
||||
can be specified multiple times or can hold multiple values in a single
|
||||
field have a key with a plural name. See :class:`Metadata` whose attributes
|
||||
match the keys of this dictionary.
|
||||
|
||||
Core metadata fields that can be specified multiple times are stored as a
|
||||
list or dict depending on which is appropriate for the field. Any fields
|
||||
which hold multiple values in a single field are stored as a list.
|
||||
|
||||
"""
|
||||
|
||||
# Metadata 1.0 - PEP 241
|
||||
metadata_version: str
|
||||
name: str
|
||||
version: str
|
||||
platforms: list[str]
|
||||
summary: str
|
||||
description: str
|
||||
keywords: list[str]
|
||||
home_page: str
|
||||
author: str
|
||||
author_email: str
|
||||
license: str
|
||||
|
||||
# Metadata 1.1 - PEP 314
|
||||
supported_platforms: list[str]
|
||||
download_url: str
|
||||
classifiers: list[str]
|
||||
requires: list[str]
|
||||
provides: list[str]
|
||||
obsoletes: list[str]
|
||||
|
||||
# Metadata 1.2 - PEP 345
|
||||
maintainer: str
|
||||
maintainer_email: str
|
||||
requires_dist: list[str]
|
||||
provides_dist: list[str]
|
||||
obsoletes_dist: list[str]
|
||||
requires_python: str
|
||||
requires_external: list[str]
|
||||
project_urls: dict[str, str]
|
||||
|
||||
# Metadata 2.0
|
||||
# PEP 426 attempted to completely revamp the metadata format
|
||||
# but got stuck without ever being able to build consensus on
|
||||
# it and ultimately ended up withdrawn.
|
||||
#
|
||||
# However, a number of tools had started emitting METADATA with
|
||||
# `2.0` Metadata-Version, so for historical reasons, this version
|
||||
# was skipped.
|
||||
|
||||
# Metadata 2.1 - PEP 566
|
||||
description_content_type: str
|
||||
provides_extra: list[str]
|
||||
|
||||
# Metadata 2.2 - PEP 643
|
||||
dynamic: list[str]
|
||||
|
||||
# Metadata 2.3 - PEP 685
|
||||
# No new fields were added in PEP 685, just some edge case were
|
||||
# tightened up to provide better interoperability.
|
||||
|
||||
# Metadata 2.4 - PEP 639
|
||||
license_expression: str
|
||||
license_files: list[str]
|
||||
|
||||
# Metadata 2.5 - PEP 794
|
||||
import_names: list[str]
|
||||
import_namespaces: list[str]
|
||||
|
||||
|
||||
# 'keywords' is special as it's a string in the core metadata spec, but we
|
||||
# represent it as a list.
|
||||
_STRING_FIELDS = {
|
||||
"author",
|
||||
"author_email",
|
||||
"description",
|
||||
"description_content_type",
|
||||
"download_url",
|
||||
"home_page",
|
||||
"license",
|
||||
"license_expression",
|
||||
"maintainer",
|
||||
"maintainer_email",
|
||||
"metadata_version",
|
||||
"name",
|
||||
"requires_python",
|
||||
"summary",
|
||||
"version",
|
||||
}
|
||||
|
||||
_LIST_FIELDS = {
|
||||
"classifiers",
|
||||
"dynamic",
|
||||
"license_files",
|
||||
"obsoletes",
|
||||
"obsoletes_dist",
|
||||
"platforms",
|
||||
"provides",
|
||||
"provides_dist",
|
||||
"provides_extra",
|
||||
"requires",
|
||||
"requires_dist",
|
||||
"requires_external",
|
||||
"supported_platforms",
|
||||
"import_names",
|
||||
"import_namespaces",
|
||||
}
|
||||
|
||||
_DICT_FIELDS = {
|
||||
"project_urls",
|
||||
}
|
||||
|
||||
|
||||
def _parse_keywords(data: str) -> list[str]:
|
||||
"""Split a string of comma-separated keywords into a list of keywords."""
|
||||
return [k.strip() for k in data.split(",")]
|
||||
|
||||
|
||||
def _parse_project_urls(data: list[str]) -> dict[str, str]:
|
||||
"""Parse a list of label/URL string pairings separated by a comma."""
|
||||
urls = {}
|
||||
for pair in data:
|
||||
# Our logic is slightly tricky here as we want to try and do
|
||||
# *something* reasonable with malformed data.
|
||||
#
|
||||
# The main thing that we have to worry about, is data that does
|
||||
# not have a ',' at all to split the label from the Value. There
|
||||
# isn't a singular right answer here, and we will fail validation
|
||||
# later on (if the caller is validating) so it doesn't *really*
|
||||
# matter, but since the missing value has to be an empty str
|
||||
# and our return value is dict[str, str], if we let the key
|
||||
# be the missing value, then they'd have multiple '' values that
|
||||
# overwrite each other in a accumulating dict.
|
||||
#
|
||||
# The other potential issue is that it's possible to have the
|
||||
# same label multiple times in the metadata, with no solid "right"
|
||||
# answer with what to do in that case. As such, we'll do the only
|
||||
# thing we can, which is treat the field as unparsable and add it
|
||||
# to our list of unparsed fields.
|
||||
#
|
||||
# TODO: The spec doesn't say anything about if the keys should be
|
||||
# considered case sensitive or not... logically they should
|
||||
# be case-preserving and case-insensitive, but doing that
|
||||
# would open up more cases where we might have duplicate
|
||||
# entries.
|
||||
label, _, url = (s.strip() for s in pair.partition(","))
|
||||
|
||||
if label in urls:
|
||||
# The label already exists in our set of urls, so this field
|
||||
# is unparsable, and we can just add the whole thing to our
|
||||
# unparsable data and stop processing it.
|
||||
raise KeyError("duplicate labels in project urls")
|
||||
urls[label] = url
|
||||
|
||||
return urls
|
||||
|
||||
|
||||
def _get_payload(msg: email.message.Message, source: bytes | str) -> str:
|
||||
"""Get the body of the message."""
|
||||
# If our source is a str, then our caller has managed encodings for us,
|
||||
# and we don't need to deal with it.
|
||||
if isinstance(source, str):
|
||||
payload = msg.get_payload()
|
||||
assert isinstance(payload, str)
|
||||
return payload
|
||||
# If our source is a bytes, then we're managing the encoding and we need
|
||||
# to deal with it.
|
||||
else:
|
||||
bpayload = msg.get_payload(decode=True)
|
||||
assert isinstance(bpayload, bytes)
|
||||
try:
|
||||
return bpayload.decode("utf8", "strict")
|
||||
except UnicodeDecodeError as exc:
|
||||
raise ValueError("payload in an invalid encoding") from exc
|
||||
|
||||
|
||||
# The various parse_FORMAT functions here are intended to be as lenient as
|
||||
# possible in their parsing, while still returning a correctly typed
|
||||
# RawMetadata.
|
||||
#
|
||||
# To aid in this, we also generally want to do as little touching of the
|
||||
# data as possible, except where there are possibly some historic holdovers
|
||||
# that make valid data awkward to work with.
|
||||
#
|
||||
# While this is a lower level, intermediate format than our ``Metadata``
|
||||
# class, some light touch ups can make a massive difference in usability.
|
||||
|
||||
# Map METADATA fields to RawMetadata.
|
||||
_EMAIL_TO_RAW_MAPPING = {
|
||||
"author": "author",
|
||||
"author-email": "author_email",
|
||||
"classifier": "classifiers",
|
||||
"description": "description",
|
||||
"description-content-type": "description_content_type",
|
||||
"download-url": "download_url",
|
||||
"dynamic": "dynamic",
|
||||
"home-page": "home_page",
|
||||
"import-name": "import_names",
|
||||
"import-namespace": "import_namespaces",
|
||||
"keywords": "keywords",
|
||||
"license": "license",
|
||||
"license-expression": "license_expression",
|
||||
"license-file": "license_files",
|
||||
"maintainer": "maintainer",
|
||||
"maintainer-email": "maintainer_email",
|
||||
"metadata-version": "metadata_version",
|
||||
"name": "name",
|
||||
"obsoletes": "obsoletes",
|
||||
"obsoletes-dist": "obsoletes_dist",
|
||||
"platform": "platforms",
|
||||
"project-url": "project_urls",
|
||||
"provides": "provides",
|
||||
"provides-dist": "provides_dist",
|
||||
"provides-extra": "provides_extra",
|
||||
"requires": "requires",
|
||||
"requires-dist": "requires_dist",
|
||||
"requires-external": "requires_external",
|
||||
"requires-python": "requires_python",
|
||||
"summary": "summary",
|
||||
"supported-platform": "supported_platforms",
|
||||
"version": "version",
|
||||
}
|
||||
_RAW_TO_EMAIL_MAPPING = {raw: email for email, raw in _EMAIL_TO_RAW_MAPPING.items()}
|
||||
|
||||
|
||||
# This class is for writing RFC822 messages
|
||||
class RFC822Policy(email.policy.EmailPolicy):
|
||||
"""
|
||||
This is :class:`email.policy.EmailPolicy`, but with a simple ``header_store_parse``
|
||||
implementation that handles multi-line values, and some nice defaults.
|
||||
"""
|
||||
|
||||
utf8 = True
|
||||
mangle_from_ = False
|
||||
max_line_length = 0
|
||||
|
||||
def header_store_parse(self, name: str, value: str) -> tuple[str, str]:
|
||||
size = len(name) + 2
|
||||
value = value.replace("\n", "\n" + " " * size)
|
||||
return (name, value)
|
||||
|
||||
|
||||
# This class is for writing RFC822 messages
|
||||
class RFC822Message(email.message.EmailMessage):
|
||||
"""
|
||||
This is :class:`email.message.EmailMessage` with two small changes: it defaults to
|
||||
our `RFC822Policy`, and it correctly writes unicode when being called
|
||||
with `bytes()`.
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
super().__init__(policy=RFC822Policy())
|
||||
|
||||
def as_bytes(
|
||||
self, unixfrom: bool = False, policy: email.policy.Policy | None = None
|
||||
) -> bytes:
|
||||
"""
|
||||
Return the bytes representation of the message.
|
||||
|
||||
This handles unicode encoding.
|
||||
"""
|
||||
return self.as_string(unixfrom, policy=policy).encode("utf-8")
|
||||
|
||||
|
||||
def parse_email(data: bytes | str) -> tuple[RawMetadata, dict[str, list[str]]]:
|
||||
"""Parse a distribution's metadata stored as email headers (e.g. from ``METADATA``).
|
||||
|
||||
This function returns a two-item tuple of dicts. The first dict is of
|
||||
recognized fields from the core metadata specification. Fields that can be
|
||||
parsed and translated into Python's built-in types are converted
|
||||
appropriately. All other fields are left as-is. Fields that are allowed to
|
||||
appear multiple times are stored as lists.
|
||||
|
||||
The second dict contains all other fields from the metadata. This includes
|
||||
any unrecognized fields. It also includes any fields which are expected to
|
||||
be parsed into a built-in type but were not formatted appropriately. Finally,
|
||||
any fields that are expected to appear only once but are repeated are
|
||||
included in this dict.
|
||||
|
||||
"""
|
||||
raw: dict[str, str | list[str] | dict[str, str]] = {}
|
||||
unparsed: dict[str, list[str]] = {}
|
||||
|
||||
if isinstance(data, str):
|
||||
parsed = email.parser.Parser(policy=email.policy.compat32).parsestr(data)
|
||||
else:
|
||||
parsed = email.parser.BytesParser(policy=email.policy.compat32).parsebytes(data)
|
||||
|
||||
# We have to wrap parsed.keys() in a set, because in the case of multiple
|
||||
# values for a key (a list), the key will appear multiple times in the
|
||||
# list of keys, but we're avoiding that by using get_all().
|
||||
for name_with_case in frozenset(parsed.keys()):
|
||||
# Header names in RFC are case insensitive, so we'll normalize to all
|
||||
# lower case to make comparisons easier.
|
||||
name = name_with_case.lower()
|
||||
|
||||
# We use get_all() here, even for fields that aren't multiple use,
|
||||
# because otherwise someone could have e.g. two Name fields, and we
|
||||
# would just silently ignore it rather than doing something about it.
|
||||
headers = parsed.get_all(name) or []
|
||||
|
||||
# The way the email module works when parsing bytes is that it
|
||||
# unconditionally decodes the bytes as ascii using the surrogateescape
|
||||
# handler. When you pull that data back out (such as with get_all() ),
|
||||
# it looks to see if the str has any surrogate escapes, and if it does
|
||||
# it wraps it in a Header object instead of returning the string.
|
||||
#
|
||||
# As such, we'll look for those Header objects, and fix up the encoding.
|
||||
value = []
|
||||
# Flag if we have run into any issues processing the headers, thus
|
||||
# signalling that the data belongs in 'unparsed'.
|
||||
valid_encoding = True
|
||||
for h in headers:
|
||||
# It's unclear if this can return more types than just a Header or
|
||||
# a str, so we'll just assert here to make sure.
|
||||
assert isinstance(h, (email.header.Header, str))
|
||||
|
||||
# If it's a header object, we need to do our little dance to get
|
||||
# the real data out of it. In cases where there is invalid data
|
||||
# we're going to end up with mojibake, but there's no obvious, good
|
||||
# way around that without reimplementing parts of the Header object
|
||||
# ourselves.
|
||||
#
|
||||
# That should be fine since, if mojibacked happens, this key is
|
||||
# going into the unparsed dict anyways.
|
||||
if isinstance(h, email.header.Header):
|
||||
# The Header object stores it's data as chunks, and each chunk
|
||||
# can be independently encoded, so we'll need to check each
|
||||
# of them.
|
||||
chunks: list[tuple[bytes, str | None]] = []
|
||||
for binary, _encoding in email.header.decode_header(h):
|
||||
try:
|
||||
binary.decode("utf8", "strict")
|
||||
except UnicodeDecodeError:
|
||||
# Enable mojibake.
|
||||
encoding = "latin1"
|
||||
valid_encoding = False
|
||||
else:
|
||||
encoding = "utf8"
|
||||
chunks.append((binary, encoding))
|
||||
|
||||
# Turn our chunks back into a Header object, then let that
|
||||
# Header object do the right thing to turn them into a
|
||||
# string for us.
|
||||
value.append(str(email.header.make_header(chunks)))
|
||||
# This is already a string, so just add it.
|
||||
else:
|
||||
value.append(h)
|
||||
|
||||
# We've processed all of our values to get them into a list of str,
|
||||
# but we may have mojibake data, in which case this is an unparsed
|
||||
# field.
|
||||
if not valid_encoding:
|
||||
unparsed[name] = value
|
||||
continue
|
||||
|
||||
raw_name = _EMAIL_TO_RAW_MAPPING.get(name)
|
||||
if raw_name is None:
|
||||
# This is a bit of a weird situation, we've encountered a key that
|
||||
# we don't know what it means, so we don't know whether it's meant
|
||||
# to be a list or not.
|
||||
#
|
||||
# Since we can't really tell one way or another, we'll just leave it
|
||||
# as a list, even though it may be a single item list, because that's
|
||||
# what makes the most sense for email headers.
|
||||
unparsed[name] = value
|
||||
continue
|
||||
|
||||
# If this is one of our string fields, then we'll check to see if our
|
||||
# value is a list of a single item. If it is then we'll assume that
|
||||
# it was emitted as a single string, and unwrap the str from inside
|
||||
# the list.
|
||||
#
|
||||
# If it's any other kind of data, then we haven't the faintest clue
|
||||
# what we should parse it as, and we have to just add it to our list
|
||||
# of unparsed stuff.
|
||||
if raw_name in _STRING_FIELDS and len(value) == 1:
|
||||
raw[raw_name] = value[0]
|
||||
# If this is import_names, we need to special case the empty field
|
||||
# case, which converts to an empty list instead of None. We can't let
|
||||
# the empty case slip through, as it will fail validation.
|
||||
elif raw_name == "import_names" and value == [""]:
|
||||
raw[raw_name] = []
|
||||
# If this is one of our list of string fields, then we can just assign
|
||||
# the value, since email *only* has strings, and our get_all() call
|
||||
# above ensures that this is a list.
|
||||
elif raw_name in _LIST_FIELDS:
|
||||
raw[raw_name] = value
|
||||
# Special Case: Keywords
|
||||
# The keywords field is implemented in the metadata spec as a str,
|
||||
# but it conceptually is a list of strings, and is serialized using
|
||||
# ", ".join(keywords), so we'll do some light data massaging to turn
|
||||
# this into what it logically is.
|
||||
elif raw_name == "keywords" and len(value) == 1:
|
||||
raw[raw_name] = _parse_keywords(value[0])
|
||||
# Special Case: Project-URL
|
||||
# The project urls is implemented in the metadata spec as a list of
|
||||
# specially-formatted strings that represent a key and a value, which
|
||||
# is fundamentally a mapping, however the email format doesn't support
|
||||
# mappings in a sane way, so it was crammed into a list of strings
|
||||
# instead.
|
||||
#
|
||||
# We will do a little light data massaging to turn this into a map as
|
||||
# it logically should be.
|
||||
elif raw_name == "project_urls":
|
||||
try:
|
||||
raw[raw_name] = _parse_project_urls(value)
|
||||
except KeyError:
|
||||
unparsed[name] = value
|
||||
# Nothing that we've done has managed to parse this, so it'll just
|
||||
# throw it in our unparsable data and move on.
|
||||
else:
|
||||
unparsed[name] = value
|
||||
|
||||
# We need to support getting the Description from the message payload in
|
||||
# addition to getting it from the the headers. This does mean, though, there
|
||||
# is the possibility of it being set both ways, in which case we put both
|
||||
# in 'unparsed' since we don't know which is right.
|
||||
try:
|
||||
payload = _get_payload(parsed, data)
|
||||
except ValueError:
|
||||
unparsed.setdefault("description", []).append(
|
||||
parsed.get_payload(decode=isinstance(data, bytes)) # type: ignore[call-overload]
|
||||
)
|
||||
else:
|
||||
if payload:
|
||||
# Check to see if we've already got a description, if so then both
|
||||
# it, and this body move to unparsable.
|
||||
if "description" in raw:
|
||||
description_header = cast("str", raw.pop("description"))
|
||||
unparsed.setdefault("description", []).extend(
|
||||
[description_header, payload]
|
||||
)
|
||||
elif "description" in unparsed:
|
||||
unparsed["description"].append(payload)
|
||||
else:
|
||||
raw["description"] = payload
|
||||
|
||||
# We need to cast our `raw` to a metadata, because a TypedDict only support
|
||||
# literal key names, but we're computing our key names on purpose, but the
|
||||
# way this function is implemented, our `TypedDict` can only have valid key
|
||||
# names.
|
||||
return cast("RawMetadata", raw), unparsed
|
||||
|
||||
|
||||
_NOT_FOUND = object()
|
||||
|
||||
|
||||
# Keep the two values in sync.
|
||||
_VALID_METADATA_VERSIONS = ["1.0", "1.1", "1.2", "2.1", "2.2", "2.3", "2.4", "2.5"]
|
||||
_MetadataVersion = Literal["1.0", "1.1", "1.2", "2.1", "2.2", "2.3", "2.4", "2.5"]
|
||||
|
||||
_REQUIRED_ATTRS = frozenset(["metadata_version", "name", "version"])
|
||||
|
||||
|
||||
class _Validator(Generic[T]):
|
||||
"""Validate a metadata field.
|
||||
|
||||
All _process_*() methods correspond to a core metadata field. The method is
|
||||
called with the field's raw value. If the raw value is valid it is returned
|
||||
in its "enriched" form (e.g. ``version.Version`` for the ``Version`` field).
|
||||
If the raw value is invalid, :exc:`InvalidMetadata` is raised (with a cause
|
||||
as appropriate).
|
||||
"""
|
||||
|
||||
name: str
|
||||
raw_name: str
|
||||
added: _MetadataVersion
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
added: _MetadataVersion = "1.0",
|
||||
) -> None:
|
||||
self.added = added
|
||||
|
||||
def __set_name__(self, _owner: Metadata, name: str) -> None:
|
||||
self.name = name
|
||||
self.raw_name = _RAW_TO_EMAIL_MAPPING[name]
|
||||
|
||||
def __get__(self, instance: Metadata, _owner: type[Metadata]) -> T:
|
||||
# With Python 3.8, the caching can be replaced with functools.cached_property().
|
||||
# No need to check the cache as attribute lookup will resolve into the
|
||||
# instance's __dict__ before __get__ is called.
|
||||
cache = instance.__dict__
|
||||
value = instance._raw.get(self.name)
|
||||
|
||||
# To make the _process_* methods easier, we'll check if the value is None
|
||||
# and if this field is NOT a required attribute, and if both of those
|
||||
# things are true, we'll skip the the converter. This will mean that the
|
||||
# converters never have to deal with the None union.
|
||||
if self.name in _REQUIRED_ATTRS or value is not None:
|
||||
try:
|
||||
converter: Callable[[Any], T] = getattr(self, f"_process_{self.name}")
|
||||
except AttributeError:
|
||||
pass
|
||||
else:
|
||||
value = converter(value)
|
||||
|
||||
cache[self.name] = value
|
||||
try:
|
||||
del instance._raw[self.name] # type: ignore[misc]
|
||||
except KeyError:
|
||||
pass
|
||||
|
||||
return cast("T", value)
|
||||
|
||||
def _invalid_metadata(
|
||||
self, msg: str, cause: Exception | None = None
|
||||
) -> InvalidMetadata:
|
||||
exc = InvalidMetadata(
|
||||
self.raw_name, msg.format_map({"field": repr(self.raw_name)})
|
||||
)
|
||||
exc.__cause__ = cause
|
||||
return exc
|
||||
|
||||
def _process_metadata_version(self, value: str) -> _MetadataVersion:
|
||||
# Implicitly makes Metadata-Version required.
|
||||
if value not in _VALID_METADATA_VERSIONS:
|
||||
raise self._invalid_metadata(f"{value!r} is not a valid metadata version")
|
||||
return cast("_MetadataVersion", value)
|
||||
|
||||
def _process_name(self, value: str) -> str:
|
||||
if not value:
|
||||
raise self._invalid_metadata("{field} is a required field")
|
||||
# Validate the name as a side-effect.
|
||||
try:
|
||||
utils.canonicalize_name(value, validate=True)
|
||||
except utils.InvalidName as exc:
|
||||
raise self._invalid_metadata(
|
||||
f"{value!r} is invalid for {{field}}", cause=exc
|
||||
) from exc
|
||||
else:
|
||||
return value
|
||||
|
||||
def _process_version(self, value: str) -> version_module.Version:
|
||||
if not value:
|
||||
raise self._invalid_metadata("{field} is a required field")
|
||||
try:
|
||||
return version_module.parse(value)
|
||||
except version_module.InvalidVersion as exc:
|
||||
raise self._invalid_metadata(
|
||||
f"{value!r} is invalid for {{field}}", cause=exc
|
||||
) from exc
|
||||
|
||||
def _process_summary(self, value: str) -> str:
|
||||
"""Check the field contains no newlines."""
|
||||
if "\n" in value:
|
||||
raise self._invalid_metadata("{field} must be a single line")
|
||||
return value
|
||||
|
||||
def _process_description_content_type(self, value: str) -> str:
|
||||
content_types = {"text/plain", "text/x-rst", "text/markdown"}
|
||||
message = email.message.EmailMessage()
|
||||
message["content-type"] = value
|
||||
|
||||
content_type, parameters = (
|
||||
# Defaults to `text/plain` if parsing failed.
|
||||
message.get_content_type().lower(),
|
||||
message["content-type"].params,
|
||||
)
|
||||
# Check if content-type is valid or defaulted to `text/plain` and thus was
|
||||
# not parseable.
|
||||
if content_type not in content_types or content_type not in value.lower():
|
||||
raise self._invalid_metadata(
|
||||
f"{{field}} must be one of {list(content_types)}, not {value!r}"
|
||||
)
|
||||
|
||||
charset = parameters.get("charset", "UTF-8")
|
||||
if charset != "UTF-8":
|
||||
raise self._invalid_metadata(
|
||||
f"{{field}} can only specify the UTF-8 charset, not {list(charset)}"
|
||||
)
|
||||
|
||||
markdown_variants = {"GFM", "CommonMark"}
|
||||
variant = parameters.get("variant", "GFM") # Use an acceptable default.
|
||||
if content_type == "text/markdown" and variant not in markdown_variants:
|
||||
raise self._invalid_metadata(
|
||||
f"valid Markdown variants for {{field}} are {list(markdown_variants)}, "
|
||||
f"not {variant!r}",
|
||||
)
|
||||
return value
|
||||
|
||||
def _process_dynamic(self, value: list[str]) -> list[str]:
|
||||
for dynamic_field in map(str.lower, value):
|
||||
if dynamic_field in {"name", "version", "metadata-version"}:
|
||||
raise self._invalid_metadata(
|
||||
f"{dynamic_field!r} is not allowed as a dynamic field"
|
||||
)
|
||||
elif dynamic_field not in _EMAIL_TO_RAW_MAPPING:
|
||||
raise self._invalid_metadata(
|
||||
f"{dynamic_field!r} is not a valid dynamic field"
|
||||
)
|
||||
return list(map(str.lower, value))
|
||||
|
||||
def _process_provides_extra(
|
||||
self,
|
||||
value: list[str],
|
||||
) -> list[utils.NormalizedName]:
|
||||
normalized_names = []
|
||||
try:
|
||||
for name in value:
|
||||
normalized_names.append(utils.canonicalize_name(name, validate=True))
|
||||
except utils.InvalidName as exc:
|
||||
raise self._invalid_metadata(
|
||||
f"{name!r} is invalid for {{field}}", cause=exc
|
||||
) from exc
|
||||
else:
|
||||
return normalized_names
|
||||
|
||||
def _process_requires_python(self, value: str) -> specifiers.SpecifierSet:
|
||||
try:
|
||||
return specifiers.SpecifierSet(value)
|
||||
except specifiers.InvalidSpecifier as exc:
|
||||
raise self._invalid_metadata(
|
||||
f"{value!r} is invalid for {{field}}", cause=exc
|
||||
) from exc
|
||||
|
||||
def _process_requires_dist(
|
||||
self,
|
||||
value: list[str],
|
||||
) -> list[requirements.Requirement]:
|
||||
reqs = []
|
||||
try:
|
||||
for req in value:
|
||||
reqs.append(requirements.Requirement(req))
|
||||
except requirements.InvalidRequirement as exc:
|
||||
raise self._invalid_metadata(
|
||||
f"{req!r} is invalid for {{field}}", cause=exc
|
||||
) from exc
|
||||
else:
|
||||
return reqs
|
||||
|
||||
def _process_license_expression(self, value: str) -> NormalizedLicenseExpression:
|
||||
try:
|
||||
return licenses.canonicalize_license_expression(value)
|
||||
except ValueError as exc:
|
||||
raise self._invalid_metadata(
|
||||
f"{value!r} is invalid for {{field}}", cause=exc
|
||||
) from exc
|
||||
|
||||
def _process_license_files(self, value: list[str]) -> list[str]:
|
||||
paths = []
|
||||
for path in value:
|
||||
if ".." in path:
|
||||
raise self._invalid_metadata(
|
||||
f"{path!r} is invalid for {{field}}, "
|
||||
"parent directory indicators are not allowed"
|
||||
)
|
||||
if "*" in path:
|
||||
raise self._invalid_metadata(
|
||||
f"{path!r} is invalid for {{field}}, paths must be resolved"
|
||||
)
|
||||
if (
|
||||
pathlib.PurePosixPath(path).is_absolute()
|
||||
or pathlib.PureWindowsPath(path).is_absolute()
|
||||
):
|
||||
raise self._invalid_metadata(
|
||||
f"{path!r} is invalid for {{field}}, paths must be relative"
|
||||
)
|
||||
if pathlib.PureWindowsPath(path).as_posix() != path:
|
||||
raise self._invalid_metadata(
|
||||
f"{path!r} is invalid for {{field}}, paths must use '/' delimiter"
|
||||
)
|
||||
paths.append(path)
|
||||
return paths
|
||||
|
||||
def _process_import_names(self, value: list[str]) -> list[str]:
|
||||
for import_name in value:
|
||||
name, semicolon, private = import_name.partition(";")
|
||||
name = name.rstrip()
|
||||
for identifier in name.split("."):
|
||||
if not identifier.isidentifier():
|
||||
raise self._invalid_metadata(
|
||||
f"{name!r} is invalid for {{field}}; "
|
||||
f"{identifier!r} is not a valid identifier"
|
||||
)
|
||||
elif keyword.iskeyword(identifier):
|
||||
raise self._invalid_metadata(
|
||||
f"{name!r} is invalid for {{field}}; "
|
||||
f"{identifier!r} is a keyword"
|
||||
)
|
||||
if semicolon and private.lstrip() != "private":
|
||||
raise self._invalid_metadata(
|
||||
f"{import_name!r} is invalid for {{field}}; "
|
||||
"the only valid option is 'private'"
|
||||
)
|
||||
return value
|
||||
|
||||
_process_import_namespaces = _process_import_names
|
||||
|
||||
|
||||
class Metadata:
|
||||
"""Representation of distribution metadata.
|
||||
|
||||
Compared to :class:`RawMetadata`, this class provides objects representing
|
||||
metadata fields instead of only using built-in types. Any invalid metadata
|
||||
will cause :exc:`InvalidMetadata` to be raised (with a
|
||||
:py:attr:`~BaseException.__cause__` attribute as appropriate).
|
||||
"""
|
||||
|
||||
_raw: RawMetadata
|
||||
|
||||
@classmethod
|
||||
def from_raw(cls, data: RawMetadata, *, validate: bool = True) -> Metadata:
|
||||
"""Create an instance from :class:`RawMetadata`.
|
||||
|
||||
If *validate* is true, all metadata will be validated. All exceptions
|
||||
related to validation will be gathered and raised as an :class:`ExceptionGroup`.
|
||||
"""
|
||||
ins = cls()
|
||||
ins._raw = data.copy() # Mutations occur due to caching enriched values.
|
||||
|
||||
if validate:
|
||||
exceptions: list[Exception] = []
|
||||
try:
|
||||
metadata_version = ins.metadata_version
|
||||
metadata_age = _VALID_METADATA_VERSIONS.index(metadata_version)
|
||||
except InvalidMetadata as metadata_version_exc:
|
||||
exceptions.append(metadata_version_exc)
|
||||
metadata_version = None
|
||||
|
||||
# Make sure to check for the fields that are present, the required
|
||||
# fields (so their absence can be reported).
|
||||
fields_to_check = frozenset(ins._raw) | _REQUIRED_ATTRS
|
||||
# Remove fields that have already been checked.
|
||||
fields_to_check -= {"metadata_version"}
|
||||
|
||||
for key in fields_to_check:
|
||||
try:
|
||||
if metadata_version:
|
||||
# Can't use getattr() as that triggers descriptor protocol which
|
||||
# will fail due to no value for the instance argument.
|
||||
try:
|
||||
field_metadata_version = cls.__dict__[key].added
|
||||
except KeyError:
|
||||
exc = InvalidMetadata(key, f"unrecognized field: {key!r}")
|
||||
exceptions.append(exc)
|
||||
continue
|
||||
field_age = _VALID_METADATA_VERSIONS.index(
|
||||
field_metadata_version
|
||||
)
|
||||
if field_age > metadata_age:
|
||||
field = _RAW_TO_EMAIL_MAPPING[key]
|
||||
exc = InvalidMetadata(
|
||||
field,
|
||||
f"{field} introduced in metadata version "
|
||||
f"{field_metadata_version}, not {metadata_version}",
|
||||
)
|
||||
exceptions.append(exc)
|
||||
continue
|
||||
getattr(ins, key)
|
||||
except InvalidMetadata as exc:
|
||||
exceptions.append(exc)
|
||||
|
||||
if exceptions:
|
||||
raise ExceptionGroup("invalid metadata", exceptions)
|
||||
|
||||
return ins
|
||||
|
||||
@classmethod
|
||||
def from_email(cls, data: bytes | str, *, validate: bool = True) -> Metadata:
|
||||
"""Parse metadata from email headers.
|
||||
|
||||
If *validate* is true, the metadata will be validated. All exceptions
|
||||
related to validation will be gathered and raised as an :class:`ExceptionGroup`.
|
||||
"""
|
||||
raw, unparsed = parse_email(data)
|
||||
|
||||
if validate:
|
||||
exceptions: list[Exception] = []
|
||||
for unparsed_key in unparsed:
|
||||
if unparsed_key in _EMAIL_TO_RAW_MAPPING:
|
||||
message = f"{unparsed_key!r} has invalid data"
|
||||
else:
|
||||
message = f"unrecognized field: {unparsed_key!r}"
|
||||
exceptions.append(InvalidMetadata(unparsed_key, message))
|
||||
|
||||
if exceptions:
|
||||
raise ExceptionGroup("unparsed", exceptions)
|
||||
|
||||
try:
|
||||
return cls.from_raw(raw, validate=validate)
|
||||
except ExceptionGroup as exc_group:
|
||||
raise ExceptionGroup(
|
||||
"invalid or unparsed metadata", exc_group.exceptions
|
||||
) from None
|
||||
|
||||
metadata_version: _Validator[_MetadataVersion] = _Validator()
|
||||
""":external:ref:`core-metadata-metadata-version`
|
||||
(required; validated to be a valid metadata version)"""
|
||||
# `name` is not normalized/typed to NormalizedName so as to provide access to
|
||||
# the original/raw name.
|
||||
name: _Validator[str] = _Validator()
|
||||
""":external:ref:`core-metadata-name`
|
||||
(required; validated using :func:`~packaging.utils.canonicalize_name` and its
|
||||
*validate* parameter)"""
|
||||
version: _Validator[version_module.Version] = _Validator()
|
||||
""":external:ref:`core-metadata-version` (required)"""
|
||||
dynamic: _Validator[list[str] | None] = _Validator(
|
||||
added="2.2",
|
||||
)
|
||||
""":external:ref:`core-metadata-dynamic`
|
||||
(validated against core metadata field names and lowercased)"""
|
||||
platforms: _Validator[list[str] | None] = _Validator()
|
||||
""":external:ref:`core-metadata-platform`"""
|
||||
supported_platforms: _Validator[list[str] | None] = _Validator(added="1.1")
|
||||
""":external:ref:`core-metadata-supported-platform`"""
|
||||
summary: _Validator[str | None] = _Validator()
|
||||
""":external:ref:`core-metadata-summary` (validated to contain no newlines)"""
|
||||
description: _Validator[str | None] = _Validator() # TODO 2.1: can be in body
|
||||
""":external:ref:`core-metadata-description`"""
|
||||
description_content_type: _Validator[str | None] = _Validator(added="2.1")
|
||||
""":external:ref:`core-metadata-description-content-type` (validated)"""
|
||||
keywords: _Validator[list[str] | None] = _Validator()
|
||||
""":external:ref:`core-metadata-keywords`"""
|
||||
home_page: _Validator[str | None] = _Validator()
|
||||
""":external:ref:`core-metadata-home-page`"""
|
||||
download_url: _Validator[str | None] = _Validator(added="1.1")
|
||||
""":external:ref:`core-metadata-download-url`"""
|
||||
author: _Validator[str | None] = _Validator()
|
||||
""":external:ref:`core-metadata-author`"""
|
||||
author_email: _Validator[str | None] = _Validator()
|
||||
""":external:ref:`core-metadata-author-email`"""
|
||||
maintainer: _Validator[str | None] = _Validator(added="1.2")
|
||||
""":external:ref:`core-metadata-maintainer`"""
|
||||
maintainer_email: _Validator[str | None] = _Validator(added="1.2")
|
||||
""":external:ref:`core-metadata-maintainer-email`"""
|
||||
license: _Validator[str | None] = _Validator()
|
||||
""":external:ref:`core-metadata-license`"""
|
||||
license_expression: _Validator[NormalizedLicenseExpression | None] = _Validator(
|
||||
added="2.4"
|
||||
)
|
||||
""":external:ref:`core-metadata-license-expression`"""
|
||||
license_files: _Validator[list[str] | None] = _Validator(added="2.4")
|
||||
""":external:ref:`core-metadata-license-file`"""
|
||||
classifiers: _Validator[list[str] | None] = _Validator(added="1.1")
|
||||
""":external:ref:`core-metadata-classifier`"""
|
||||
requires_dist: _Validator[list[requirements.Requirement] | None] = _Validator(
|
||||
added="1.2"
|
||||
)
|
||||
""":external:ref:`core-metadata-requires-dist`"""
|
||||
requires_python: _Validator[specifiers.SpecifierSet | None] = _Validator(
|
||||
added="1.2"
|
||||
)
|
||||
""":external:ref:`core-metadata-requires-python`"""
|
||||
# Because `Requires-External` allows for non-PEP 440 version specifiers, we
|
||||
# don't do any processing on the values.
|
||||
requires_external: _Validator[list[str] | None] = _Validator(added="1.2")
|
||||
""":external:ref:`core-metadata-requires-external`"""
|
||||
project_urls: _Validator[dict[str, str] | None] = _Validator(added="1.2")
|
||||
""":external:ref:`core-metadata-project-url`"""
|
||||
# PEP 685 lets us raise an error if an extra doesn't pass `Name` validation
|
||||
# regardless of metadata version.
|
||||
provides_extra: _Validator[list[utils.NormalizedName] | None] = _Validator(
|
||||
added="2.1",
|
||||
)
|
||||
""":external:ref:`core-metadata-provides-extra`"""
|
||||
provides_dist: _Validator[list[str] | None] = _Validator(added="1.2")
|
||||
""":external:ref:`core-metadata-provides-dist`"""
|
||||
obsoletes_dist: _Validator[list[str] | None] = _Validator(added="1.2")
|
||||
""":external:ref:`core-metadata-obsoletes-dist`"""
|
||||
import_names: _Validator[list[str] | None] = _Validator(added="2.5")
|
||||
""":external:ref:`core-metadata-import-name`"""
|
||||
import_namespaces: _Validator[list[str] | None] = _Validator(added="2.5")
|
||||
""":external:ref:`core-metadata-import-namespace`"""
|
||||
requires: _Validator[list[str] | None] = _Validator(added="1.1")
|
||||
"""``Requires`` (deprecated)"""
|
||||
provides: _Validator[list[str] | None] = _Validator(added="1.1")
|
||||
"""``Provides`` (deprecated)"""
|
||||
obsoletes: _Validator[list[str] | None] = _Validator(added="1.1")
|
||||
"""``Obsoletes`` (deprecated)"""
|
||||
|
||||
def as_rfc822(self) -> RFC822Message:
|
||||
"""
|
||||
Return an RFC822 message with the metadata.
|
||||
"""
|
||||
message = RFC822Message()
|
||||
self._write_metadata(message)
|
||||
return message
|
||||
|
||||
def _write_metadata(self, message: RFC822Message) -> None:
|
||||
"""
|
||||
Return an RFC822 message with the metadata.
|
||||
"""
|
||||
for name, validator in self.__class__.__dict__.items():
|
||||
if isinstance(validator, _Validator) and name != "description":
|
||||
value = getattr(self, name)
|
||||
email_name = _RAW_TO_EMAIL_MAPPING[name]
|
||||
if value is not None:
|
||||
if email_name == "project-url":
|
||||
for label, url in value.items():
|
||||
message[email_name] = f"{label}, {url}"
|
||||
elif email_name == "keywords":
|
||||
message[email_name] = ",".join(value)
|
||||
elif email_name == "import-name" and value == []:
|
||||
message[email_name] = ""
|
||||
elif isinstance(value, list):
|
||||
for item in value:
|
||||
message[email_name] = str(item)
|
||||
else:
|
||||
message[email_name] = str(value)
|
||||
|
||||
# The description is a special case because it is in the body of the message.
|
||||
if self.description is not None:
|
||||
message.set_payload(self.description)
|
||||
BIN
venv/Lib/site-packages/pip/_vendor/packaking/py.typed
Normal file
BIN
venv/Lib/site-packages/pip/_vendor/packaking/py.typed
Normal file
Binary file not shown.
635
venv/Lib/site-packages/pip/_vendor/packaking/pylock.py
Normal file
635
venv/Lib/site-packages/pip/_vendor/packaking/pylock.py
Normal file
@@ -0,0 +1,635 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import dataclasses
|
||||
import logging
|
||||
import re
|
||||
from collections.abc import Mapping, Sequence
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
from typing import (
|
||||
TYPE_CHECKING,
|
||||
Any,
|
||||
Callable,
|
||||
Protocol,
|
||||
TypeVar,
|
||||
)
|
||||
|
||||
from .markers import Marker
|
||||
from .specifiers import SpecifierSet
|
||||
from .utils import NormalizedName, is_normalized_name
|
||||
from .version import Version
|
||||
|
||||
if TYPE_CHECKING: # pragma: no cover
|
||||
from pathlib import Path
|
||||
|
||||
from typing_extensions import Self
|
||||
|
||||
_logger = logging.getLogger(__name__)
|
||||
|
||||
__all__ = [
|
||||
"Package",
|
||||
"PackageArchive",
|
||||
"PackageDirectory",
|
||||
"PackageSdist",
|
||||
"PackageVcs",
|
||||
"PackageWheel",
|
||||
"Pylock",
|
||||
"PylockUnsupportedVersionError",
|
||||
"PylockValidationError",
|
||||
"is_valid_pylock_path",
|
||||
]
|
||||
|
||||
_T = TypeVar("_T")
|
||||
_T2 = TypeVar("_T2")
|
||||
|
||||
|
||||
class _FromMappingProtocol(Protocol): # pragma: no cover
|
||||
@classmethod
|
||||
def _from_dict(cls, d: Mapping[str, Any]) -> Self: ...
|
||||
|
||||
|
||||
_FromMappingProtocolT = TypeVar("_FromMappingProtocolT", bound=_FromMappingProtocol)
|
||||
|
||||
|
||||
_PYLOCK_FILE_NAME_RE = re.compile(r"^pylock\.([^.]+)\.toml$")
|
||||
|
||||
|
||||
def is_valid_pylock_path(path: Path) -> bool:
|
||||
"""Check if the given path is a valid pylock file path."""
|
||||
return path.name == "pylock.toml" or bool(_PYLOCK_FILE_NAME_RE.match(path.name))
|
||||
|
||||
|
||||
def _toml_key(key: str) -> str:
|
||||
return key.replace("_", "-")
|
||||
|
||||
|
||||
def _toml_value(key: str, value: Any) -> Any: # noqa: ANN401
|
||||
if isinstance(value, (Version, Marker, SpecifierSet)):
|
||||
return str(value)
|
||||
if isinstance(value, Sequence) and key == "environments":
|
||||
return [str(v) for v in value]
|
||||
return value
|
||||
|
||||
|
||||
def _toml_dict_factory(data: list[tuple[str, Any]]) -> dict[str, Any]:
|
||||
return {
|
||||
_toml_key(key): _toml_value(key, value)
|
||||
for key, value in data
|
||||
if value is not None
|
||||
}
|
||||
|
||||
|
||||
def _get(d: Mapping[str, Any], expected_type: type[_T], key: str) -> _T | None:
|
||||
"""Get a value from the dictionary and verify it's the expected type."""
|
||||
if (value := d.get(key)) is None:
|
||||
return None
|
||||
if not isinstance(value, expected_type):
|
||||
raise PylockValidationError(
|
||||
f"Unexpected type {type(value).__name__} "
|
||||
f"(expected {expected_type.__name__})",
|
||||
context=key,
|
||||
)
|
||||
return value
|
||||
|
||||
|
||||
def _get_required(d: Mapping[str, Any], expected_type: type[_T], key: str) -> _T:
|
||||
"""Get a required value from the dictionary and verify it's the expected type."""
|
||||
if (value := _get(d, expected_type, key)) is None:
|
||||
raise _PylockRequiredKeyError(key)
|
||||
return value
|
||||
|
||||
|
||||
def _get_sequence(
|
||||
d: Mapping[str, Any], expected_item_type: type[_T], key: str
|
||||
) -> Sequence[_T] | None:
|
||||
"""Get a list value from the dictionary and verify it's the expected items type."""
|
||||
if (value := _get(d, Sequence, key)) is None: # type: ignore[type-abstract]
|
||||
return None
|
||||
if isinstance(value, (str, bytes)):
|
||||
# special case: str and bytes are Sequences, but we want to reject it
|
||||
raise PylockValidationError(
|
||||
f"Unexpected type {type(value).__name__} (expected Sequence)",
|
||||
context=key,
|
||||
)
|
||||
for i, item in enumerate(value):
|
||||
if not isinstance(item, expected_item_type):
|
||||
raise PylockValidationError(
|
||||
f"Unexpected type {type(item).__name__} "
|
||||
f"(expected {expected_item_type.__name__})",
|
||||
context=f"{key}[{i}]",
|
||||
)
|
||||
return value
|
||||
|
||||
|
||||
def _get_as(
|
||||
d: Mapping[str, Any],
|
||||
expected_type: type[_T],
|
||||
target_type: Callable[[_T], _T2],
|
||||
key: str,
|
||||
) -> _T2 | None:
|
||||
"""Get a value from the dictionary, verify it's the expected type,
|
||||
and convert to the target type.
|
||||
|
||||
This assumes the target_type constructor accepts the value.
|
||||
"""
|
||||
if (value := _get(d, expected_type, key)) is None:
|
||||
return None
|
||||
try:
|
||||
return target_type(value)
|
||||
except Exception as e:
|
||||
raise PylockValidationError(e, context=key) from e
|
||||
|
||||
|
||||
def _get_required_as(
|
||||
d: Mapping[str, Any],
|
||||
expected_type: type[_T],
|
||||
target_type: Callable[[_T], _T2],
|
||||
key: str,
|
||||
) -> _T2:
|
||||
"""Get a required value from the dict, verify it's the expected type,
|
||||
and convert to the target type."""
|
||||
if (value := _get_as(d, expected_type, target_type, key)) is None:
|
||||
raise _PylockRequiredKeyError(key)
|
||||
return value
|
||||
|
||||
|
||||
def _get_sequence_as(
|
||||
d: Mapping[str, Any],
|
||||
expected_item_type: type[_T],
|
||||
target_item_type: Callable[[_T], _T2],
|
||||
key: str,
|
||||
) -> list[_T2] | None:
|
||||
"""Get list value from dictionary and verify expected items type."""
|
||||
if (value := _get_sequence(d, expected_item_type, key)) is None:
|
||||
return None
|
||||
result = []
|
||||
try:
|
||||
for item in value:
|
||||
typed_item = target_item_type(item)
|
||||
result.append(typed_item)
|
||||
except Exception as e:
|
||||
raise PylockValidationError(e, context=f"{key}[{len(result)}]") from e
|
||||
return result
|
||||
|
||||
|
||||
def _get_object(
|
||||
d: Mapping[str, Any], target_type: type[_FromMappingProtocolT], key: str
|
||||
) -> _FromMappingProtocolT | None:
|
||||
"""Get a dictionary value from the dictionary and convert it to a dataclass."""
|
||||
if (value := _get(d, Mapping, key)) is None: # type: ignore[type-abstract]
|
||||
return None
|
||||
try:
|
||||
return target_type._from_dict(value)
|
||||
except Exception as e:
|
||||
raise PylockValidationError(e, context=key) from e
|
||||
|
||||
|
||||
def _get_sequence_of_objects(
|
||||
d: Mapping[str, Any], target_item_type: type[_FromMappingProtocolT], key: str
|
||||
) -> list[_FromMappingProtocolT] | None:
|
||||
"""Get a list value from the dictionary and convert its items to a dataclass."""
|
||||
if (value := _get_sequence(d, Mapping, key)) is None: # type: ignore[type-abstract]
|
||||
return None
|
||||
result: list[_FromMappingProtocolT] = []
|
||||
try:
|
||||
for item in value:
|
||||
typed_item = target_item_type._from_dict(item)
|
||||
result.append(typed_item)
|
||||
except Exception as e:
|
||||
raise PylockValidationError(e, context=f"{key}[{len(result)}]") from e
|
||||
return result
|
||||
|
||||
|
||||
def _get_required_sequence_of_objects(
|
||||
d: Mapping[str, Any], target_item_type: type[_FromMappingProtocolT], key: str
|
||||
) -> Sequence[_FromMappingProtocolT]:
|
||||
"""Get a required list value from the dictionary and convert its items to a
|
||||
dataclass."""
|
||||
if (result := _get_sequence_of_objects(d, target_item_type, key)) is None:
|
||||
raise _PylockRequiredKeyError(key)
|
||||
return result
|
||||
|
||||
|
||||
def _validate_normalized_name(name: str) -> NormalizedName:
|
||||
"""Validate that a string is a NormalizedName."""
|
||||
if not is_normalized_name(name):
|
||||
raise PylockValidationError(f"Name {name!r} is not normalized")
|
||||
return NormalizedName(name)
|
||||
|
||||
|
||||
def _validate_path_url(path: str | None, url: str | None) -> None:
|
||||
if not path and not url:
|
||||
raise PylockValidationError("path or url must be provided")
|
||||
|
||||
|
||||
def _validate_hashes(hashes: Mapping[str, Any]) -> Mapping[str, Any]:
|
||||
if not hashes:
|
||||
raise PylockValidationError("At least one hash must be provided")
|
||||
if not all(isinstance(hash_val, str) for hash_val in hashes.values()):
|
||||
raise PylockValidationError("Hash values must be strings")
|
||||
return hashes
|
||||
|
||||
|
||||
class PylockValidationError(Exception):
|
||||
"""Raised when when input data is not spec-compliant."""
|
||||
|
||||
context: str | None = None
|
||||
message: str
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
cause: str | Exception,
|
||||
*,
|
||||
context: str | None = None,
|
||||
) -> None:
|
||||
if isinstance(cause, PylockValidationError):
|
||||
if cause.context:
|
||||
self.context = (
|
||||
f"{context}.{cause.context}" if context else cause.context
|
||||
)
|
||||
else:
|
||||
self.context = context
|
||||
self.message = cause.message
|
||||
else:
|
||||
self.context = context
|
||||
self.message = str(cause)
|
||||
|
||||
def __str__(self) -> str:
|
||||
if self.context:
|
||||
return f"{self.message} in {self.context!r}"
|
||||
return self.message
|
||||
|
||||
|
||||
class _PylockRequiredKeyError(PylockValidationError):
|
||||
def __init__(self, key: str) -> None:
|
||||
super().__init__("Missing required value", context=key)
|
||||
|
||||
|
||||
class PylockUnsupportedVersionError(PylockValidationError):
|
||||
"""Raised when encountering an unsupported `lock_version`."""
|
||||
|
||||
|
||||
@dataclass(frozen=True, init=False)
|
||||
class PackageVcs:
|
||||
type: str
|
||||
url: str | None = None
|
||||
path: str | None = None
|
||||
requested_revision: str | None = None
|
||||
commit_id: str # type: ignore[misc]
|
||||
subdirectory: str | None = None
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
type: str,
|
||||
url: str | None = None,
|
||||
path: str | None = None,
|
||||
requested_revision: str | None = None,
|
||||
commit_id: str,
|
||||
subdirectory: str | None = None,
|
||||
) -> None:
|
||||
# In Python 3.10+ make dataclass kw_only=True and remove __init__
|
||||
object.__setattr__(self, "type", type)
|
||||
object.__setattr__(self, "url", url)
|
||||
object.__setattr__(self, "path", path)
|
||||
object.__setattr__(self, "requested_revision", requested_revision)
|
||||
object.__setattr__(self, "commit_id", commit_id)
|
||||
object.__setattr__(self, "subdirectory", subdirectory)
|
||||
|
||||
@classmethod
|
||||
def _from_dict(cls, d: Mapping[str, Any]) -> Self:
|
||||
package_vcs = cls(
|
||||
type=_get_required(d, str, "type"),
|
||||
url=_get(d, str, "url"),
|
||||
path=_get(d, str, "path"),
|
||||
requested_revision=_get(d, str, "requested-revision"),
|
||||
commit_id=_get_required(d, str, "commit-id"),
|
||||
subdirectory=_get(d, str, "subdirectory"),
|
||||
)
|
||||
_validate_path_url(package_vcs.path, package_vcs.url)
|
||||
return package_vcs
|
||||
|
||||
|
||||
@dataclass(frozen=True, init=False)
|
||||
class PackageDirectory:
|
||||
path: str
|
||||
editable: bool | None = None
|
||||
subdirectory: str | None = None
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
path: str,
|
||||
editable: bool | None = None,
|
||||
subdirectory: str | None = None,
|
||||
) -> None:
|
||||
# In Python 3.10+ make dataclass kw_only=True and remove __init__
|
||||
object.__setattr__(self, "path", path)
|
||||
object.__setattr__(self, "editable", editable)
|
||||
object.__setattr__(self, "subdirectory", subdirectory)
|
||||
|
||||
@classmethod
|
||||
def _from_dict(cls, d: Mapping[str, Any]) -> Self:
|
||||
return cls(
|
||||
path=_get_required(d, str, "path"),
|
||||
editable=_get(d, bool, "editable"),
|
||||
subdirectory=_get(d, str, "subdirectory"),
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, init=False)
|
||||
class PackageArchive:
|
||||
url: str | None = None
|
||||
path: str | None = None
|
||||
size: int | None = None
|
||||
upload_time: datetime | None = None
|
||||
hashes: Mapping[str, str] # type: ignore[misc]
|
||||
subdirectory: str | None = None
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
url: str | None = None,
|
||||
path: str | None = None,
|
||||
size: int | None = None,
|
||||
upload_time: datetime | None = None,
|
||||
hashes: Mapping[str, str],
|
||||
subdirectory: str | None = None,
|
||||
) -> None:
|
||||
# In Python 3.10+ make dataclass kw_only=True and remove __init__
|
||||
object.__setattr__(self, "url", url)
|
||||
object.__setattr__(self, "path", path)
|
||||
object.__setattr__(self, "size", size)
|
||||
object.__setattr__(self, "upload_time", upload_time)
|
||||
object.__setattr__(self, "hashes", hashes)
|
||||
object.__setattr__(self, "subdirectory", subdirectory)
|
||||
|
||||
@classmethod
|
||||
def _from_dict(cls, d: Mapping[str, Any]) -> Self:
|
||||
package_archive = cls(
|
||||
url=_get(d, str, "url"),
|
||||
path=_get(d, str, "path"),
|
||||
size=_get(d, int, "size"),
|
||||
upload_time=_get(d, datetime, "upload-time"),
|
||||
hashes=_get_required_as(d, Mapping, _validate_hashes, "hashes"), # type: ignore[type-abstract]
|
||||
subdirectory=_get(d, str, "subdirectory"),
|
||||
)
|
||||
_validate_path_url(package_archive.path, package_archive.url)
|
||||
return package_archive
|
||||
|
||||
|
||||
@dataclass(frozen=True, init=False)
|
||||
class PackageSdist:
|
||||
name: str | None = None
|
||||
upload_time: datetime | None = None
|
||||
url: str | None = None
|
||||
path: str | None = None
|
||||
size: int | None = None
|
||||
hashes: Mapping[str, str] # type: ignore[misc]
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
name: str | None = None,
|
||||
upload_time: datetime | None = None,
|
||||
url: str | None = None,
|
||||
path: str | None = None,
|
||||
size: int | None = None,
|
||||
hashes: Mapping[str, str],
|
||||
) -> None:
|
||||
# In Python 3.10+ make dataclass kw_only=True and remove __init__
|
||||
object.__setattr__(self, "name", name)
|
||||
object.__setattr__(self, "upload_time", upload_time)
|
||||
object.__setattr__(self, "url", url)
|
||||
object.__setattr__(self, "path", path)
|
||||
object.__setattr__(self, "size", size)
|
||||
object.__setattr__(self, "hashes", hashes)
|
||||
|
||||
@classmethod
|
||||
def _from_dict(cls, d: Mapping[str, Any]) -> Self:
|
||||
package_sdist = cls(
|
||||
name=_get(d, str, "name"),
|
||||
upload_time=_get(d, datetime, "upload-time"),
|
||||
url=_get(d, str, "url"),
|
||||
path=_get(d, str, "path"),
|
||||
size=_get(d, int, "size"),
|
||||
hashes=_get_required_as(d, Mapping, _validate_hashes, "hashes"), # type: ignore[type-abstract]
|
||||
)
|
||||
_validate_path_url(package_sdist.path, package_sdist.url)
|
||||
return package_sdist
|
||||
|
||||
|
||||
@dataclass(frozen=True, init=False)
|
||||
class PackageWheel:
|
||||
name: str | None = None
|
||||
upload_time: datetime | None = None
|
||||
url: str | None = None
|
||||
path: str | None = None
|
||||
size: int | None = None
|
||||
hashes: Mapping[str, str] # type: ignore[misc]
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
name: str | None = None,
|
||||
upload_time: datetime | None = None,
|
||||
url: str | None = None,
|
||||
path: str | None = None,
|
||||
size: int | None = None,
|
||||
hashes: Mapping[str, str],
|
||||
) -> None:
|
||||
# In Python 3.10+ make dataclass kw_only=True and remove __init__
|
||||
object.__setattr__(self, "name", name)
|
||||
object.__setattr__(self, "upload_time", upload_time)
|
||||
object.__setattr__(self, "url", url)
|
||||
object.__setattr__(self, "path", path)
|
||||
object.__setattr__(self, "size", size)
|
||||
object.__setattr__(self, "hashes", hashes)
|
||||
|
||||
@classmethod
|
||||
def _from_dict(cls, d: Mapping[str, Any]) -> Self:
|
||||
package_wheel = cls(
|
||||
name=_get(d, str, "name"),
|
||||
upload_time=_get(d, datetime, "upload-time"),
|
||||
url=_get(d, str, "url"),
|
||||
path=_get(d, str, "path"),
|
||||
size=_get(d, int, "size"),
|
||||
hashes=_get_required_as(d, Mapping, _validate_hashes, "hashes"), # type: ignore[type-abstract]
|
||||
)
|
||||
_validate_path_url(package_wheel.path, package_wheel.url)
|
||||
return package_wheel
|
||||
|
||||
|
||||
@dataclass(frozen=True, init=False)
|
||||
class Package:
|
||||
name: NormalizedName
|
||||
version: Version | None = None
|
||||
marker: Marker | None = None
|
||||
requires_python: SpecifierSet | None = None
|
||||
dependencies: Sequence[Mapping[str, Any]] | None = None
|
||||
vcs: PackageVcs | None = None
|
||||
directory: PackageDirectory | None = None
|
||||
archive: PackageArchive | None = None
|
||||
index: str | None = None
|
||||
sdist: PackageSdist | None = None
|
||||
wheels: Sequence[PackageWheel] | None = None
|
||||
attestation_identities: Sequence[Mapping[str, Any]] | None = None
|
||||
tool: Mapping[str, Any] | None = None
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
name: NormalizedName,
|
||||
version: Version | None = None,
|
||||
marker: Marker | None = None,
|
||||
requires_python: SpecifierSet | None = None,
|
||||
dependencies: Sequence[Mapping[str, Any]] | None = None,
|
||||
vcs: PackageVcs | None = None,
|
||||
directory: PackageDirectory | None = None,
|
||||
archive: PackageArchive | None = None,
|
||||
index: str | None = None,
|
||||
sdist: PackageSdist | None = None,
|
||||
wheels: Sequence[PackageWheel] | None = None,
|
||||
attestation_identities: Sequence[Mapping[str, Any]] | None = None,
|
||||
tool: Mapping[str, Any] | None = None,
|
||||
) -> None:
|
||||
# In Python 3.10+ make dataclass kw_only=True and remove __init__
|
||||
object.__setattr__(self, "name", name)
|
||||
object.__setattr__(self, "version", version)
|
||||
object.__setattr__(self, "marker", marker)
|
||||
object.__setattr__(self, "requires_python", requires_python)
|
||||
object.__setattr__(self, "dependencies", dependencies)
|
||||
object.__setattr__(self, "vcs", vcs)
|
||||
object.__setattr__(self, "directory", directory)
|
||||
object.__setattr__(self, "archive", archive)
|
||||
object.__setattr__(self, "index", index)
|
||||
object.__setattr__(self, "sdist", sdist)
|
||||
object.__setattr__(self, "wheels", wheels)
|
||||
object.__setattr__(self, "attestation_identities", attestation_identities)
|
||||
object.__setattr__(self, "tool", tool)
|
||||
|
||||
@classmethod
|
||||
def _from_dict(cls, d: Mapping[str, Any]) -> Self:
|
||||
package = cls(
|
||||
name=_get_required_as(d, str, _validate_normalized_name, "name"),
|
||||
version=_get_as(d, str, Version, "version"),
|
||||
requires_python=_get_as(d, str, SpecifierSet, "requires-python"),
|
||||
dependencies=_get_sequence(d, Mapping, "dependencies"), # type: ignore[type-abstract]
|
||||
marker=_get_as(d, str, Marker, "marker"),
|
||||
vcs=_get_object(d, PackageVcs, "vcs"),
|
||||
directory=_get_object(d, PackageDirectory, "directory"),
|
||||
archive=_get_object(d, PackageArchive, "archive"),
|
||||
index=_get(d, str, "index"),
|
||||
sdist=_get_object(d, PackageSdist, "sdist"),
|
||||
wheels=_get_sequence_of_objects(d, PackageWheel, "wheels"),
|
||||
attestation_identities=_get_sequence(d, Mapping, "attestation-identities"), # type: ignore[type-abstract]
|
||||
tool=_get(d, Mapping, "tool"), # type: ignore[type-abstract]
|
||||
)
|
||||
distributions = bool(package.sdist) + len(package.wheels or [])
|
||||
direct_urls = (
|
||||
bool(package.vcs) + bool(package.directory) + bool(package.archive)
|
||||
)
|
||||
if distributions > 0 and direct_urls > 0:
|
||||
raise PylockValidationError(
|
||||
"None of vcs, directory, archive must be set if sdist or wheels are set"
|
||||
)
|
||||
if distributions == 0 and direct_urls != 1:
|
||||
raise PylockValidationError(
|
||||
"Exactly one of vcs, directory, archive must be set "
|
||||
"if sdist and wheels are not set"
|
||||
)
|
||||
try:
|
||||
for i, attestation_identity in enumerate( # noqa: B007
|
||||
package.attestation_identities or []
|
||||
):
|
||||
_get_required(attestation_identity, str, "kind")
|
||||
except Exception as e:
|
||||
raise PylockValidationError(
|
||||
e, context=f"attestation-identities[{i}]"
|
||||
) from e
|
||||
return package
|
||||
|
||||
@property
|
||||
def is_direct(self) -> bool:
|
||||
return not (self.sdist or self.wheels)
|
||||
|
||||
|
||||
@dataclass(frozen=True, init=False)
|
||||
class Pylock:
|
||||
"""A class representing a pylock file."""
|
||||
|
||||
lock_version: Version
|
||||
environments: Sequence[Marker] | None = None
|
||||
requires_python: SpecifierSet | None = None
|
||||
extras: Sequence[NormalizedName] | None = None
|
||||
dependency_groups: Sequence[str] | None = None
|
||||
default_groups: Sequence[str] | None = None
|
||||
created_by: str # type: ignore[misc]
|
||||
packages: Sequence[Package] # type: ignore[misc]
|
||||
tool: Mapping[str, Any] | None = None
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
lock_version: Version,
|
||||
environments: Sequence[Marker] | None = None,
|
||||
requires_python: SpecifierSet | None = None,
|
||||
extras: Sequence[NormalizedName] | None = None,
|
||||
dependency_groups: Sequence[str] | None = None,
|
||||
default_groups: Sequence[str] | None = None,
|
||||
created_by: str,
|
||||
packages: Sequence[Package],
|
||||
tool: Mapping[str, Any] | None = None,
|
||||
) -> None:
|
||||
# In Python 3.10+ make dataclass kw_only=True and remove __init__
|
||||
object.__setattr__(self, "lock_version", lock_version)
|
||||
object.__setattr__(self, "environments", environments)
|
||||
object.__setattr__(self, "requires_python", requires_python)
|
||||
object.__setattr__(self, "extras", extras)
|
||||
object.__setattr__(self, "dependency_groups", dependency_groups)
|
||||
object.__setattr__(self, "default_groups", default_groups)
|
||||
object.__setattr__(self, "created_by", created_by)
|
||||
object.__setattr__(self, "packages", packages)
|
||||
object.__setattr__(self, "tool", tool)
|
||||
|
||||
@classmethod
|
||||
def _from_dict(cls, d: Mapping[str, Any]) -> Self:
|
||||
pylock = cls(
|
||||
lock_version=_get_required_as(d, str, Version, "lock-version"),
|
||||
environments=_get_sequence_as(d, str, Marker, "environments"),
|
||||
extras=_get_sequence_as(d, str, _validate_normalized_name, "extras"),
|
||||
dependency_groups=_get_sequence(d, str, "dependency-groups"),
|
||||
default_groups=_get_sequence(d, str, "default-groups"),
|
||||
created_by=_get_required(d, str, "created-by"),
|
||||
requires_python=_get_as(d, str, SpecifierSet, "requires-python"),
|
||||
packages=_get_required_sequence_of_objects(d, Package, "packages"),
|
||||
tool=_get(d, Mapping, "tool"), # type: ignore[type-abstract]
|
||||
)
|
||||
if not Version("1") <= pylock.lock_version < Version("2"):
|
||||
raise PylockUnsupportedVersionError(
|
||||
f"pylock version {pylock.lock_version} is not supported"
|
||||
)
|
||||
if pylock.lock_version > Version("1.0"):
|
||||
_logger.warning(
|
||||
"pylock minor version %s is not supported", pylock.lock_version
|
||||
)
|
||||
return pylock
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, d: Mapping[str, Any], /) -> Self:
|
||||
"""Create and validate a Pylock instance from a TOML dictionary.
|
||||
|
||||
Raises :class:`PylockValidationError` if the input data is not
|
||||
spec-compliant.
|
||||
"""
|
||||
return cls._from_dict(d)
|
||||
|
||||
def to_dict(self) -> Mapping[str, Any]:
|
||||
"""Convert the Pylock instance to a TOML dictionary."""
|
||||
return dataclasses.asdict(self, dict_factory=_toml_dict_factory)
|
||||
|
||||
def validate(self) -> None:
|
||||
"""Validate the Pylock instance against the specification.
|
||||
|
||||
Raises :class:`PylockValidationError` otherwise."""
|
||||
self.from_dict(self.to_dict())
|
||||
86
venv/Lib/site-packages/pip/_vendor/packaking/requirements.py
Normal file
86
venv/Lib/site-packages/pip/_vendor/packaking/requirements.py
Normal file
@@ -0,0 +1,86 @@
|
||||
# This file is dual licensed under the terms of the Apache License, Version
|
||||
# 2.0, and the BSD License. See the LICENSE file in the root of this repository
|
||||
# for complete details.
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Iterator
|
||||
|
||||
from ._parser import parse_requirement as _parse_requirement
|
||||
from ._tokenizer import ParserSyntaxError
|
||||
from .markers import Marker, _normalize_extra_values
|
||||
from .specifiers import SpecifierSet
|
||||
from .utils import canonicalize_name
|
||||
|
||||
|
||||
class InvalidRequirement(ValueError):
|
||||
"""
|
||||
An invalid requirement was found, users should refer to PEP 508.
|
||||
"""
|
||||
|
||||
|
||||
class Requirement:
|
||||
"""Parse a requirement.
|
||||
|
||||
Parse a given requirement string into its parts, such as name, specifier,
|
||||
URL, and extras. Raises InvalidRequirement on a badly-formed requirement
|
||||
string.
|
||||
"""
|
||||
|
||||
# TODO: Can we test whether something is contained within a requirement?
|
||||
# If so how do we do that? Do we need to test against the _name_ of
|
||||
# the thing as well as the version? What about the markers?
|
||||
# TODO: Can we normalize the name and extra name?
|
||||
|
||||
def __init__(self, requirement_string: str) -> None:
|
||||
try:
|
||||
parsed = _parse_requirement(requirement_string)
|
||||
except ParserSyntaxError as e:
|
||||
raise InvalidRequirement(str(e)) from e
|
||||
|
||||
self.name: str = parsed.name
|
||||
self.url: str | None = parsed.url or None
|
||||
self.extras: set[str] = set(parsed.extras or [])
|
||||
self.specifier: SpecifierSet = SpecifierSet(parsed.specifier)
|
||||
self.marker: Marker | None = None
|
||||
if parsed.marker is not None:
|
||||
self.marker = Marker.__new__(Marker)
|
||||
self.marker._markers = _normalize_extra_values(parsed.marker)
|
||||
|
||||
def _iter_parts(self, name: str) -> Iterator[str]:
|
||||
yield name
|
||||
|
||||
if self.extras:
|
||||
formatted_extras = ",".join(sorted(self.extras))
|
||||
yield f"[{formatted_extras}]"
|
||||
|
||||
if self.specifier:
|
||||
yield str(self.specifier)
|
||||
|
||||
if self.url:
|
||||
yield f" @ {self.url}"
|
||||
if self.marker:
|
||||
yield " "
|
||||
|
||||
if self.marker:
|
||||
yield f"; {self.marker}"
|
||||
|
||||
def __str__(self) -> str:
|
||||
return "".join(self._iter_parts(self.name))
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"<{self.__class__.__name__}('{self}')>"
|
||||
|
||||
def __hash__(self) -> int:
|
||||
return hash(tuple(self._iter_parts(canonicalize_name(self.name))))
|
||||
|
||||
def __eq__(self, other: object) -> bool:
|
||||
if not isinstance(other, Requirement):
|
||||
return NotImplemented
|
||||
|
||||
return (
|
||||
canonicalize_name(self.name) == canonicalize_name(other.name)
|
||||
and self.extras == other.extras
|
||||
and self.specifier == other.specifier
|
||||
and self.url == other.url
|
||||
and self.marker == other.marker
|
||||
)
|
||||
Reference in New Issue
Block a user