26380816428

Committed 25 May 2026 02:57AM UTC coverage: 52.312% (-40.6%) from 92.89%

Build # 26380816428

Build Type

Pull #23368

github

Committed by

web-flow

Commit Message

Merge 7410b48e1 into 7b1060c81

Pull Request Pull Request #23368: Run Linux ARM CI on Depot runners (Cherry-pick of #23363)

Coverage Stats

31807 of 60802 relevant lines covered (52.31%)

1.05 hits per line

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

65.43

/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.env_vars import environment_vars_subset
from pants.core.util_rules.subprocess_environment import SubprocessEnvironmentVars
from pants.core.util_rules.system_binaries import BashBinary, RealpathBinary
from pants.engine.env_vars import EnvironmentVarsRequest
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)

    extra_env = await environment_vars_subset(
        EnvironmentVarsRequest(uv_env_aware.extra_env_vars), **implicitly()
    )

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

pantsbuild / pants / 26380816428

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