20332790708

Committed 18 Dec 2025 09:48AM UTC coverage: 64.992% (-15.3%) from 80.295%

Build # 20332790708

Build Type

Pull #22949

github

Committed by

web-flow

Commit Message

Merge f730a56cd into 407284c67

Pull Request Pull Request #22949: Add experimental uv resolver for Python lockfiles

Run Details

54 of 97 new or added lines in 5 files covered. (55.67%)

8270 existing lines in 295 files now uncovered.

48990 of 75379 relevant lines covered (64.99%)

1.81 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

30.17

/src/python/pants/backend/python/macros/poetry_requirements.py

# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).

from __future__ import annotations

import itertools
import logging
import urllib.parse
from collections.abc import Iterator, Mapping, Sequence
from dataclasses import dataclass
from functools import partial
from pathlib import Path, PurePath
from typing import Any, cast

import toml
from packaging.version import InvalidVersion, Version
from typing_extensions import TypedDict

from pants.backend.python.macros.common_fields import (
    ModuleMappingField,
    RequirementsOverrideField,
    TypeStubsModuleMappingField,
)
from pants.backend.python.macros.common_requirements_rule import _generate_requirements
from pants.backend.python.subsystems.setup import PythonSetup
from pants.backend.python.target_types import PythonRequirementResolveField, PythonRequirementTarget
from pants.base.build_root import BuildRoot
from pants.engine.rules import collect_rules, rule
from pants.engine.target import (
    COMMON_TARGET_FIELDS,
    GeneratedTargets,
    GenerateTargetsRequest,
    SingleSourceField,
    TargetGenerator,
)
from pants.engine.unions import UnionMembership, UnionRule
from pants.util.logging import LogLevel
from pants.util.pip_requirement import PipRequirement
from pants.util.strutil import softwrap

logger = logging.getLogger(__name__)

# ---------------------------------------------------------------------------------
# pyproject.toml parsing
# ---------------------------------------------------------------------------------


class PyprojectAttr(TypedDict, total=False):
    extras: list[str]
    git: str
    rev: str
    branch: str
    python: str
    markers: str
    tag: str
    version: str
    url: str
    path: str


def get_max_caret(parsed_version: Version) -> str:
    major = 0
    minor = 0
    micro = 0

    if parsed_version.major != 0:
        major = parsed_version.major + 1
    elif parsed_version.minor != 0:
        minor = parsed_version.minor + 1
    elif parsed_version.micro != 0:
        micro = parsed_version.micro + 1
    else:
        base_len = len(parsed_version.base_version.split("."))
        if base_len >= 3:
            micro = 1
        elif base_len == 2:
            minor = 1
        elif base_len == 1:
            major = 1

    return f"{major}.{minor}.{micro}"


def get_max_tilde(parsed_version: Version) -> str:
    major = 0
    minor = 0
    base_len = len(parsed_version.base_version.split("."))
    if base_len >= 2:
        minor = int(str(parsed_version.minor)) + 1
        major = int(str(parsed_version.major))
    elif base_len == 1:
        major = int(str(parsed_version.major)) + 1

    return f"{major}.{minor}.0"


def get_max_wildcard(parsed_version: Version) -> str:
    # Note: Assumes this is not a global wildcard, so parsed_version.release has
    # at least two components.
    release = list(parsed_version.release)
    release[-2] += 1
    major = release[0]
    minor = release[1]
    return f"{major}.{minor}.0"


def parse_str_version(attributes: str, **kwargs: str) -> str:
    valid_specifiers = "<>!~="
    pep440_reqs = []
    proj_name = kwargs["proj_name"]
    fp = kwargs["file_path"]
    extras_str = kwargs["extras_str"]
    comma_split_reqs = (i.strip() for i in attributes.split(","))
    for req in comma_split_reqs:

        def parse_version(version_str: str) -> Version:
            try:
                return Version(version_str)
            except InvalidVersion:
                raise InvalidVersion(
                    softwrap(
                        f"""
                        Failed to parse requirement {proj_name} = "{req}" in {fp} loaded by the
                        poetry_requirements macro.

                        If you believe this requirement is valid, consider opening an issue at
                        https://github.com/pantsbuild/pants/issues so that we can update Pants'
                        Poetry macro to support this.
                        """
                    )
                )

        if not req:
            continue
        if req[0] == "^":
            parsed_version = parse_version(req[1:])
            max_ver = get_max_caret(parsed_version)
            min_ver = f"{parsed_version.public}"
            pep440_reqs.append(f">={min_ver},<{max_ver}")
        elif req[0] == "~" and req[1] != "=":
            # ~= is an acceptable default operator; however, ~ is not, and IS NOT the same as ~=
            parsed_version = parse_version(req[1:])
            max_ver = get_max_tilde(parsed_version)
            min_ver = f"{parsed_version.public}"
            pep440_reqs.append(f">={min_ver},<{max_ver}")
        elif req[-1] == "*":
            if req != "*":  # This is not a global wildcard.
                # To parse we replace the * with a 0.
                parsed_version = parse_version(f"{req[:-1]}0")
                max_ver = get_max_wildcard(parsed_version)
                min_ver = f"{parsed_version.public}"
                pep440_reqs.append(f">={min_ver},<{max_ver}")
        else:
            pep440_reqs.append(req if req[0] in valid_specifiers else f"=={req}")
    return f"{proj_name}{extras_str} {','.join(pep440_reqs)}".rstrip()


def parse_python_constraint(constr: str | None, fp: str) -> str:
    if constr is None:
        return ""
    valid_specifiers = "<>!~= "
    # If the user passes multiple Python constraints, they're separated by
    # either '||' signifying a logical 'or', or a comma signifying a logical
    # 'and'. Hence, or_and_split is a 2D list where each inner list is a set of and-ed
    # requirements; every list in the second layer is then or-ed together.
    or_and_split = [[j.strip() for j in i.split(",")] for i in constr.split("||")]

    # We only use parse_str_version to address the version parsing; we don't
    # care about having an actual Requirement object so things like the project name
    # and extras that would ordinarily exist for a project with a string version are left blank here.
    ver_parsed = [
        [parse_str_version(j, proj_name="", file_path=fp, extras_str="") for j in i]
        for i in or_and_split
    ]

    def conv_and(lst: list[str]) -> list:
        return list(itertools.chain(*[i.split(",") for i in lst]))

    def prepend(version: str) -> str:
        valid_versions = "".join(i for i in version if i in valid_specifiers)
        invalid_versions = "".join(i for i in version if i not in valid_specifiers)
        return f"python_version{valid_versions} '{invalid_versions}'"

    prepend_and_clean = [
        [prepend(".".join(j.split(".")[:2])) for j in conv_and(i)] for i in ver_parsed
    ]
    return (
        ("(" if len(or_and_split) > 1 else "")
        + (") or (".join([" and ".join(i) for i in prepend_and_clean]))
        + (")" if len(or_and_split) > 1 else "")
    )


@dataclass(frozen=True)
class PyProjectToml:
    build_root: PurePath
    toml_relpath: PurePath
    toml_contents: str

    def parse(self) -> Mapping[str, Any]:
        return toml.loads(self.toml_contents)

    def _non_pants_project_abs_path(self, path: Path) -> Path | None:
        resolved = path.resolve()
        if resolved.is_file():
            return resolved

        try:
            resolved.relative_to(self.build_root)
        except ValueError:
            return resolved

        return None

    def non_pants_project_abs_path(self, path: str) -> Path | None:
        """Determine if the given path represents a non-Pants controlled project.

        If the path points to a file, it's assumed the file is a distribution ( a wheel or sdist)
        and the absolute path of that file is returned.

        If the path points to a directory and that directory is outside of the build root, it's
        assumed the directory is the root of a buildable Python project (i.e.: it contains a
        pyproject.toml or setup.py) and the absolute path of the project is returned.

        Otherwise, `None` is returned since the directory lies inside the build root and is assumed
        to be a Pants controlled project.
        """
        # TODO(John Sirois): This leaves the case where the path is a Python project directory
        #  inside the build root that the user actually wants Pex / Pip to build. A concrete case
        #  for this would be a repo where third party is partially handled with vendored exploded
        #  source distributions. If someone in the wild needs the described case, plumb a
        #  PoetryRequirements parameter that can list paths to treat as Pants controlled or
        #  vice-versa.
        given_path = Path(path)
        if given_path.is_absolute():
            return self._non_pants_project_abs_path(given_path)
        else:
            return self._non_pants_project_abs_path(
                Path(self.build_root / self.toml_relpath).parent / given_path
            )


def produce_match(sep: str, feat: Any) -> str:
    return f"{sep}{feat}" if feat else ""


def add_markers(base: str, attributes: PyprojectAttr, fp) -> str:
    markers_lookup = produce_match("", attributes.get("markers"))
    python_lookup = parse_python_constraint(attributes.get("python"), fp)

    # Python constraints are passed as a `python_version` environment marker; if we have multiple
    # markers, we evaluate them as one whole, and then AND with the new marker for the Python constraint.
    # E.g. (marker1 AND marker2 OR marker3...) AND (python_version)
    # rather than (marker1 AND marker2 OR marker3 AND python_version)
    if not markers_lookup and not python_lookup:
        return base

    result = f"{base} ;("

    if markers_lookup:
        result += f"{markers_lookup})"
    if python_lookup and markers_lookup:
        result += " and ("
    if python_lookup:
        result += f"{python_lookup})"

    return result


def handle_dict_attr(
    proj_name: str, attributes: PyprojectAttr, pyproject_toml: PyProjectToml
) -> str | None:
    base = ""
    fp = str(pyproject_toml.toml_relpath)

    extras_lookup = attributes.get("extras")
    if isinstance(extras_lookup, list):
        extras_str = f"[{','.join(extras_lookup)}]"
    else:
        extras_str = ""

    git_lookup = attributes.get("git")
    if git_lookup is not None:
        # If no URL scheme (e.g., `{git = "git@github.com:foo/bar.git"}`) we assume ssh,
        # i.e., we convert to git+ssh://git@github.com/foo/bar.git.
        if not urllib.parse.urlsplit(git_lookup).scheme:
            git_lookup = f"ssh://{git_lookup.replace(':', '/', 1)}"
        rev_lookup = produce_match("#", attributes.get("rev"))
        branch_lookup = produce_match("@", attributes.get("branch"))
        tag_lookup = produce_match("@", attributes.get("tag"))

        base = f"{proj_name}{extras_str} @ git+{git_lookup}{tag_lookup}{branch_lookup}{rev_lookup}"

    path_lookup = attributes.get("path")
    if path_lookup is not None:
        non_pants_project_abs_path = pyproject_toml.non_pants_project_abs_path(path_lookup)
        if non_pants_project_abs_path:
            base = f"{proj_name}{extras_str} @ file://{non_pants_project_abs_path}"
        else:
            # An internal path will be handled by normal Pants dependencies and dependency inference;
            # i.e.: it never represents a third party requirement.
            return None

    url_lookup = attributes.get("url")
    if url_lookup is not None:
        base = f"{proj_name}{extras_str} @ {url_lookup}"

    version_lookup = attributes.get("version")
    if version_lookup is not None:
        base = parse_str_version(
            version_lookup, file_path=fp, extras_str=extras_str, proj_name=proj_name
        )

    if len(base) == 0:
        raise ValueError(
            softwrap(
                f"""
                {proj_name} is not formatted correctly; at minimum provide either a version, url,
                path or git location for your dependency.
                """
            )
        )

    return add_markers(base, attributes, fp)


def parse_single_dependency(
    proj_name: str,
    attributes: str | Mapping[str, str | Sequence] | Sequence[Mapping[str, str | Sequence]],
    pyproject_toml: PyProjectToml,
) -> Iterator[PipRequirement]:
    if isinstance(attributes, str):
        # E.g. `foo = "~1.1~'.
        yield PipRequirement.parse(
            parse_str_version(
                attributes,
                proj_name=proj_name,
                file_path=str(pyproject_toml.toml_relpath),
                extras_str="",
            )
        )
    elif isinstance(attributes, dict):
        # E.g. `foo = {version = "~1.1"}`.
        pyproject_attr = cast(PyprojectAttr, attributes)
        req_str = handle_dict_attr(proj_name, pyproject_attr, pyproject_toml)
        if req_str:
            yield PipRequirement.parse(req_str)
    elif isinstance(attributes, list):
        # E.g. ` foo = [{version = "1.1","python" = "2.7"}, {version = "1.1","python" = "2.7"}]
        for attr in attributes:
            req_str = handle_dict_attr(proj_name, attr, pyproject_toml)
            if req_str:
                yield PipRequirement.parse(req_str)
    else:
        raise AssertionError(
            softwrap(
                f"""
                Error: invalid Poetry requirement format. Expected type of requirement attributes to
                be string, dict, or list, but was of type {type(attributes).__name__}.
                """
            )
        )


def parse_pyproject_toml(pyproject_toml: PyProjectToml) -> set[PipRequirement]:
    parsed = pyproject_toml.parse()
    try:
        poetry_vals = parsed["tool"]["poetry"]
    except KeyError:
        raise KeyError(
            softwrap(
                f"""
                No section `tool.poetry` found in {pyproject_toml.toml_relpath}, which
                is loaded by Pants from a `poetry_requirements` macro.

                Did you mean to set up Poetry?
                """
            )
        )
    dependencies = poetry_vals.get("dependencies", {})
    # N.B.: The "python" dependency is a special dependency required by Poetry that only serves to
    # constraint the python interpreter versions the project works with; so we skip that.
    # See: https://python-poetry.org/docs/pyproject/#dependencies-and-dev-dependencies
    dependencies.pop("python", None)

    groups = poetry_vals.get("group", {})
    group_deps: dict[str, PyprojectAttr] = {}

    for group in groups.values():
        group_deps.update(group.get("dependencies", {}))

    dev_dependencies = poetry_vals.get("dev-dependencies", {})
    if not dependencies and not dev_dependencies and not group_deps:
        logger.warning(
            softwrap(
                f"""
                No requirements defined in any Poetry dependency groups, tool.poetry.dependencies
                and tool.poetry.dev-dependencies in {pyproject_toml.toml_relpath}, which is loaded
                by Pants from a poetry_requirements macro. Did you mean to populate these
                with requirements?
                """
            )
        )

    return set(
        itertools.chain.from_iterable(
            parse_single_dependency(proj, attr, pyproject_toml)
            for proj, attr in {**dependencies, **dev_dependencies, **group_deps}.items()
        )
    )


def parse_poetry_requirements(
    build_root: BuildRoot, file_contents: bytes, file_path: str
) -> set[PipRequirement]:
    return parse_pyproject_toml(
        PyProjectToml(
            build_root=PurePath(build_root.path),
            toml_relpath=PurePath(file_path),
            toml_contents=file_contents.decode(),
        )
    )


# ---------------------------------------------------------------------------------
# Target generator
# ---------------------------------------------------------------------------------


class PoetryRequirementsSourceField(SingleSourceField):
    default = "pyproject.toml"
    required = False


class PoetryRequirementsTargetGenerator(TargetGenerator):
    alias = "poetry_requirements"
    help = "Generate a `python_requirement` for each entry in a Poetry `pyproject.toml`."
    generated_target_cls = PythonRequirementTarget
    # Note that this does not have a `dependencies` field.
    core_fields = (
        *COMMON_TARGET_FIELDS,
        ModuleMappingField,
        TypeStubsModuleMappingField,
        PoetryRequirementsSourceField,
        RequirementsOverrideField,
    )
    copied_fields = COMMON_TARGET_FIELDS
    moved_fields = (PythonRequirementResolveField,)


class GenerateFromPoetryRequirementsRequest(GenerateTargetsRequest):
    generate_from = PoetryRequirementsTargetGenerator


@rule(desc="Generate `python_requirement` targets from Poetry pyproject.toml", level=LogLevel.DEBUG)
async def generate_from_python_requirement(
    request: GenerateFromPoetryRequirementsRequest,
    build_root: BuildRoot,
    union_membership: UnionMembership,
    python_setup: PythonSetup,
) -> GeneratedTargets:
    result = await _generate_requirements(
        request,
        union_membership,
        python_setup,
        parse_requirements_callback=partial(parse_poetry_requirements, build_root),
    )
    return GeneratedTargets(request.generator, result)


def rules():
    return (
        *collect_rules(),
        UnionRule(GenerateTargetsRequest, GenerateFromPoetryRequirementsRequest),
    )

1	# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	1✔
5
6	import itertools	1✔
7	import logging	1✔
8	import urllib.parse	1✔
9	from collections.abc import Iterator, Mapping, Sequence	1✔
10	from dataclasses import dataclass	1✔
11	from functools import partial	1✔
12	from pathlib import Path, PurePath	1✔
13	from typing import Any, cast	1✔
14
15	import toml	1✔
16	from packaging.version import InvalidVersion, Version	1✔
17	from typing_extensions import TypedDict	1✔
18
19	from pants.backend.python.macros.common_fields import (	1✔
20	ModuleMappingField,
21	RequirementsOverrideField,
22	TypeStubsModuleMappingField,
23	)
24	from pants.backend.python.macros.common_requirements_rule import _generate_requirements	1✔
25	from pants.backend.python.subsystems.setup import PythonSetup	1✔
26	from pants.backend.python.target_types import PythonRequirementResolveField, PythonRequirementTarget	1✔
27	from pants.base.build_root import BuildRoot	1✔
28	from pants.engine.rules import collect_rules, rule	1✔
29	from pants.engine.target import (	1✔
30	COMMON_TARGET_FIELDS,
31	GeneratedTargets,
32	GenerateTargetsRequest,
33	SingleSourceField,
34	TargetGenerator,
35	)
36	from pants.engine.unions import UnionMembership, UnionRule	1✔
37	from pants.util.logging import LogLevel	1✔
38	from pants.util.pip_requirement import PipRequirement	1✔
39	from pants.util.strutil import softwrap	1✔
40
41	logger = logging.getLogger(__name__)	1✔
42
43	# ---------------------------------------------------------------------------------
44	# pyproject.toml parsing
45	# ---------------------------------------------------------------------------------
46
47
48	class PyprojectAttr(TypedDict, total=False):	1✔
49	extras: list[str]	1✔
50	git: str	1✔
51	rev: str	1✔
52	branch: str	1✔
53	python: str	1✔
54	markers: str	1✔
55	tag: str	1✔
56	version: str	1✔
57	url: str	1✔
58	path: str	1✔
59
60
61	def get_max_caret(parsed_version: Version) -> str:	1✔
UNCOV 62	major = 0	×
UNCOV 63	minor = 0	×
UNCOV 64	micro = 0	×
65
UNCOV 66	if parsed_version.major != 0:	×
UNCOV 67	major = parsed_version.major + 1	×
UNCOV 68	elif parsed_version.minor != 0:	×
UNCOV 69	minor = parsed_version.minor + 1	×
UNCOV 70	elif parsed_version.micro != 0:	×
UNCOV 71	micro = parsed_version.micro + 1	×
72	else:
UNCOV 73	base_len = len(parsed_version.base_version.split("."))	×
UNCOV 74	if base_len >= 3:	×
75	micro = 1	×
UNCOV 76	elif base_len == 2:	×
UNCOV 77	minor = 1	×
UNCOV 78	elif base_len == 1:	×
UNCOV 79	major = 1	×
80
UNCOV 81	return f"{major}.{minor}.{micro}"	×
82
83
84	def get_max_tilde(parsed_version: Version) -> str:	1✔
UNCOV 85	major = 0	×
UNCOV 86	minor = 0	×
UNCOV 87	base_len = len(parsed_version.base_version.split("."))	×
UNCOV 88	if base_len >= 2:	×
UNCOV 89	minor = int(str(parsed_version.minor)) + 1	×
UNCOV 90	major = int(str(parsed_version.major))	×
UNCOV 91	elif base_len == 1:	×
UNCOV 92	major = int(str(parsed_version.major)) + 1	×
93
UNCOV 94	return f"{major}.{minor}.0"	×
95
96
97	def get_max_wildcard(parsed_version: Version) -> str:	1✔
98	# Note: Assumes this is not a global wildcard, so parsed_version.release has
99	# at least two components.
UNCOV 100	release = list(parsed_version.release)	×
UNCOV 101	release[-2] += 1	×
UNCOV 102	major = release[0]	×
UNCOV 103	minor = release[1]	×
UNCOV 104	return f"{major}.{minor}.0"	×
105
106
107	def parse_str_version(attributes: str, **kwargs: str) -> str:	1✔
UNCOV 108	valid_specifiers = "<>!~="	×
UNCOV 109	pep440_reqs = []	×
UNCOV 110	proj_name = kwargs["proj_name"]	×
UNCOV 111	fp = kwargs["file_path"]	×
UNCOV 112	extras_str = kwargs["extras_str"]	×
UNCOV 113	comma_split_reqs = (i.strip() for i in attributes.split(","))	×
UNCOV 114	for req in comma_split_reqs:	×
115
UNCOV 116	def parse_version(version_str: str) -> Version:	×
UNCOV 117	try:	×
UNCOV 118	return Version(version_str)	×
119	except InvalidVersion:	×
120	raise InvalidVersion(	×
121	softwrap(
122	f"""
123	Failed to parse requirement {proj_name} = "{req}" in {fp} loaded by the
124	poetry_requirements macro.
125
126	If you believe this requirement is valid, consider opening an issue at
127	https://github.com/pantsbuild/pants/issues so that we can update Pants'
128	Poetry macro to support this.
129	"""
130	)
131	)
132
UNCOV 133	if not req:	×
134	continue	×
UNCOV 135	if req[0] == "^":	×
UNCOV 136	parsed_version = parse_version(req[1:])	×
UNCOV 137	max_ver = get_max_caret(parsed_version)	×
UNCOV 138	min_ver = f"{parsed_version.public}"	×
UNCOV 139	pep440_reqs.append(f">={min_ver},<{max_ver}")	×
UNCOV 140	elif req[0] == "~" and req[1] != "=":	×
141	# ~= is an acceptable default operator; however, ~ is not, and IS NOT the same as ~=
UNCOV 142	parsed_version = parse_version(req[1:])	×
UNCOV 143	max_ver = get_max_tilde(parsed_version)	×
UNCOV 144	min_ver = f"{parsed_version.public}"	×
UNCOV 145	pep440_reqs.append(f">={min_ver},<{max_ver}")	×
UNCOV 146	elif req[-1] == "*":	×
UNCOV 147	if req != "*": # This is not a global wildcard.	×
148	# To parse we replace the * with a 0.
UNCOV 149	parsed_version = parse_version(f"{req[:-1]}0")	×
UNCOV 150	max_ver = get_max_wildcard(parsed_version)	×
UNCOV 151	min_ver = f"{parsed_version.public}"	×
UNCOV 152	pep440_reqs.append(f">={min_ver},<{max_ver}")	×
153	else:
UNCOV 154	pep440_reqs.append(req if req[0] in valid_specifiers else f"=={req}")	×
UNCOV 155	return f"{proj_name}{extras_str} {','.join(pep440_reqs)}".rstrip()	×
156
157
158	def parse_python_constraint(constr: str \| None, fp: str) -> str:	1✔
UNCOV 159	if constr is None:	×
UNCOV 160	return ""	×
UNCOV 161	valid_specifiers = "<>!~= "	×
162	# If the user passes multiple Python constraints, they're separated by
163	# either '\|\|' signifying a logical 'or', or a comma signifying a logical
164	# 'and'. Hence, or_and_split is a 2D list where each inner list is a set of and-ed
165	# requirements; every list in the second layer is then or-ed together.
UNCOV 166	or_and_split = [[j.strip() for j in i.split(",")] for i in constr.split("\|\|")]	×
167
168	# We only use parse_str_version to address the version parsing; we don't
169	# care about having an actual Requirement object so things like the project name
170	# and extras that would ordinarily exist for a project with a string version are left blank here.
UNCOV 171	ver_parsed = [	×
172	[parse_str_version(j, proj_name="", file_path=fp, extras_str="") for j in i]
173	for i in or_and_split
174	]
175
UNCOV 176	def conv_and(lst: list[str]) -> list:	×
UNCOV 177	return list(itertools.chain(*[i.split(",") for i in lst]))	×
178
UNCOV 179	def prepend(version: str) -> str:	×
UNCOV 180	valid_versions = "".join(i for i in version if i in valid_specifiers)	×
UNCOV 181	invalid_versions = "".join(i for i in version if i not in valid_specifiers)	×
UNCOV 182	return f"python_version{valid_versions} '{invalid_versions}'"	×
183
UNCOV 184	prepend_and_clean = [	×
185	[prepend(".".join(j.split(".")[:2])) for j in conv_and(i)] for i in ver_parsed
186	]
UNCOV 187	return (	×
188	("(" if len(or_and_split) > 1 else "")
189	+ (") or (".join([" and ".join(i) for i in prepend_and_clean]))
190	+ (")" if len(or_and_split) > 1 else "")
191	)
192
193
194	@dataclass(frozen=True)	1✔
195	class PyProjectToml:	1✔
196	build_root: PurePath	1✔
197	toml_relpath: PurePath	1✔
198	toml_contents: str	1✔
199
200	def parse(self) -> Mapping[str, Any]:	1✔
UNCOV 201	return toml.loads(self.toml_contents)	×
202
203	def _non_pants_project_abs_path(self, path: Path) -> Path \| None:	1✔
UNCOV 204	resolved = path.resolve()	×
UNCOV 205	if resolved.is_file():	×
UNCOV 206	return resolved	×
207
UNCOV 208	try:	×
UNCOV 209	resolved.relative_to(self.build_root)	×
UNCOV 210	except ValueError:	×
UNCOV 211	return resolved	×
212
UNCOV 213	return None	×
214
215	def non_pants_project_abs_path(self, path: str) -> Path \| None:	1✔
216	"""Determine if the given path represents a non-Pants controlled project.
217
218	If the path points to a file, it's assumed the file is a distribution ( a wheel or sdist)
219	and the absolute path of that file is returned.
220
221	If the path points to a directory and that directory is outside of the build root, it's
222	assumed the directory is the root of a buildable Python project (i.e.: it contains a
223	pyproject.toml or setup.py) and the absolute path of the project is returned.
224
225	Otherwise, `None` is returned since the directory lies inside the build root and is assumed
226	to be a Pants controlled project.
227	"""
228	# TODO(John Sirois): This leaves the case where the path is a Python project directory
229	# inside the build root that the user actually wants Pex / Pip to build. A concrete case
230	# for this would be a repo where third party is partially handled with vendored exploded
231	# source distributions. If someone in the wild needs the described case, plumb a
232	# PoetryRequirements parameter that can list paths to treat as Pants controlled or
233	# vice-versa.
UNCOV 234	given_path = Path(path)	×
UNCOV 235	if given_path.is_absolute():	×
236	return self._non_pants_project_abs_path(given_path)	×
237	else:
UNCOV 238	return self._non_pants_project_abs_path(	×
239	Path(self.build_root / self.toml_relpath).parent / given_path
240	)
241
242
243	def produce_match(sep: str, feat: Any) -> str:	1✔
UNCOV 244	return f"{sep}{feat}" if feat else ""	×
245
246
247	def add_markers(base: str, attributes: PyprojectAttr, fp) -> str:	1✔
UNCOV 248	markers_lookup = produce_match("", attributes.get("markers"))	×
UNCOV 249	python_lookup = parse_python_constraint(attributes.get("python"), fp)	×
250
251	# Python constraints are passed as a `python_version` environment marker; if we have multiple
252	# markers, we evaluate them as one whole, and then AND with the new marker for the Python constraint.
253	# E.g. (marker1 AND marker2 OR marker3...) AND (python_version)
254	# rather than (marker1 AND marker2 OR marker3 AND python_version)
UNCOV 255	if not markers_lookup and not python_lookup:	×
UNCOV 256	return base	×
257
UNCOV 258	result = f"{base} ;("	×
259
UNCOV 260	if markers_lookup:	×
UNCOV 261	result += f"{markers_lookup})"	×
UNCOV 262	if python_lookup and markers_lookup:	×
UNCOV 263	result += " and ("	×
UNCOV 264	if python_lookup:	×
UNCOV 265	result += f"{python_lookup})"	×
266
UNCOV 267	return result	×
268
269
270	def handle_dict_attr(	1✔
271	proj_name: str, attributes: PyprojectAttr, pyproject_toml: PyProjectToml
272	) -> str \| None:
UNCOV 273	base = ""	×
UNCOV 274	fp = str(pyproject_toml.toml_relpath)	×
275
UNCOV 276	extras_lookup = attributes.get("extras")	×
UNCOV 277	if isinstance(extras_lookup, list):	×
UNCOV 278	extras_str = f"[{','.join(extras_lookup)}]"	×
279	else:
UNCOV 280	extras_str = ""	×
281
UNCOV 282	git_lookup = attributes.get("git")	×
UNCOV 283	if git_lookup is not None:	×
284	# If no URL scheme (e.g., `{git = "git@github.com:foo/bar.git"}`) we assume ssh,
285	# i.e., we convert to git+ssh://git@github.com/foo/bar.git.
UNCOV 286	if not urllib.parse.urlsplit(git_lookup).scheme:	×
UNCOV 287	git_lookup = f"ssh://{git_lookup.replace(':', '/', 1)}"	×
UNCOV 288	rev_lookup = produce_match("#", attributes.get("rev"))	×
UNCOV 289	branch_lookup = produce_match("@", attributes.get("branch"))	×
UNCOV 290	tag_lookup = produce_match("@", attributes.get("tag"))	×
291
UNCOV 292	base = f"{proj_name}{extras_str} @ git+{git_lookup}{tag_lookup}{branch_lookup}{rev_lookup}"	×
293
UNCOV 294	path_lookup = attributes.get("path")	×
UNCOV 295	if path_lookup is not None:	×
UNCOV 296	non_pants_project_abs_path = pyproject_toml.non_pants_project_abs_path(path_lookup)	×
UNCOV 297	if non_pants_project_abs_path:	×
UNCOV 298	base = f"{proj_name}{extras_str} @ file://{non_pants_project_abs_path}"	×
299	else:
300	# An internal path will be handled by normal Pants dependencies and dependency inference;
301	# i.e.: it never represents a third party requirement.
UNCOV 302	return None	×
303
UNCOV 304	url_lookup = attributes.get("url")	×
UNCOV 305	if url_lookup is not None:	×
UNCOV 306	base = f"{proj_name}{extras_str} @ {url_lookup}"	×
307
UNCOV 308	version_lookup = attributes.get("version")	×
UNCOV 309	if version_lookup is not None:	×
UNCOV 310	base = parse_str_version(	×
311	version_lookup, file_path=fp, extras_str=extras_str, proj_name=proj_name
312	)
313
UNCOV 314	if len(base) == 0:	×
315	raise ValueError(	×
316	softwrap(
317	f"""
318	{proj_name} is not formatted correctly; at minimum provide either a version, url,
319	path or git location for your dependency.
320	"""
321	)
322	)
323
UNCOV 324	return add_markers(base, attributes, fp)	×
325
326
327	def parse_single_dependency(	1✔
328	proj_name: str,
329	attributes: str \| Mapping[str, str \| Sequence] \| Sequence[Mapping[str, str \| Sequence]],
330	pyproject_toml: PyProjectToml,
331	) -> Iterator[PipRequirement]:
UNCOV 332	if isinstance(attributes, str):	×
333	# E.g. `foo = "~1.1~'.
UNCOV 334	yield PipRequirement.parse(	×
335	parse_str_version(
336	attributes,
337	proj_name=proj_name,
338	file_path=str(pyproject_toml.toml_relpath),
339	extras_str="",
340	)
341	)
UNCOV 342	elif isinstance(attributes, dict):	×
343	# E.g. `foo = {version = "~1.1"}`.
UNCOV 344	pyproject_attr = cast(PyprojectAttr, attributes)	×
UNCOV 345	req_str = handle_dict_attr(proj_name, pyproject_attr, pyproject_toml)	×
UNCOV 346	if req_str:	×
UNCOV 347	yield PipRequirement.parse(req_str)	×
UNCOV 348	elif isinstance(attributes, list):	×
349	# E.g. ` foo = [{version = "1.1","python" = "2.7"}, {version = "1.1","python" = "2.7"}]
UNCOV 350	for attr in attributes:	×
UNCOV 351	req_str = handle_dict_attr(proj_name, attr, pyproject_toml)	×
UNCOV 352	if req_str:	×
UNCOV 353	yield PipRequirement.parse(req_str)	×
354	else:
355	raise AssertionError(	×
356	softwrap(
357	f"""
358	Error: invalid Poetry requirement format. Expected type of requirement attributes to
359	be string, dict, or list, but was of type {type(attributes).__name__}.
360	"""
361	)
362	)
363
364
365	def parse_pyproject_toml(pyproject_toml: PyProjectToml) -> set[PipRequirement]:	1✔
UNCOV 366	parsed = pyproject_toml.parse()	×
UNCOV 367	try:	×
UNCOV 368	poetry_vals = parsed["tool"]["poetry"]	×
369	except KeyError:	×
370	raise KeyError(	×
371	softwrap(
372	f"""
373	No section `tool.poetry` found in {pyproject_toml.toml_relpath}, which
374	is loaded by Pants from a `poetry_requirements` macro.
375
376	Did you mean to set up Poetry?
377	"""
378	)
379	)
UNCOV 380	dependencies = poetry_vals.get("dependencies", {})	×
381	# N.B.: The "python" dependency is a special dependency required by Poetry that only serves to
382	# constraint the python interpreter versions the project works with; so we skip that.
383	# See: https://python-poetry.org/docs/pyproject/#dependencies-and-dev-dependencies
UNCOV 384	dependencies.pop("python", None)	×
385
UNCOV 386	groups = poetry_vals.get("group", {})	×
UNCOV 387	group_deps: dict[str, PyprojectAttr] = {}	×
388
UNCOV 389	for group in groups.values():	×
UNCOV 390	group_deps.update(group.get("dependencies", {}))	×
391
UNCOV 392	dev_dependencies = poetry_vals.get("dev-dependencies", {})	×
UNCOV 393	if not dependencies and not dev_dependencies and not group_deps:	×
394	logger.warning(	×
395	softwrap(
396	f"""
397	No requirements defined in any Poetry dependency groups, tool.poetry.dependencies
398	and tool.poetry.dev-dependencies in {pyproject_toml.toml_relpath}, which is loaded
399	by Pants from a poetry_requirements macro. Did you mean to populate these
400	with requirements?
401	"""
402	)
403	)
404
UNCOV 405	return set(	×
406	itertools.chain.from_iterable(
407	parse_single_dependency(proj, attr, pyproject_toml)
408	for proj, attr in {dependencies, dev_dependencies, **group_deps}.items()
409	)
410	)
411
412
413	def parse_poetry_requirements(	1✔
414	build_root: BuildRoot, file_contents: bytes, file_path: str
415	) -> set[PipRequirement]:
416	return parse_pyproject_toml(	×
417	PyProjectToml(
418	build_root=PurePath(build_root.path),
419	toml_relpath=PurePath(file_path),
420	toml_contents=file_contents.decode(),
421	)
422	)
423
424
425	# ---------------------------------------------------------------------------------
426	# Target generator
427	# ---------------------------------------------------------------------------------
428
429
430	class PoetryRequirementsSourceField(SingleSourceField):	1✔
431	default = "pyproject.toml"	1✔
432	required = False	1✔
433
434
435	class PoetryRequirementsTargetGenerator(TargetGenerator):	1✔
436	alias = "poetry_requirements"	1✔
437	help = "Generate a `python_requirement` for each entry in a Poetry `pyproject.toml`."	1✔
438	generated_target_cls = PythonRequirementTarget	1✔
439	# Note that this does not have a `dependencies` field.
440	core_fields = (	1✔
441	*COMMON_TARGET_FIELDS,
442	ModuleMappingField,
443	TypeStubsModuleMappingField,
444	PoetryRequirementsSourceField,
445	RequirementsOverrideField,
446	)
447	copied_fields = COMMON_TARGET_FIELDS	1✔
448	moved_fields = (PythonRequirementResolveField,)	1✔
449
450
451	class GenerateFromPoetryRequirementsRequest(GenerateTargetsRequest):	1✔
452	generate_from = PoetryRequirementsTargetGenerator	1✔
453
454
455	@rule(desc="Generate `python_requirement` targets from Poetry pyproject.toml", level=LogLevel.DEBUG)	1✔
456	async def generate_from_python_requirement(	1✔
457	request: GenerateFromPoetryRequirementsRequest,
458	build_root: BuildRoot,
459	union_membership: UnionMembership,
460	python_setup: PythonSetup,
461	) -> GeneratedTargets:
462	result = await _generate_requirements(	×
463	request,
464	union_membership,
465	python_setup,
466	parse_requirements_callback=partial(parse_poetry_requirements, build_root),
467	)
468	return GeneratedTargets(request.generator, result)	×
469
470
471	def rules():	1✔
472	return (	1✔
473	*collect_rules(),
474	UnionRule(GenerateTargetsRequest, GenerateFromPoetryRequirementsRequest),
475	)

pantsbuild / pants / 20332790708

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous