25443604553

Committed 06 May 2026 03:05PM UTC coverage: 92.879% (-0.04%) from 92.915%

Build # 25443604553

Build Type

push

github

Committed by

web-flow

Commit Message

[pants_ng] Scaffolding for a pants_ng mode. (#23319)

In this mode the command line is parsed as an
NG invocation, and dispatched appropriately.

Of course at the moment there are no
implementations to dispatch to. That will follow.

This does expose a new option, `pants_ng` to users. 
There is a big warning not to set it, but we're not trying
to hide that we're working on a new thing, so I am
comfortable with this.

Coverage Stats

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

1294 existing lines in 76 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

96.15

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

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

from __future__ import annotations

import hashlib
import logging
import os
import shlex
from collections.abc import Iterable
from dataclasses import dataclass
from textwrap import dedent  # noqa: PNT20
from typing import ClassVar, cast

from pants.backend.python.subsystems import uv as uv_subsystem
from pants.backend.python.subsystems.python_native_code import PythonNativeCodeSubsystem
from pants.backend.python.subsystems.uv import (
    DownloadedUv,
    Uv,
)
from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints
from pants.backend.python.util_rules.lockfile_metadata import (
    LockfileFormat,
    PythonLockfileMetadataV8,
)
from pants.backend.python.util_rules.pex_environment import PythonExecutable
from pants.backend.python.util_rules.pex_requirements import (
    LoadedLockfile,
)
from pants.base.build_root import BuildRoot
from pants.core.util_rules import system_binaries
from pants.core.util_rules.subprocess_environment import SubprocessEnvironmentVars
from pants.core.util_rules.system_binaries import BashBinary, RealpathBinary
from pants.engine.fs import (
    CreateDigest,
    FileContent,
    MergeDigests,
)
from pants.engine.intrinsics import (
    create_digest,
    get_digest_contents,
    merge_digests,
)
from pants.engine.process import (
    Process,
    execute_process_or_raise,
)
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
from pants.util.docutil import bin_name
from pants.util.frozendict import FrozenDict
from pants.util.logging import LogLevel
from pants.util.strutil import softwrap

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class VenvFromUvLockfileRequest:
    """Request to install all packages from a uv lockfile into a virtualenv."""

    lockfile: LoadedLockfile
    python: PythonExecutable


@dataclass(frozen=True)
class VenvRepository:
    """A virtualenv directory that Pex can use as a --venv-repository."""

    cache_name: ClassVar[str] = "venv_cache"
    cache_dir: ClassVar[str] = f".cache/{cache_name}/"

    venv_path_suffix: str

    def relpath(self) -> str:
        # The path to the venv in any sandbox that has the venv_cache append-only cache.
        return os.path.join(self.cache_dir, self.venv_path_suffix)

    @classmethod
    def append_only_caches(cls) -> FrozenDict[str, str]:
        return FrozenDict({cls.cache_name: cls.cache_dir})


@dataclass(frozen=True)
class UvEnvironment:
    env: FrozenDict[str, str]


@rule
async def get_uv_environment(
    subprocess_env_vars: SubprocessEnvironmentVars,
    uv_env_aware: Uv.EnvironmentAware,
    python_native_code: PythonNativeCodeSubsystem.EnvironmentAware,
) -> UvEnvironment:
    path = os.pathsep.join(uv_env_aware.path)
    subprocess_env_dict = dict(subprocess_env_vars.vars)

    if "PATH" in subprocess_env_dict:
        path = os.pathsep.join([path, subprocess_env_dict.pop("PATH")])
    return UvEnvironment(
        env=FrozenDict(
            {
                "PATH": path,
                **subprocess_env_dict,
                **python_native_code.subprocess_env_vars,
            }
        )
    )


# A utility function to generate a transient, minimal pyproject.toml for uv to interact with.
# The synthetic project name (pants-lockfile-for-*) must not collide with any real requirement.
# uv will include this project as a virtual package in the lockfile, and we set package = false,
# so it won't try to install it.
def generate_pyproject_toml(resolve: str, ics: InterpreterConstraints, reqs: Iterable[str]) -> str:
    def escape_double_quotes(s: str) -> str:
        return s.replace('"', '\\"')

    requires_python = ",".join(str(constraint.specifier) for constraint in ics)
    deps_lines = "\n".join(f'    "{escape_double_quotes(r)}",' for r in sorted(reqs))

    return dedent(
        """
        [project]
        name = "pants-lockfile-for-{resolve}"
        version = "0.0.0"
        requires-python = "{requires_python}"
        dependencies = [
        {deps_lines}
        ]

        [tool.uv]
        package = false
        """
    ).format(resolve=resolve, requires_python=requires_python, deps_lines=deps_lines)


@rule
async def create_venv_repository_from_uv_lockfile(
    request: VenvFromUvLockfileRequest,
    downloaded_uv: DownloadedUv,
    uv_env: UvEnvironment,
    bash_binary: BashBinary,
    realpath_binary: RealpathBinary,
    buildroot: BuildRoot,
) -> VenvRepository:
    """Install all packages from a uv lockfile into a virtualenv."""
    if request.lockfile.lockfile_format != LockfileFormat.UV:
        raise ValueError(f"Expected a uv lockfile, got {request.lockfile.lockfile_format}")
    if request.lockfile.metadata is None:
        raise ValueError(
            softwrap(
                f"""
                Cannot install from uv lockfile {request.lockfile.lockfile_path}: metadata is
                missing. uv lockfiles must have a separate metadata file. Please regenerate
                the lockfile by running `{bin_name()} generate-lockfiles`.
                """
            )
        )
    metadata: PythonLockfileMetadataV8 = cast(PythonLockfileMetadataV8, request.lockfile.metadata)

    pyproject_content = generate_pyproject_toml(
        metadata.resolve,
        metadata.valid_for_interpreter_constraints,
        tuple(str(req) for req in metadata.requirements),
    )

    uv_config_digest, uv_lock_contents = await concurrently(
        create_digest(
            CreateDigest(
                (
                    FileContent("pyproject.toml", pyproject_content.encode()),
                    # Nothing to put in config right now, but we need it to be present.
                    FileContent("uv.toml", b""),
                )
            )
        ),
        get_digest_contents(request.lockfile.lockfile_digest),
    )
    uv_lock_digest = await create_digest(
        CreateDigest([FileContent("uv.lock", uv_lock_contents[0].content)])
    )

    input_digest = await merge_digests(
        MergeDigests(
            (
                downloaded_uv.digest,
                uv_config_digest,
                uv_lock_digest,
            )
        )
    )

    buildroot_entropy = hashlib.sha256(buildroot.path.encode()).hexdigest()
    venv_repository = VenvRepository(
        # We maintain one cached venv per buildroot+interpreter+resolve. uv will efficiently incrementally
        # update the venv as the lockfile changes, and will handle concurrency of `uv sync` with
        # appropriate locking.
        venv_path_suffix=os.path.join(
            buildroot_entropy, metadata.resolve, request.python.fingerprint
        )
    )

    uv_cmd = shlex.join(
        (
            *downloaded_uv.args(),
            "sync",
            "--frozen",
            "--no-install-project",
            # TODO: extras can conflict, so we might need to be more selective.
            "--all-extras",
            "--no-progress",
            "--python",
            request.python.path,
        )
    )
    # We use `realpath` to resolve the named cache symlink to an absolute path in whatever
    # environment this process runs in. This gives uv a stable absolute path for the venv
    # so that any entry point scripts it creates exec a valid path that doesn't reference
    # the sandbox.
    script = dedent(
        f"""\
        cache_root="$({realpath_binary.path} {shlex.quote(venv_repository.cache_dir)})"
        UV_PROJECT_ENVIRONMENT="${{cache_root}}/{venv_repository.venv_path_suffix}" {uv_cmd}
        """
    )
    await execute_process_or_raise(
        **implicitly(
            Process(
                argv=(bash_binary.path, "-c", script),
                input_digest=input_digest,
                env=uv_env.env,
                append_only_caches={
                    **downloaded_uv.append_only_caches(),
                    **venv_repository.append_only_caches(),
                },
                level=LogLevel.INFO,
                description=f"Create venv from uv lockfile at {request.lockfile.lockfile_path}",
                # TODO: We might need to set cache_scope=ProcessCacheScope.PER_SESSION if
                #  running in a non-local environment (e.g., a docker environment), as we're
                #  less sure that a venv created by a previous run still exists. For example
                #  the docker container might have been recreated. However this impacts performance
                #  in the overwhelmingly common local run case. Alternatively we can look at
                #  creating the uv venv in the same Process as whatever is consuming it, which
                #  would ensure availability even in a completely ephemeral remote environment
                #  where we can't guarantee that the venv exists even in the same Pants run.
                #  But that is a more general issue and a much bigger change.
            )
        )
    )

    return venv_repository


def rules():
    return [
        *collect_rules(),
        *uv_subsystem.rules(),
        *system_binaries.rules(),
    ]

1	# Copyright 2026 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 hashlib	12✔
7	import logging	12✔
8	import os	12✔
9	import shlex	12✔
10	from collections.abc import Iterable	12✔
11	from dataclasses import dataclass	12✔
12	from textwrap import dedent # noqa: PNT20	12✔
13	from typing import ClassVar, cast	12✔
14
15	from pants.backend.python.subsystems import uv as uv_subsystem	12✔
16	from pants.backend.python.subsystems.python_native_code import PythonNativeCodeSubsystem	12✔
17	from pants.backend.python.subsystems.uv import (	12✔
18	DownloadedUv,
19	Uv,
20	)
21	from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints	12✔
22	from pants.backend.python.util_rules.lockfile_metadata import (	12✔
23	LockfileFormat,
24	PythonLockfileMetadataV8,
25	)
26	from pants.backend.python.util_rules.pex_environment import PythonExecutable	12✔
27	from pants.backend.python.util_rules.pex_requirements import (	12✔
28	LoadedLockfile,
29	)
30	from pants.base.build_root import BuildRoot	12✔
31	from pants.core.util_rules import system_binaries	12✔
32	from pants.core.util_rules.subprocess_environment import SubprocessEnvironmentVars	12✔
33	from pants.core.util_rules.system_binaries import BashBinary, RealpathBinary	12✔
34	from pants.engine.fs import (	12✔
35	CreateDigest,
36	FileContent,
37	MergeDigests,
38	)
39	from pants.engine.intrinsics import (	12✔
40	create_digest,
41	get_digest_contents,
42	merge_digests,
43	)
44	from pants.engine.process import (	12✔
45	Process,
46	execute_process_or_raise,
47	)
48	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	12✔
49	from pants.util.docutil import bin_name	12✔
50	from pants.util.frozendict import FrozenDict	12✔
51	from pants.util.logging import LogLevel	12✔
52	from pants.util.strutil import softwrap	12✔
53
54	logger = logging.getLogger(__name__)	12✔
55
56
57	@dataclass(frozen=True)	12✔
58	class VenvFromUvLockfileRequest:	12✔
59	"""Request to install all packages from a uv lockfile into a virtualenv."""
60
61	lockfile: LoadedLockfile	12✔
62	python: PythonExecutable	12✔
63
64
65	@dataclass(frozen=True)	12✔
66	class VenvRepository:	12✔
67	"""A virtualenv directory that Pex can use as a --venv-repository."""
68
69	cache_name: ClassVar[str] = "venv_cache"	12✔
70	cache_dir: ClassVar[str] = f".cache/{cache_name}/"	12✔
71
72	venv_path_suffix: str	12✔
73
74	def relpath(self) -> str:	12✔
75	# The path to the venv in any sandbox that has the venv_cache append-only cache.
UNCOV 76	return os.path.join(self.cache_dir, self.venv_path_suffix)	1✔
77
78	@classmethod	12✔
79	def append_only_caches(cls) -> FrozenDict[str, str]:	12✔
UNCOV 80	return FrozenDict({cls.cache_name: cls.cache_dir})	1✔
81
82
83	@dataclass(frozen=True)	12✔
84	class UvEnvironment:	12✔
85	env: FrozenDict[str, str]	12✔
86
87
88	@rule	12✔
89	async def get_uv_environment(	12✔
90	subprocess_env_vars: SubprocessEnvironmentVars,
91	uv_env_aware: Uv.EnvironmentAware,
92	python_native_code: PythonNativeCodeSubsystem.EnvironmentAware,
93	) -> UvEnvironment:
94	path = os.pathsep.join(uv_env_aware.path)	2✔
95	subprocess_env_dict = dict(subprocess_env_vars.vars)	2✔
96
97	if "PATH" in subprocess_env_dict:	2✔
98	path = os.pathsep.join([path, subprocess_env_dict.pop("PATH")])	×
99	return UvEnvironment(	2✔
100	env=FrozenDict(
101	{
102	"PATH": path,
103	**subprocess_env_dict,
104	**python_native_code.subprocess_env_vars,
105	}
106	)
107	)
108
109
110	# A utility function to generate a transient, minimal pyproject.toml for uv to interact with.
111	# The synthetic project name (pants-lockfile-for-*) must not collide with any real requirement.
112	# uv will include this project as a virtual package in the lockfile, and we set package = false,
113	# so it won't try to install it.
114	def generate_pyproject_toml(resolve: str, ics: InterpreterConstraints, reqs: Iterable[str]) -> str:	12✔
115	def escape_double_quotes(s: str) -> str:	2✔
116	return s.replace('"', '\\"')	2✔
117
118	requires_python = ",".join(str(constraint.specifier) for constraint in ics)	2✔
119	deps_lines = "\n".join(f' "{escape_double_quotes(r)}",' for r in sorted(reqs))	2✔
120
121	return dedent(	2✔
122	"""
123	[project]
124	name = "pants-lockfile-for-{resolve}"
125	version = "0.0.0"
126	requires-python = "{requires_python}"
127	dependencies = [
128	{deps_lines}
129	]
130
131	[tool.uv]
132	package = false
133	"""
134	).format(resolve=resolve, requires_python=requires_python, deps_lines=deps_lines)
135
136
137	@rule	12✔
138	async def create_venv_repository_from_uv_lockfile(	12✔
139	request: VenvFromUvLockfileRequest,
140	downloaded_uv: DownloadedUv,
141	uv_env: UvEnvironment,
142	bash_binary: BashBinary,
143	realpath_binary: RealpathBinary,
144	buildroot: BuildRoot,
145	) -> VenvRepository:
146	"""Install all packages from a uv lockfile into a virtualenv."""
UNCOV 147	if request.lockfile.lockfile_format != LockfileFormat.UV:	1✔
148	raise ValueError(f"Expected a uv lockfile, got {request.lockfile.lockfile_format}")	×
UNCOV 149	if request.lockfile.metadata is None:	1✔
150	raise ValueError(	×
151	softwrap(
152	f"""
153	Cannot install from uv lockfile {request.lockfile.lockfile_path}: metadata is
154	missing. uv lockfiles must have a separate metadata file. Please regenerate
155	the lockfile by running `{bin_name()} generate-lockfiles`.
156	"""
157	)
158	)
UNCOV 159	metadata: PythonLockfileMetadataV8 = cast(PythonLockfileMetadataV8, request.lockfile.metadata)	1✔
160
UNCOV 161	pyproject_content = generate_pyproject_toml(	1✔
162	metadata.resolve,
163	metadata.valid_for_interpreter_constraints,
164	tuple(str(req) for req in metadata.requirements),
165	)
166
UNCOV 167	uv_config_digest, uv_lock_contents = await concurrently(	1✔
168	create_digest(
169	CreateDigest(
170	(
171	FileContent("pyproject.toml", pyproject_content.encode()),
172	# Nothing to put in config right now, but we need it to be present.
173	FileContent("uv.toml", b""),
174	)
175	)
176	),
177	get_digest_contents(request.lockfile.lockfile_digest),
178	)
UNCOV 179	uv_lock_digest = await create_digest(	1✔
180	CreateDigest([FileContent("uv.lock", uv_lock_contents[0].content)])
181	)
182
UNCOV 183	input_digest = await merge_digests(	1✔
184	MergeDigests(
185	(
186	downloaded_uv.digest,
187	uv_config_digest,
188	uv_lock_digest,
189	)
190	)
191	)
192
UNCOV 193	buildroot_entropy = hashlib.sha256(buildroot.path.encode()).hexdigest()	1✔
UNCOV 194	venv_repository = VenvRepository(	1✔
195	# We maintain one cached venv per buildroot+interpreter+resolve. uv will efficiently incrementally
196	# update the venv as the lockfile changes, and will handle concurrency of `uv sync` with
197	# appropriate locking.
198	venv_path_suffix=os.path.join(
199	buildroot_entropy, metadata.resolve, request.python.fingerprint
200	)
201	)
202
UNCOV 203	uv_cmd = shlex.join(	1✔
204	(
205	*downloaded_uv.args(),
206	"sync",
207	"--frozen",
208	"--no-install-project",
209	# TODO: extras can conflict, so we might need to be more selective.
210	"--all-extras",
211	"--no-progress",
212	"--python",
213	request.python.path,
214	)
215	)
216	# We use `realpath` to resolve the named cache symlink to an absolute path in whatever
217	# environment this process runs in. This gives uv a stable absolute path for the venv
218	# so that any entry point scripts it creates exec a valid path that doesn't reference
219	# the sandbox.
UNCOV 220	script = dedent(	1✔
221	f"""\
222	cache_root="$({realpath_binary.path} {shlex.quote(venv_repository.cache_dir)})"
223	UV_PROJECT_ENVIRONMENT="${{cache_root}}/{venv_repository.venv_path_suffix}" {uv_cmd}
224	"""
225	)
UNCOV 226	await execute_process_or_raise(	1✔
227	**implicitly(
228	Process(
229	argv=(bash_binary.path, "-c", script),
230	input_digest=input_digest,
231	env=uv_env.env,
232	append_only_caches={
233	**downloaded_uv.append_only_caches(),
234	**venv_repository.append_only_caches(),
235	},
236	level=LogLevel.INFO,
237	description=f"Create venv from uv lockfile at {request.lockfile.lockfile_path}",
238	# TODO: We might need to set cache_scope=ProcessCacheScope.PER_SESSION if
239	# running in a non-local environment (e.g., a docker environment), as we're
240	# less sure that a venv created by a previous run still exists. For example
241	# the docker container might have been recreated. However this impacts performance
242	# in the overwhelmingly common local run case. Alternatively we can look at
243	# creating the uv venv in the same Process as whatever is consuming it, which
244	# would ensure availability even in a completely ephemeral remote environment
245	# where we can't guarantee that the venv exists even in the same Pants run.
246	# But that is a more general issue and a much bigger change.
247	)
248	)
249	)
250
UNCOV 251	return venv_repository	1✔
252
253
254	def rules():	12✔
255	return [	12✔
256	*collect_rules(),
257	*uv_subsystem.rules(),
258	*system_binaries.rules(),
259	]

pantsbuild / pants / 25443604553

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