26080722777

Committed 19 May 2026 06:37AM UTC coverage: 52.106% (-11.5%) from 63.597%

Build # 26080722777

Build Type

Pull #23250

github

Committed by

web-flow

Commit Message

Merge 63ec06323 into 2693df832

Pull Request Pull Request #23250: Feature: Add generic option to docker image

Coverage Stats

12 of 50 new or added lines in 3 files covered. (24.0%)

5382 existing lines in 201 files now uncovered.

32053 of 61515 relevant lines covered (52.11%)

1.04 hits per line

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

75.31

/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

import toml

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.15"]
    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})"
            )

        # Note that this code relies on knowing the internal structure of uv and pex lockfiles,
        # neither of which are guaranteed. If parsing fails we'll return None and the calling
        # code will deal with it (by omitting some information from help strings).
        first_default_requirement = PipRequirement.parse(cls.default_requirements[0])
        stripped_lock_bytes = strip_comments_from_pex_json_lockfile(lock_bytes)

        try:  # Is it uv.lock?
            lockfile_contents = toml.loads(lock_bytes.decode())
            # The first requirement must contain the primary package for this tool, otherwise
            # this will pick up the wrong requirement.
            return next(
                _PackageNameAndVersion(name=first_default_requirement.name, version=pkg["version"])
                for pkg in lockfile_contents["package"]
                if pkg["name"] == first_default_requirement.name
            )
        except KeyError:
            # It's toml, but the schema is off.
            logger.warning(f"File at {lockfile.url} was not of expected uv.lock schema.")
            return None
        except toml.TomlDecodeError:
            # It's not toml, so it's not uv.lock format. Fall through to try pex lockfile format.
            # TODO: This is only used internally, for generating help, based on the *default*
            #  lockfiles. Once those are all uv, and we are confident they will remain so,
            #  we can remove the JSON handling here.
            pass

        try:
            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.
            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
            )
        except KeyError:
            # It's json, but the schema is off.
            logger.warning(f"File at {lockfile.url} was not of expected pex lock schema.")
            return None
        except json.JSONDecodeError:
            # It's not json, so it's not pex lock format.
            raise Exception(f"File at {lockfile.url} was not of any recognized lockfile format")

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

pantsbuild / pants / 26080722777

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