19015773527

Committed 02 Nov 2025 05:33PM UTC coverage: 17.872% (-62.4%) from 80.3%

Build # 19015773527

Build Type

Pull #22816

github

Committed by

web-flow

Commit Message

Merge a12d75757 into 6c024e162

Pull Request Pull Request #22816: Update Pants internal Python to 3.14

Run Details

4 of 5 new or added lines in 3 files covered. (80.0%)

28452 existing lines in 683 files now uncovered.

9831 of 55007 relevant lines covered (17.87%)

0.18 hits per line

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

47.62

/src/python/pants/backend/python/subsystems/python_tool_base.py

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

from __future__ import annotations

import importlib.resources
import json
import logging
import os
from collections.abc import Callable, Iterable, Sequence
from dataclasses import dataclass
from functools import cache
from typing import ClassVar
from urllib.parse import urlparse

from pants.backend.python.target_types import ConsoleScript, EntryPoint, MainSpecification
from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints
from pants.backend.python.util_rules.lockfile_metadata import PythonLockfileMetadata
from pants.backend.python.util_rules.pex import PexRequest
from pants.backend.python.util_rules.pex_requirements import (
    EntireLockfile,
    LoadedLockfile,
    LoadedLockfileRequest,
    Lockfile,
    PexRequirements,
    Resolve,
    get_lockfile_for_resolve,
    load_lockfile,
    strip_comments_from_pex_json_lockfile,
)
from pants.core.goals.resolves import ExportableTool
from pants.engine.fs import Digest
from pants.engine.rules import implicitly
from pants.option.errors import OptionsError
from pants.option.option_types import StrListOption, StrOption
from pants.option.subsystem import Subsystem
from pants.util.docutil import doc_url, git_url
from pants.util.meta import classproperty
from pants.util.pip_requirement import PipRequirement
from pants.util.strutil import softwrap, strval

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class _PackageNameAndVersion:
    name: str
    version: str


class PythonToolRequirementsBase(Subsystem, ExportableTool):
    """Base class for subsystems that configure a set of requirements for a python tool."""

    # Subclasses must set.
    default_version: ClassVar[str]
    # Must be set by subclasses - will be used to set the help text in this class.
    help_short: ClassVar[str | Callable[[], str]]
    # Subclasses do not need to override.
    default_extra_requirements: ClassVar[Sequence[str]] = []

    # Subclasses may set to override the value computed from default_version and
    # default_extra_requirements.
    # The primary package used in the subsystem must always be the first requirement.
    # TODO: Once we get rid of those options, subclasses must set this to loose
    #  requirements that reflect any minimum capabilities Pants assumes about the tool.
    default_requirements: Sequence[str] = []

    default_interpreter_constraints: ClassVar[Sequence[str]] = ["CPython>=3.9,<3.14"]
    register_interpreter_constraints: ClassVar[bool] = False

    default_lockfile_resource: ClassVar[tuple[str, str] | None] = None

    @classmethod
    def _help_extended(cls) -> str:
        base_help = strval(cls.help_short)
        help_paragraphs = [base_help]
        package_and_version = cls._default_package_name_and_version()
        if package_and_version:
            new_paragraph = f"This version of Pants uses `{package_and_version.name}` version {package_and_version.version} by default. Use a dedicated lockfile and the `install_from_resolve` option to control this."
            help_paragraphs.append(new_paragraph)

        return "\n\n".join(help_paragraphs)

    help = classproperty(_help_extended)

    @classmethod
    def _install_from_resolve_help(cls) -> str:
        package_and_version = cls._default_package_name_and_version()
        version_clause = (
            f", which uses `{package_and_version.name}` version {package_and_version.version}"
            if package_and_version
            else ""
        )
        return softwrap(
            f"""\
            If specified, install the tool using the lockfile for this named resolve.

            This resolve must be defined in `[python].resolves`, as described in
            {doc_url("docs/python/overview/lockfiles#lockfiles-for-tools")}.

            The resolve's entire lockfile will be installed, unless specific requirements are
            listed via the `requirements` option, in which case only those requirements
            will be installed. This is useful if you don't want to invalidate the tool's
            outputs when the resolve incurs changes to unrelated requirements.

            If unspecified, and the `lockfile` option is unset, the tool will be installed
            using the default lockfile shipped with Pants{version_clause}.

            If unspecified, and the `lockfile` option is set, the tool will use the custom
            `{cls.options_scope}` "tool lockfile" generated from the `version` and
            `extra_requirements` options. But note that this mechanism is deprecated.
            """
        )

    install_from_resolve = StrOption(
        advanced=True,
        default=None,
        help=lambda cls: cls._install_from_resolve_help(),
    )

    requirements = StrListOption(
        advanced=True,
        help=lambda cls: softwrap(
            """\
            If `install_from_resolve` is specified, install these requirements,
            at the versions provided by the specified resolve's lockfile.

            Values can be pip-style requirements (e.g., `tool` or `tool==1.2.3` or `tool>=1.2.3`),
            or addresses of `python_requirement` targets (or targets that generate or depend on
            `python_requirement` targets). Make sure to use the `//` prefix to refer to targets
            using their full address from the root (e.g. `//3rdparty/python:tool`). This is necessary
            to distinguish address specs from local or VCS requirements.

            The lockfile will be validated against the requirements - if a lockfile doesn't
            provide the requirement (at a suitable version, if the requirement specifies version
            constraints) Pants will error.

            If unspecified, install the entire lockfile.
            """
        ),
    )
    _interpreter_constraints = StrListOption(
        register_if=lambda cls: cls.register_interpreter_constraints,
        advanced=True,
        default=lambda cls: cls.default_interpreter_constraints,
        help="Python interpreter constraints for this tool.",
    )

    def __init__(self, *args, **kwargs):
        if (
            self.default_interpreter_constraints
            != PythonToolRequirementsBase.default_interpreter_constraints
            and not self.register_interpreter_constraints
        ):
            raise ValueError(
                softwrap(
                    f"""
                    `default_interpreter_constraints` are configured for `{self.options_scope}`, but
                    `register_interpreter_constraints` is not set to `True`, so the
                    `--interpreter-constraints` option will not be registered. Did you mean to set
                    this?
                    """
                )
            )

        if not self.default_lockfile_resource:
            raise ValueError(
                softwrap(
                    f"""
                    The class property `default_lockfile_resource` must be set. See `{self.options_scope}`.
                    """
                )
            )

        super().__init__(*args, **kwargs)

    @classproperty
    def default_lockfile_url(cls) -> str:
        assert cls.default_lockfile_resource is not None
        return git_url(
            os.path.join(
                "src",
                "python",
                cls.default_lockfile_resource[0].replace(".", os.path.sep),
                cls.default_lockfile_resource[1],
            )
        )

    @classmethod
    def help_for_generate_lockfile_with_default_location(cls, resolve_name):
        return softwrap(
            f"""
            You requested to generate a lockfile for {resolve_name} because
            you included it in `--generate-lockfiles-resolve`, but
            {resolve_name} is a tool using its default lockfile.

            If you would like to generate a lockfile for {resolve_name},
            follow the instructions for setting up lockfiles for tools
            {doc_url("docs/python/overview/lockfiles#lockfiles-for-tools")}
        """
        )

    @classmethod
    def pex_requirements_for_default_lockfile(cls):
        """Generate the pex requirements using this subsystem's default lockfile resource."""
        assert cls.default_lockfile_resource is not None
        pkg, path = cls.default_lockfile_resource
        url = f"resource://{pkg}/{path}"
        origin = f"The built-in default lockfile for {cls.options_scope}"
        return Lockfile(
            url=url,
            url_description_of_origin=origin,
            resolve_name=cls.options_scope,
        )

    @classmethod
    @cache
    def _default_package_name_and_version(cls) -> _PackageNameAndVersion | None:
        if cls.default_lockfile_resource is None:
            return None

        lockfile = cls.pex_requirements_for_default_lockfile()
        parts = urlparse(lockfile.url)
        # urlparse retains the leading / in URLs with a netloc.
        lockfile_path = parts.path[1:] if parts.path.startswith("/") else parts.path
        if parts.scheme in {"", "file"}:
            with open(lockfile_path, "rb") as fp:
                lock_bytes = fp.read()
        elif parts.scheme == "resource":
            # The "netloc" in our made-up "resource://" scheme is the package.
            lock_bytes = (
                importlib.resources.files(parts.netloc).joinpath(lockfile_path).read_bytes()
            )
        else:
            raise ValueError(
                f"Unsupported scheme {parts.scheme} for lockfile URL: {lockfile.url} "
                f"(origin: {lockfile.url_description_of_origin})"
            )

        stripped_lock_bytes = strip_comments_from_pex_json_lockfile(lock_bytes)
        lockfile_contents = json.loads(stripped_lock_bytes)
        # The first requirement must contain the primary package for this tool, otherwise
        # this will pick up the wrong requirement.
        first_default_requirement = PipRequirement.parse(cls.default_requirements[0])
        return next(
            _PackageNameAndVersion(
                name=first_default_requirement.name, version=requirement["version"]
            )
            for resolve in lockfile_contents["locked_resolves"]
            for requirement in resolve["locked_requirements"]
            if requirement["project_name"] == first_default_requirement.name
        )

    def pex_requirements(
        self,
        *,
        extra_requirements: Iterable[str] = (),
    ) -> PexRequirements | EntireLockfile:
        """The requirements to be used when installing the tool."""
        description_of_origin = f"the requirements of the `{self.options_scope}` tool"
        if self.install_from_resolve:
            use_entire_lockfile = not self.requirements
            return PexRequirements(
                (*self.requirements, *extra_requirements),
                from_superset=Resolve(self.install_from_resolve, use_entire_lockfile),
                description_of_origin=description_of_origin,
            )
        else:
            return EntireLockfile(self.pex_requirements_for_default_lockfile())

    @property
    def interpreter_constraints(self) -> InterpreterConstraints:
        """The interpreter constraints to use when installing and running the tool.

        This assumes you have set the class property `register_interpreter_constraints = True`.
        """
        return InterpreterConstraints(self._interpreter_constraints)

    def to_pex_request(
        self,
        *,
        interpreter_constraints: InterpreterConstraints | None = None,
        extra_requirements: Iterable[str] = (),
        main: MainSpecification | None = None,
        sources: Digest | None = None,
    ) -> PexRequest:
        requirements = self.pex_requirements(extra_requirements=extra_requirements)
        if not interpreter_constraints:
            if self.options.is_default("interpreter_constraints") and (
                isinstance(requirements, EntireLockfile)
                or (
                    isinstance(requirements, PexRequirements)
                    and isinstance(requirements.from_superset, Resolve)
                )
            ):
                # If installing the tool from a resolve, and custom ICs weren't explicitly set,
                # leave these blank. This will cause the ones for the resolve to be used,
                # which is clearly what the user intends, rather than forcing the
                # user to override interpreter_constraints to match those of the resolve.
                interpreter_constraints = InterpreterConstraints()
            else:
                interpreter_constraints = self.interpreter_constraints
        return PexRequest(
            output_filename=f"{self.options_scope.replace('-', '_')}.pex",
            internal_only=True,
            requirements=requirements,
            interpreter_constraints=interpreter_constraints,
            main=main,
            sources=sources,
        )


class PythonToolBase(PythonToolRequirementsBase):
    """Base class for subsystems that configure a python tool to be invoked out-of-process."""

    # Subclasses must set.
    default_main: ClassVar[MainSpecification]

    # Though possible, we do not recommend setting `default_main` to an Executable
    # instead of a ConsoleScript or an EntryPoint. Executable is a niche pex feature
    # designed to support poorly named executable python scripts that cannot be imported
    # (eg when a file has a character like "-" that is not valid in python identifiers).
    # As this should be rare or even non-existent, we do NOT add an `executable` option
    # to mirror the other MainSpecification options.

    console_script = StrOption(
        advanced=True,
        default=lambda cls: (
            cls.default_main.spec if isinstance(cls.default_main, ConsoleScript) else None
        ),
        help=softwrap(
            """
            The console script for the tool. Using this option is generally preferable to
            (and mutually exclusive with) specifying an `--entry-point` since console script
            names have a higher expectation of staying stable across releases of the tool.
            Usually, you will not want to change this from the default.
            """
        ),
    )
    entry_point = StrOption(
        advanced=True,
        default=lambda cls: (
            cls.default_main.spec if isinstance(cls.default_main, EntryPoint) else None
        ),
        help=softwrap(
            """
            The entry point for the tool. Generally you only want to use this option if the
            tool does not offer a `--console-script` (which this option is mutually exclusive
            with). Usually, you will not want to change this from the default.
            """
        ),
    )

    @property
    def main(self) -> MainSpecification:
        is_default_console_script = self.options.is_default("console_script")
        is_default_entry_point = self.options.is_default("entry_point")
        if not is_default_console_script and not is_default_entry_point:
            raise OptionsError(
                softwrap(
                    f"""
                    Both [{self.options_scope}].console-script={self.console_script} and
                    [{self.options_scope}].entry-point={self.entry_point} are configured
                    but these options are mutually exclusive. Please pick one.
                    """
                )
            )
        if not is_default_console_script:
            assert self.console_script is not None
            return ConsoleScript(self.console_script)
        if not is_default_entry_point:
            assert self.entry_point is not None
            return EntryPoint.parse(self.entry_point)
        return self.default_main

    def to_pex_request(
        self,
        *,
        interpreter_constraints: InterpreterConstraints | None = None,
        extra_requirements: Iterable[str] = (),
        main: MainSpecification | None = None,
        sources: Digest | None = None,
    ) -> PexRequest:
        return super().to_pex_request(
            interpreter_constraints=interpreter_constraints,
            extra_requirements=extra_requirements,
            main=main or self.main,
            sources=sources,
        )


async def get_loaded_lockfile(subsystem: PythonToolBase) -> LoadedLockfile:
    requirements = subsystem.pex_requirements()
    if isinstance(requirements, EntireLockfile):
        lockfile = requirements.lockfile
    else:
        assert isinstance(requirements, PexRequirements)
        assert isinstance(requirements.from_superset, Resolve)
        lockfile = await get_lockfile_for_resolve(requirements.from_superset, **implicitly())
    loaded_lockfile = await load_lockfile(LoadedLockfileRequest(lockfile), **implicitly())
    return loaded_lockfile


async def get_lockfile_metadata(subsystem: PythonToolBase) -> PythonLockfileMetadata:
    loaded_lockfile = await get_loaded_lockfile(subsystem)
    assert loaded_lockfile.metadata is not None
    return loaded_lockfile.metadata


async def get_lockfile_interpreter_constraints(
    subsystem: PythonToolBase,
) -> InterpreterConstraints:
    """If a lockfile is used, will try to find the interpreter constraints used to generate the
    lock.

    This allows us to work around https://github.com/pantsbuild/pants/issues/14912.
    """
    # If the tool's interpreter constraints are explicitly set, or it is not using a lockfile at
    # all, then we should use the tool's interpreter constraints option.
    if not subsystem.options.is_default("interpreter_constraints"):
        return subsystem.interpreter_constraints

    lockfile_metadata = await get_lockfile_metadata(subsystem)
    return lockfile_metadata.valid_for_interpreter_constraints

1	# Copyright 2018 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 importlib.resources	1✔
7	import json	1✔
8	import logging	1✔
9	import os	1✔
10	from collections.abc import Callable, Iterable, Sequence	1✔
11	from dataclasses import dataclass	1✔
12	from functools import cache	1✔
13	from typing import ClassVar	1✔
14	from urllib.parse import urlparse	1✔
15
16	from pants.backend.python.target_types import ConsoleScript, EntryPoint, MainSpecification	1✔
17	from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints	1✔
18	from pants.backend.python.util_rules.lockfile_metadata import PythonLockfileMetadata	1✔
19	from pants.backend.python.util_rules.pex import PexRequest	1✔
20	from pants.backend.python.util_rules.pex_requirements import (	1✔
21	EntireLockfile,
22	LoadedLockfile,
23	LoadedLockfileRequest,
24	Lockfile,
25	PexRequirements,
26	Resolve,
27	get_lockfile_for_resolve,
28	load_lockfile,
29	strip_comments_from_pex_json_lockfile,
30	)
31	from pants.core.goals.resolves import ExportableTool	1✔
32	from pants.engine.fs import Digest	1✔
33	from pants.engine.rules import implicitly	1✔
34	from pants.option.errors import OptionsError	1✔
35	from pants.option.option_types import StrListOption, StrOption	1✔
36	from pants.option.subsystem import Subsystem	1✔
37	from pants.util.docutil import doc_url, git_url	1✔
38	from pants.util.meta import classproperty	1✔
39	from pants.util.pip_requirement import PipRequirement	1✔
40	from pants.util.strutil import softwrap, strval	1✔
41
42	logger = logging.getLogger(__name__)	1✔
43
44
45	@dataclass(frozen=True)	1✔
46	class _PackageNameAndVersion:	1✔
47	name: str	1✔
48	version: str	1✔
49
50
51	class PythonToolRequirementsBase(Subsystem, ExportableTool):	1✔
52	"""Base class for subsystems that configure a set of requirements for a python tool."""
53
54	# Subclasses must set.
55	default_version: ClassVar[str]	1✔
56	# Must be set by subclasses - will be used to set the help text in this class.
57	help_short: ClassVar[str \| Callable[[], str]]	1✔
58	# Subclasses do not need to override.
59	default_extra_requirements: ClassVar[Sequence[str]] = []	1✔
60
61	# Subclasses may set to override the value computed from default_version and
62	# default_extra_requirements.
63	# The primary package used in the subsystem must always be the first requirement.
64	# TODO: Once we get rid of those options, subclasses must set this to loose
65	# requirements that reflect any minimum capabilities Pants assumes about the tool.
66	default_requirements: Sequence[str] = []	1✔
67
68	default_interpreter_constraints: ClassVar[Sequence[str]] = ["CPython>=3.9,<3.14"]	1✔
69	register_interpreter_constraints: ClassVar[bool] = False	1✔
70
71	default_lockfile_resource: ClassVar[tuple[str, str] \| None] = None	1✔
72
73	@classmethod	1✔
74	def _help_extended(cls) -> str:	1✔
UNCOV 75	base_help = strval(cls.help_short)	×
UNCOV 76	help_paragraphs = [base_help]	×
UNCOV 77	package_and_version = cls._default_package_name_and_version()	×
UNCOV 78	if package_and_version:	×
UNCOV 79	new_paragraph = f"This version of Pants uses `{package_and_version.name}` version {package_and_version.version} by default. Use a dedicated lockfile and the `install_from_resolve` option to control this."	×
UNCOV 80	help_paragraphs.append(new_paragraph)	×
81
UNCOV 82	return "\n\n".join(help_paragraphs)	×
83
84	help = classproperty(_help_extended)	1✔
85
86	@classmethod	1✔
87	def _install_from_resolve_help(cls) -> str:	1✔
UNCOV 88	package_and_version = cls._default_package_name_and_version()	×
UNCOV 89	version_clause = (	×
90	f", which uses `{package_and_version.name}` version {package_and_version.version}"
91	if package_and_version
92	else ""
93	)
UNCOV 94	return softwrap(	×
95	f"""\
96	If specified, install the tool using the lockfile for this named resolve.
97
98	This resolve must be defined in `[python].resolves`, as described in
99	{doc_url("docs/python/overview/lockfiles#lockfiles-for-tools")}.
100
101	The resolve's entire lockfile will be installed, unless specific requirements are
102	listed via the `requirements` option, in which case only those requirements
103	will be installed. This is useful if you don't want to invalidate the tool's
104	outputs when the resolve incurs changes to unrelated requirements.
105
106	If unspecified, and the `lockfile` option is unset, the tool will be installed
107	using the default lockfile shipped with Pants{version_clause}.
108
109	If unspecified, and the `lockfile` option is set, the tool will use the custom
110	`{cls.options_scope}` "tool lockfile" generated from the `version` and
111	`extra_requirements` options. But note that this mechanism is deprecated.
112	"""
113	)
114
115	install_from_resolve = StrOption(	1✔
116	advanced=True,
117	default=None,
118	help=lambda cls: cls._install_from_resolve_help(),
119	)
120
121	requirements = StrListOption(	1✔
122	advanced=True,
123	help=lambda cls: softwrap(
124	"""\
125	If `install_from_resolve` is specified, install these requirements,
126	at the versions provided by the specified resolve's lockfile.
127
128	Values can be pip-style requirements (e.g., `tool` or `tool==1.2.3` or `tool>=1.2.3`),
129	or addresses of `python_requirement` targets (or targets that generate or depend on
130	`python_requirement` targets). Make sure to use the `//` prefix to refer to targets
131	using their full address from the root (e.g. `//3rdparty/python:tool`). This is necessary
132	to distinguish address specs from local or VCS requirements.
133
134	The lockfile will be validated against the requirements - if a lockfile doesn't
135	provide the requirement (at a suitable version, if the requirement specifies version
136	constraints) Pants will error.
137
138	If unspecified, install the entire lockfile.
139	"""
140	),
141	)
142	_interpreter_constraints = StrListOption(	1✔
143	register_if=lambda cls: cls.register_interpreter_constraints,
144	advanced=True,
145	default=lambda cls: cls.default_interpreter_constraints,
146	help="Python interpreter constraints for this tool.",
147	)
148
149	def __init__(self, args, *kwargs):	1✔
UNCOV 150	if (	×
151	self.default_interpreter_constraints
152	!= PythonToolRequirementsBase.default_interpreter_constraints
153	and not self.register_interpreter_constraints
154	):
155	raise ValueError(	×
156	softwrap(
157	f"""
158	`default_interpreter_constraints` are configured for `{self.options_scope}`, but
159	`register_interpreter_constraints` is not set to `True`, so the
160	`--interpreter-constraints` option will not be registered. Did you mean to set
161	this?
162	"""
163	)
164	)
165
UNCOV 166	if not self.default_lockfile_resource:	×
167	raise ValueError(	×
168	softwrap(
169	f"""
170	The class property `default_lockfile_resource` must be set. See `{self.options_scope}`.
171	"""
172	)
173	)
174
UNCOV 175	super().__init__(args, *kwargs)	×
176
177	@classproperty	1✔
178	def default_lockfile_url(cls) -> str:	1✔
UNCOV 179	assert cls.default_lockfile_resource is not None	×
UNCOV 180	return git_url(	×
181	os.path.join(
182	"src",
183	"python",
184	cls.default_lockfile_resource[0].replace(".", os.path.sep),
185	cls.default_lockfile_resource[1],
186	)
187	)
188
189	@classmethod	1✔
190	def help_for_generate_lockfile_with_default_location(cls, resolve_name):	1✔
191	return softwrap(	×
192	f"""
193	You requested to generate a lockfile for {resolve_name} because
194	you included it in `--generate-lockfiles-resolve`, but
195	{resolve_name} is a tool using its default lockfile.
196
197	If you would like to generate a lockfile for {resolve_name},
198	follow the instructions for setting up lockfiles for tools
199	{doc_url("docs/python/overview/lockfiles#lockfiles-for-tools")}
200	"""
201	)
202
203	@classmethod	1✔
204	def pex_requirements_for_default_lockfile(cls):	1✔
205	"""Generate the pex requirements using this subsystem's default lockfile resource."""
UNCOV 206	assert cls.default_lockfile_resource is not None	×
UNCOV 207	pkg, path = cls.default_lockfile_resource	×
UNCOV 208	url = f"resource://{pkg}/{path}"	×
UNCOV 209	origin = f"The built-in default lockfile for {cls.options_scope}"	×
UNCOV 210	return Lockfile(	×
211	url=url,
212	url_description_of_origin=origin,
213	resolve_name=cls.options_scope,
214	)
215
216	@classmethod	1✔
217	@cache	1✔
218	def _default_package_name_and_version(cls) -> _PackageNameAndVersion \| None:	1✔
UNCOV 219	if cls.default_lockfile_resource is None:	×
220	return None	×
221
UNCOV 222	lockfile = cls.pex_requirements_for_default_lockfile()	×
UNCOV 223	parts = urlparse(lockfile.url)	×
224	# urlparse retains the leading / in URLs with a netloc.
UNCOV 225	lockfile_path = parts.path[1:] if parts.path.startswith("/") else parts.path	×
UNCOV 226	if parts.scheme in {"", "file"}:	×
227	with open(lockfile_path, "rb") as fp:	×
228	lock_bytes = fp.read()	×
UNCOV 229	elif parts.scheme == "resource":	×
230	# The "netloc" in our made-up "resource://" scheme is the package.
UNCOV 231	lock_bytes = (	×
232	importlib.resources.files(parts.netloc).joinpath(lockfile_path).read_bytes()
233	)
234	else:
235	raise ValueError(	×
236	f"Unsupported scheme {parts.scheme} for lockfile URL: {lockfile.url} "
237	f"(origin: {lockfile.url_description_of_origin})"
238	)
239
UNCOV 240	stripped_lock_bytes = strip_comments_from_pex_json_lockfile(lock_bytes)	×
UNCOV 241	lockfile_contents = json.loads(stripped_lock_bytes)	×
242	# The first requirement must contain the primary package for this tool, otherwise
243	# this will pick up the wrong requirement.
UNCOV 244	first_default_requirement = PipRequirement.parse(cls.default_requirements[0])	×
UNCOV 245	return next(	×
246	_PackageNameAndVersion(
247	name=first_default_requirement.name, version=requirement["version"]
248	)
249	for resolve in lockfile_contents["locked_resolves"]
250	for requirement in resolve["locked_requirements"]
251	if requirement["project_name"] == first_default_requirement.name
252	)
253
254	def pex_requirements(	1✔
255	self,
256	*,
257	extra_requirements: Iterable[str] = (),
258	) -> PexRequirements \| EntireLockfile:
259	"""The requirements to be used when installing the tool."""
UNCOV 260	description_of_origin = f"the requirements of the `{self.options_scope}` tool"	×
UNCOV 261	if self.install_from_resolve:	×
UNCOV 262	use_entire_lockfile = not self.requirements	×
UNCOV 263	return PexRequirements(	×
264	(self.requirements, extra_requirements),
265	from_superset=Resolve(self.install_from_resolve, use_entire_lockfile),
266	description_of_origin=description_of_origin,
267	)
268	else:
UNCOV 269	return EntireLockfile(self.pex_requirements_for_default_lockfile())	×
270
271	@property	1✔
272	def interpreter_constraints(self) -> InterpreterConstraints:	1✔
273	"""The interpreter constraints to use when installing and running the tool.
274
275	This assumes you have set the class property `register_interpreter_constraints = True`.
276	"""
UNCOV 277	return InterpreterConstraints(self._interpreter_constraints)	×
278
279	def to_pex_request(	1✔
280	self,
281	*,
282	interpreter_constraints: InterpreterConstraints \| None = None,
283	extra_requirements: Iterable[str] = (),
284	main: MainSpecification \| None = None,
285	sources: Digest \| None = None,
286	) -> PexRequest:
287	requirements = self.pex_requirements(extra_requirements=extra_requirements)	×
288	if not interpreter_constraints:	×
289	if self.options.is_default("interpreter_constraints") and (	×
290	isinstance(requirements, EntireLockfile)
291	or (
292	isinstance(requirements, PexRequirements)
293	and isinstance(requirements.from_superset, Resolve)
294	)
295	):
296	# If installing the tool from a resolve, and custom ICs weren't explicitly set,
297	# leave these blank. This will cause the ones for the resolve to be used,
298	# which is clearly what the user intends, rather than forcing the
299	# user to override interpreter_constraints to match those of the resolve.
300	interpreter_constraints = InterpreterConstraints()	×
301	else:
302	interpreter_constraints = self.interpreter_constraints	×
303	return PexRequest(	×
304	output_filename=f"{self.options_scope.replace('-', '_')}.pex",
305	internal_only=True,
306	requirements=requirements,
307	interpreter_constraints=interpreter_constraints,
308	main=main,
309	sources=sources,
310	)
311
312
313	class PythonToolBase(PythonToolRequirementsBase):	1✔
314	"""Base class for subsystems that configure a python tool to be invoked out-of-process."""
315
316	# Subclasses must set.
317	default_main: ClassVar[MainSpecification]	1✔
318
319	# Though possible, we do not recommend setting `default_main` to an Executable
320	# instead of a ConsoleScript or an EntryPoint. Executable is a niche pex feature
321	# designed to support poorly named executable python scripts that cannot be imported
322	# (eg when a file has a character like "-" that is not valid in python identifiers).
323	# As this should be rare or even non-existent, we do NOT add an `executable` option
324	# to mirror the other MainSpecification options.
325
326	console_script = StrOption(	1✔
327	advanced=True,
328	default=lambda cls: (
329	cls.default_main.spec if isinstance(cls.default_main, ConsoleScript) else None
330	),
331	help=softwrap(
332	"""
333	The console script for the tool. Using this option is generally preferable to
334	(and mutually exclusive with) specifying an `--entry-point` since console script
335	names have a higher expectation of staying stable across releases of the tool.
336	Usually, you will not want to change this from the default.
337	"""
338	),
339	)
340	entry_point = StrOption(	1✔
341	advanced=True,
342	default=lambda cls: (
343	cls.default_main.spec if isinstance(cls.default_main, EntryPoint) else None
344	),
345	help=softwrap(
346	"""
347	The entry point for the tool. Generally you only want to use this option if the
348	tool does not offer a `--console-script` (which this option is mutually exclusive
349	with). Usually, you will not want to change this from the default.
350	"""
351	),
352	)
353
354	@property	1✔
355	def main(self) -> MainSpecification:	1✔
356	is_default_console_script = self.options.is_default("console_script")	×
357	is_default_entry_point = self.options.is_default("entry_point")	×
358	if not is_default_console_script and not is_default_entry_point:	×
359	raise OptionsError(	×
360	softwrap(
361	f"""
362	Both [{self.options_scope}].console-script={self.console_script} and
363	[{self.options_scope}].entry-point={self.entry_point} are configured
364	but these options are mutually exclusive. Please pick one.
365	"""
366	)
367	)
368	if not is_default_console_script:	×
369	assert self.console_script is not None	×
370	return ConsoleScript(self.console_script)	×
371	if not is_default_entry_point:	×
372	assert self.entry_point is not None	×
373	return EntryPoint.parse(self.entry_point)	×
374	return self.default_main	×
375
376	def to_pex_request(	1✔
377	self,
378	*,
379	interpreter_constraints: InterpreterConstraints \| None = None,
380	extra_requirements: Iterable[str] = (),
381	main: MainSpecification \| None = None,
382	sources: Digest \| None = None,
383	) -> PexRequest:
384	return super().to_pex_request(	×
385	interpreter_constraints=interpreter_constraints,
386	extra_requirements=extra_requirements,
387	main=main or self.main,
388	sources=sources,
389	)
390
391
392	async def get_loaded_lockfile(subsystem: PythonToolBase) -> LoadedLockfile:	1✔
UNCOV 393	requirements = subsystem.pex_requirements()	×
UNCOV 394	if isinstance(requirements, EntireLockfile):	×
UNCOV 395	lockfile = requirements.lockfile	×
396	else:
UNCOV 397	assert isinstance(requirements, PexRequirements)	×
UNCOV 398	assert isinstance(requirements.from_superset, Resolve)	×
UNCOV 399	lockfile = await get_lockfile_for_resolve(requirements.from_superset, **implicitly())	×
UNCOV 400	loaded_lockfile = await load_lockfile(LoadedLockfileRequest(lockfile), **implicitly())	×
UNCOV 401	return loaded_lockfile	×
402
403
404	async def get_lockfile_metadata(subsystem: PythonToolBase) -> PythonLockfileMetadata:	1✔
UNCOV 405	loaded_lockfile = await get_loaded_lockfile(subsystem)	×
UNCOV 406	assert loaded_lockfile.metadata is not None	×
UNCOV 407	return loaded_lockfile.metadata	×
408
409
410	async def get_lockfile_interpreter_constraints(	1✔
411	subsystem: PythonToolBase,
412	) -> InterpreterConstraints:
413	"""If a lockfile is used, will try to find the interpreter constraints used to generate the
414	lock.
415
416	This allows us to work around https://github.com/pantsbuild/pants/issues/14912.
417	"""
418	# If the tool's interpreter constraints are explicitly set, or it is not using a lockfile at
419	# all, then we should use the tool's interpreter constraints option.
UNCOV 420	if not subsystem.options.is_default("interpreter_constraints"):	×
UNCOV 421	return subsystem.interpreter_constraints	×
422
UNCOV 423	lockfile_metadata = await get_lockfile_metadata(subsystem)	×
UNCOV 424	return lockfile_metadata.valid_for_interpreter_constraints	×

pantsbuild / pants / 19015773527

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