25405422172

Committed 05 May 2026 10:18PM UTC coverage: 92.879% (-0.07%) from 92.944%

Build # 25405422172

Build Type

Pull #23319

github

Committed by

web-flow

Commit Message

Merge c82d0f333 into e8b784f89

Pull Request Pull Request #23319: [pants_ng] Scaffolding for a pants_ng mode.

Coverage Stats

25 of 76 new or added lines in 9 files covered. (32.89%)

209 existing lines in 15 files now uncovered.

92234 of 99306 relevant lines covered (92.88%)

4.05 hits per line

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

98.17

/src/python/pants/backend/python/util_rules/pex_cli.py

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

from __future__ import annotations

import dataclasses
import logging
import os
from collections.abc import Iterable, Mapping
from dataclasses import dataclass

from pants.backend.python.subsystems.python_native_code import PythonNativeCodeSubsystem
from pants.backend.python.subsystems.setup import PythonSetup
from pants.backend.python.util_rules import pex_environment
from pants.backend.python.util_rules.pex_environment import (
    PexEnvironment,
    PexSubsystem,
    PythonExecutable,
)
from pants.core.goals.resolves import ExportableTool
from pants.core.util_rules import adhoc_binaries, external_tool
from pants.core.util_rules.adhoc_binaries import PythonBuildStandaloneBinary
from pants.core.util_rules.external_tool import (
    DownloadedExternalTool,
    TemplatedExternalTool,
    download_external_tool,
)
from pants.engine.fs import CreateDigest, Digest, Directory, MergeDigests
from pants.engine.internals.selectors import concurrently
from pants.engine.intrinsics import create_digest, merge_digests
from pants.engine.platform import Platform
from pants.engine.process import Process, ProcessCacheScope
from pants.engine.rules import collect_rules, rule
from pants.engine.unions import UnionRule
from pants.option.global_options import GlobalOptions, ca_certs_path_to_file_content
from pants.option.option_types import ArgsListOption
from pants.util.frozendict import FrozenDict
from pants.util.logging import LogLevel
from pants.util.strutil import softwrap

logger = logging.getLogger(__name__)


_PEX_VERSION = "v2.93.2"
_PEX_BINARY_HASH = "75222f01b34cbe2bbe0aa9564dbf554d10dd22875ce59b6ea0c16665c92adc33"
_PEX_BINARY_SIZE = 5116653


class PexCli(TemplatedExternalTool):
    options_scope = "pex-cli"
    name = "pex"
    help = "The PEX (Python EXecutable) tool (https://github.com/pex-tool/pex)."

    default_version = _PEX_VERSION
    default_url_template = "https://github.com/pex-tool/pex/releases/download/{version}/pex"
    version_constraints = ">=2.76.0,<3.0"

    # extra args to be passed to the pex tool; note that they
    # are going to apply to all invocations of the pex tool.
    global_args = ArgsListOption(
        example="--check=error --no-compile",
        extra_help=softwrap(
            """
            Note that these apply to all invocations of the pex tool, including building `pex_binary`
            targets, preparing `python_test` targets to run, and generating lockfiles.
            """
        ),
    )

    default_known_versions = [
        f"{_PEX_VERSION}|{platform}|{_PEX_BINARY_HASH}|{_PEX_BINARY_SIZE}"
        for platform in ["macos_x86_64", "macos_arm64", "linux_x86_64", "linux_arm64"]
    ]


@dataclass(frozen=True)
class PexCliProcess:
    interpreter: PythonExecutable | PythonBuildStandaloneBinary | None
    subcommand: tuple[str, ...]
    extra_args: tuple[str, ...]
    description: str = dataclasses.field(compare=False)
    additional_input_digest: Digest | None
    extra_env: FrozenDict[str, str] | None
    append_only_caches: FrozenDict[str, str] | None
    output_files: tuple[str, ...] | None
    output_directories: tuple[str, ...] | None
    level: LogLevel
    concurrency_available: int
    cache_scope: ProcessCacheScope

    def __init__(
        self,
        *,
        interpreter: PythonExecutable | PythonBuildStandaloneBinary | None = None,
        subcommand: Iterable[str],
        extra_args: Iterable[str],
        description: str,
        additional_input_digest: Digest | None = None,
        extra_env: Mapping[str, str] | None = None,
        append_only_caches: Mapping[str, str] | None = None,
        output_files: Iterable[str] | None = None,
        output_directories: Iterable[str] | None = None,
        level: LogLevel = LogLevel.INFO,
        concurrency_available: int = 0,
        cache_scope: ProcessCacheScope = ProcessCacheScope.SUCCESSFUL,
    ) -> None:
        object.__setattr__(self, "interpreter", interpreter)
        object.__setattr__(self, "subcommand", tuple(subcommand))
        object.__setattr__(self, "extra_args", tuple(extra_args))
        object.__setattr__(self, "description", description)
        object.__setattr__(self, "additional_input_digest", additional_input_digest)
        object.__setattr__(self, "extra_env", FrozenDict(extra_env) if extra_env else None)
        object.__setattr__(
            self,
            "append_only_caches",
            FrozenDict(append_only_caches) if append_only_caches else None,
        )
        object.__setattr__(self, "output_files", tuple(output_files) if output_files else None)
        object.__setattr__(
            self, "output_directories", tuple(output_directories) if output_directories else None
        )
        object.__setattr__(self, "level", level)
        object.__setattr__(self, "concurrency_available", concurrency_available)
        object.__setattr__(self, "cache_scope", cache_scope)

        self.__post_init__()

    def __post_init__(self) -> None:
        if "--pex-root-path" in self.extra_args:
            raise ValueError("`--pex-root` flag not allowed. We set its value for you.")


class PexPEX(DownloadedExternalTool):
    """The Pex PEX binary."""


@rule
async def download_pex_pex(pex_cli: PexCli, platform: Platform) -> PexPEX:
    pex_pex = await download_external_tool(pex_cli.get_request(platform))
    return PexPEX(digest=pex_pex.digest, exe=pex_pex.exe)


@rule
async def setup_pex_cli_process(
    request: PexCliProcess,
    pex_pex: PexPEX,
    pex_env: PexEnvironment,
    bootstrap_python: PythonBuildStandaloneBinary,
    python_native_code: PythonNativeCodeSubsystem.EnvironmentAware,
    global_options: GlobalOptions,
    pex_subsystem: PexSubsystem,
    pex_cli_subsystem: PexCli,
    python_setup: PythonSetup,
) -> Process:
    tmpdir = ".tmp"
    gets = [create_digest(CreateDigest([Directory(tmpdir)]))]

    cert_args = []
    if global_options.ca_certs_path:
        ca_certs_fc = ca_certs_path_to_file_content(global_options.ca_certs_path)
        gets.append(create_digest(CreateDigest((ca_certs_fc,))))
        cert_args = ["--cert", ca_certs_fc.path]

    digests_to_merge = [pex_pex.digest]
    digests_to_merge.extend(await concurrently(gets))
    if request.additional_input_digest:
        digests_to_merge.append(request.additional_input_digest)
    input_digest = await merge_digests(MergeDigests(digests_to_merge))

    global_args = [
        # Ensure Pex and its subprocesses create temporary files in the process execution
        # sandbox. It may make sense to do this generally for Processes, but in the short term we
        # have known use cases where /tmp is too small to hold large wheel downloads Pex is asked to
        # perform. Making the TMPDIR local to the sandbox allows control via
        # --local-execution-root-dir for the local case and should work well with remote cases where
        # a remoting implementation has to allow for processes producing large binaries in a
        # sandbox to support reasonable workloads. Communicating TMPDIR via --tmpdir instead of via
        # environment variable allows Pex to absolutize the path ensuring subprocesses that change
        # CWD can find the TMPDIR.
        "--tmpdir",
        tmpdir,
    ]

    if request.concurrency_available > 0:
        global_args.extend(["--jobs", "{pants_concurrency}"])

    verbosity_args = [f"-{'v' * pex_subsystem.verbosity}"] if pex_subsystem.verbosity > 0 else []

    warnings_args = [] if pex_subsystem.emit_warnings else ["--no-emit-warnings"]

    # NB: We should always pass `--python-path`, as that tells Pex where to look for interpreters
    # when `--python` isn't an absolute path.
    resolve_args = [
        *cert_args,
        "--python-path",
        os.pathsep.join(pex_env.interpreter_search_paths),
    ]
    # All old-style pex runs take the --pip-version flag, but only certain subcommands of the
    # `pex3` console script do. So if invoked with a subcommand, the caller must selectively
    # set --pip-version only on subcommands that take it.
    pip_version_args = [] if request.subcommand else ["--pip-version", python_setup.pip_version]
    args = [
        *request.subcommand,
        *global_args,
        *verbosity_args,
        *warnings_args,
        *pip_version_args,
        *resolve_args,
        *pex_cli_subsystem.global_args,
        # NB: This comes at the end because it may use `--` passthrough args, # which must come at
        # the end.
        *request.extra_args,
    ]

    complete_pex_env = pex_env.in_sandbox(working_directory=None)
    normalized_argv = complete_pex_env.create_argv(
        pex_pex.exe, *args, python=request.interpreter or bootstrap_python
    )
    env = {
        **complete_pex_env.environment_dict(python_configured=True),
        **python_native_code.subprocess_env_vars,
        **(request.extra_env or {}),
        # If a subcommand is used, we need to use the `pex3` console script.
        **({"PEX_SCRIPT": "pex3"} if request.subcommand else {}),
    }

    return Process(
        normalized_argv,
        description=request.description,
        input_digest=input_digest,
        env=env,
        output_files=request.output_files,
        output_directories=request.output_directories or tuple(),
        append_only_caches={
            **complete_pex_env.append_only_caches,
            **(request.append_only_caches or FrozenDict({})),
        },
        level=request.level,
        concurrency_available=request.concurrency_available,
        cache_scope=request.cache_scope,
    )


def maybe_log_pex_stderr(stderr: bytes, pex_verbosity: int) -> None:
    """Forward Pex's stderr to a Pants logger if conditions are met."""
    log_output = stderr.decode()
    if log_output and "PEXWarning:" in log_output:
        logger.warning("%s", log_output)
    elif log_output and pex_verbosity > 0:
        logger.info("%s", log_output)


def rules():
    return [
        *collect_rules(),
        *external_tool.rules(),
        *pex_environment.rules(),
        *adhoc_binaries.rules(),
        UnionRule(ExportableTool, PexCli),
    ]

1	# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	12✔
5
6	import dataclasses	12✔
7	import logging	12✔
8	import os	12✔
9	from collections.abc import Iterable, Mapping	12✔
10	from dataclasses import dataclass	12✔
11
12	from pants.backend.python.subsystems.python_native_code import PythonNativeCodeSubsystem	12✔
13	from pants.backend.python.subsystems.setup import PythonSetup	12✔
14	from pants.backend.python.util_rules import pex_environment	12✔
15	from pants.backend.python.util_rules.pex_environment import (	12✔
16	PexEnvironment,
17	PexSubsystem,
18	PythonExecutable,
19	)
20	from pants.core.goals.resolves import ExportableTool	12✔
21	from pants.core.util_rules import adhoc_binaries, external_tool	12✔
22	from pants.core.util_rules.adhoc_binaries import PythonBuildStandaloneBinary	12✔
23	from pants.core.util_rules.external_tool import (	12✔
24	DownloadedExternalTool,
25	TemplatedExternalTool,
26	download_external_tool,
27	)
28	from pants.engine.fs import CreateDigest, Digest, Directory, MergeDigests	12✔
29	from pants.engine.internals.selectors import concurrently	12✔
30	from pants.engine.intrinsics import create_digest, merge_digests	12✔
31	from pants.engine.platform import Platform	12✔
32	from pants.engine.process import Process, ProcessCacheScope	12✔
33	from pants.engine.rules import collect_rules, rule	12✔
34	from pants.engine.unions import UnionRule	12✔
35	from pants.option.global_options import GlobalOptions, ca_certs_path_to_file_content	12✔
36	from pants.option.option_types import ArgsListOption	12✔
37	from pants.util.frozendict import FrozenDict	12✔
38	from pants.util.logging import LogLevel	12✔
39	from pants.util.strutil import softwrap	12✔
40
41	logger = logging.getLogger(__name__)	12✔
42
43
44	_PEX_VERSION = "v2.93.2"	12✔
45	_PEX_BINARY_HASH = "75222f01b34cbe2bbe0aa9564dbf554d10dd22875ce59b6ea0c16665c92adc33"	12✔
46	_PEX_BINARY_SIZE = 5116653	12✔
47
48
49	class PexCli(TemplatedExternalTool):	12✔
50	options_scope = "pex-cli"	12✔
51	name = "pex"	12✔
52	help = "The PEX (Python EXecutable) tool (https://github.com/pex-tool/pex)."	12✔
53
54	default_version = _PEX_VERSION	12✔
55	default_url_template = "https://github.com/pex-tool/pex/releases/download/{version}/pex"	12✔
56	version_constraints = ">=2.76.0,<3.0"	12✔
57
58	# extra args to be passed to the pex tool; note that they
59	# are going to apply to all invocations of the pex tool.
60	global_args = ArgsListOption(	12✔
61	example="--check=error --no-compile",
62	extra_help=softwrap(
63	"""
64	Note that these apply to all invocations of the pex tool, including building `pex_binary`
65	targets, preparing `python_test` targets to run, and generating lockfiles.
66	"""
67	),
68	)
69
70	default_known_versions = [	12✔
71	f"{_PEX_VERSION}\|{platform}\|{_PEX_BINARY_HASH}\|{_PEX_BINARY_SIZE}"
72	for platform in ["macos_x86_64", "macos_arm64", "linux_x86_64", "linux_arm64"]
73	]
74
75
76	@dataclass(frozen=True)	12✔
77	class PexCliProcess:	12✔
78	interpreter: PythonExecutable \| PythonBuildStandaloneBinary \| None	12✔
79	subcommand: tuple[str, ...]	12✔
80	extra_args: tuple[str, ...]	12✔
81	description: str = dataclasses.field(compare=False)	12✔
82	additional_input_digest: Digest \| None	12✔
83	extra_env: FrozenDict[str, str] \| None	12✔
84	append_only_caches: FrozenDict[str, str] \| None	12✔
85	output_files: tuple[str, ...] \| None	12✔
86	output_directories: tuple[str, ...] \| None	12✔
87	level: LogLevel	12✔
88	concurrency_available: int	12✔
89	cache_scope: ProcessCacheScope	12✔
90
91	def __init__(	12✔
92	self,
93	*,
94	interpreter: PythonExecutable \| PythonBuildStandaloneBinary \| None = None,
95	subcommand: Iterable[str],
96	extra_args: Iterable[str],
97	description: str,
98	additional_input_digest: Digest \| None = None,
99	extra_env: Mapping[str, str] \| None = None,
100	append_only_caches: Mapping[str, str] \| None = None,
101	output_files: Iterable[str] \| None = None,
102	output_directories: Iterable[str] \| None = None,
103	level: LogLevel = LogLevel.INFO,
104	concurrency_available: int = 0,
105	cache_scope: ProcessCacheScope = ProcessCacheScope.SUCCESSFUL,
106	) -> None:
107	object.__setattr__(self, "interpreter", interpreter)	12✔
108	object.__setattr__(self, "subcommand", tuple(subcommand))	12✔
109	object.__setattr__(self, "extra_args", tuple(extra_args))	12✔
110	object.__setattr__(self, "description", description)	12✔
111	object.__setattr__(self, "additional_input_digest", additional_input_digest)	12✔
112	object.__setattr__(self, "extra_env", FrozenDict(extra_env) if extra_env else None)	12✔
113	object.__setattr__(	12✔
114	self,
115	"append_only_caches",
116	FrozenDict(append_only_caches) if append_only_caches else None,
117	)
118	object.__setattr__(self, "output_files", tuple(output_files) if output_files else None)	12✔
119	object.__setattr__(	12✔
120	self, "output_directories", tuple(output_directories) if output_directories else None
121	)
122	object.__setattr__(self, "level", level)	12✔
123	object.__setattr__(self, "concurrency_available", concurrency_available)	12✔
124	object.__setattr__(self, "cache_scope", cache_scope)	12✔
125
126	self.__post_init__()	12✔
127
128	def __post_init__(self) -> None:	12✔
129	if "--pex-root-path" in self.extra_args:	12✔
UNCOV 130	raise ValueError("`--pex-root` flag not allowed. We set its value for you.")	×
131
132
133	class PexPEX(DownloadedExternalTool):	12✔
134	"""The Pex PEX binary."""
135
136
137	@rule	12✔
138	async def download_pex_pex(pex_cli: PexCli, platform: Platform) -> PexPEX:	12✔
139	pex_pex = await download_external_tool(pex_cli.get_request(platform))	12✔
140	return PexPEX(digest=pex_pex.digest, exe=pex_pex.exe)	12✔
141
142
143	@rule	12✔
144	async def setup_pex_cli_process(	12✔
145	request: PexCliProcess,
146	pex_pex: PexPEX,
147	pex_env: PexEnvironment,
148	bootstrap_python: PythonBuildStandaloneBinary,
149	python_native_code: PythonNativeCodeSubsystem.EnvironmentAware,
150	global_options: GlobalOptions,
151	pex_subsystem: PexSubsystem,
152	pex_cli_subsystem: PexCli,
153	python_setup: PythonSetup,
154	) -> Process:
155	tmpdir = ".tmp"	12✔
156	gets = [create_digest(CreateDigest([Directory(tmpdir)]))]	12✔
157
158	cert_args = []	12✔
159	if global_options.ca_certs_path:	12✔
160	ca_certs_fc = ca_certs_path_to_file_content(global_options.ca_certs_path)	1✔
161	gets.append(create_digest(CreateDigest((ca_certs_fc,))))	1✔
162	cert_args = ["--cert", ca_certs_fc.path]	1✔
163
164	digests_to_merge = [pex_pex.digest]	12✔
165	digests_to_merge.extend(await concurrently(gets))	12✔
166	if request.additional_input_digest:	12✔
167	digests_to_merge.append(request.additional_input_digest)	12✔
168	input_digest = await merge_digests(MergeDigests(digests_to_merge))	12✔
169
170	global_args = [	12✔
171	# Ensure Pex and its subprocesses create temporary files in the process execution
172	# sandbox. It may make sense to do this generally for Processes, but in the short term we
173	# have known use cases where /tmp is too small to hold large wheel downloads Pex is asked to
174	# perform. Making the TMPDIR local to the sandbox allows control via
175	# --local-execution-root-dir for the local case and should work well with remote cases where
176	# a remoting implementation has to allow for processes producing large binaries in a
177	# sandbox to support reasonable workloads. Communicating TMPDIR via --tmpdir instead of via
178	# environment variable allows Pex to absolutize the path ensuring subprocesses that change
179	# CWD can find the TMPDIR.
180	"--tmpdir",
181	tmpdir,
182	]
183
184	if request.concurrency_available > 0:	12✔
185	global_args.extend(["--jobs", "{pants_concurrency}"])	12✔
186
187	verbosity_args = [f"-{'v' * pex_subsystem.verbosity}"] if pex_subsystem.verbosity > 0 else []	12✔
188
189	warnings_args = [] if pex_subsystem.emit_warnings else ["--no-emit-warnings"]	12✔
190
191	# NB: We should always pass `--python-path`, as that tells Pex where to look for interpreters
192	# when `--python` isn't an absolute path.
193	resolve_args = [	12✔
194	*cert_args,
195	"--python-path",
196	os.pathsep.join(pex_env.interpreter_search_paths),
197	]
198	# All old-style pex runs take the --pip-version flag, but only certain subcommands of the
199	# `pex3` console script do. So if invoked with a subcommand, the caller must selectively
200	# set --pip-version only on subcommands that take it.
201	pip_version_args = [] if request.subcommand else ["--pip-version", python_setup.pip_version]	12✔
202	args = [	12✔
203	*request.subcommand,
204	*global_args,
205	*verbosity_args,
206	*warnings_args,
207	*pip_version_args,
208	*resolve_args,
209	*pex_cli_subsystem.global_args,
210	# NB: This comes at the end because it may use `--` passthrough args, # which must come at
211	# the end.
212	*request.extra_args,
213	]
214
215	complete_pex_env = pex_env.in_sandbox(working_directory=None)	12✔
216	normalized_argv = complete_pex_env.create_argv(	12✔
217	pex_pex.exe, *args, python=request.interpreter or bootstrap_python
218	)
219	env = {	12✔
220	**complete_pex_env.environment_dict(python_configured=True),
221	**python_native_code.subprocess_env_vars,
222	**(request.extra_env or {}),
223	# If a subcommand is used, we need to use the `pex3` console script.
224	**({"PEX_SCRIPT": "pex3"} if request.subcommand else {}),
225	}
226
227	return Process(	12✔
228	normalized_argv,
229	description=request.description,
230	input_digest=input_digest,
231	env=env,
232	output_files=request.output_files,
233	output_directories=request.output_directories or tuple(),
234	append_only_caches={
235	**complete_pex_env.append_only_caches,
236	**(request.append_only_caches or FrozenDict({})),
237	},
238	level=request.level,
239	concurrency_available=request.concurrency_available,
240	cache_scope=request.cache_scope,
241	)
242
243
244	def maybe_log_pex_stderr(stderr: bytes, pex_verbosity: int) -> None:	12✔
245	"""Forward Pex's stderr to a Pants logger if conditions are met."""
246	log_output = stderr.decode()	12✔
247	if log_output and "PEXWarning:" in log_output:	12✔
248	logger.warning("%s", log_output)	5✔
249	elif log_output and pex_verbosity > 0:	12✔
UNCOV 250	logger.info("%s", log_output)	×
251
252
253	def rules():	12✔
254	return [	12✔
255	*collect_rules(),
256	*external_tool.rules(),
257	*pex_environment.rules(),
258	*adhoc_binaries.rules(),
259	UnionRule(ExportableTool, PexCli),
260	]

pantsbuild / pants / 25405422172

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