• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

pantsbuild / pants / 26260209689

21 May 2026 11:59PM UTC coverage: 75.453% (-15.7%) from 91.156%
26260209689

Pull #23365

github

web-flow
Merge 5fe873b58 into 7ea655ba0
Pull Request #23365: uv.lock -> pex optimization

5 of 16 new or added lines in 1 file covered. (31.25%)

10118 existing lines in 378 files now uncovered.

54669 of 72454 relevant lines covered (75.45%)

2.31 hits per line

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

50.83
/src/python/pants/backend/python/util_rules/uv.py
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
5✔
5

6
import hashlib
5✔
7
import logging
5✔
8
import os
5✔
9
import shlex
5✔
10
import tomllib
5✔
11
from collections.abc import Iterable
5✔
12
from dataclasses import dataclass
5✔
13
from textwrap import dedent  # noqa: PNT20
5✔
14
from typing import ClassVar, cast
5✔
15

16
from pants.backend.python.subsystems import uv as uv_subsystem
5✔
17
from pants.backend.python.subsystems.python_native_code import PythonNativeCodeSubsystem
5✔
18
from pants.backend.python.subsystems.uv import (
5✔
19
    DownloadedUv,
20
    Uv,
21
)
22
from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints
5✔
23
from pants.backend.python.util_rules.lockfile_metadata import (
5✔
24
    LockfileFormat,
25
    PythonLockfileMetadataV8,
26
)
27
from pants.backend.python.util_rules.pex_environment import PythonExecutable
5✔
28
from pants.backend.python.util_rules.pex_requirements import (
5✔
29
    LoadedLockfile,
30
)
31
from pants.base.build_root import BuildRoot
5✔
32
from pants.core.util_rules import system_binaries
5✔
33
from pants.core.util_rules.env_vars import environment_vars_subset
5✔
34
from pants.core.util_rules.subprocess_environment import SubprocessEnvironmentVars
5✔
35
from pants.core.util_rules.system_binaries import BashBinary, RealpathBinary
5✔
36
from pants.engine.env_vars import EnvironmentVarsRequest
5✔
37
from pants.engine.fs import (
5✔
38
    CreateDigest,
39
    Digest,
40
    FileContent,
41
    MergeDigests,
42
)
43
from pants.engine.intrinsics import (
5✔
44
    create_digest,
45
    get_digest_contents,
46
    merge_digests,
47
)
48
from pants.engine.process import (
5✔
49
    Process,
50
    ProcessResult,
51
    execute_process_or_raise,
52
)
53
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
5✔
54
from pants.util.docutil import bin_name
5✔
55
from pants.util.frozendict import FrozenDict
5✔
56
from pants.util.logging import LogLevel
5✔
57
from pants.util.strutil import softwrap
5✔
58

59
logger = logging.getLogger(__name__)
5✔
60

61

62
@dataclass(frozen=True)
5✔
63
class VenvFromUvLockfileRequest:
5✔
64
    """Request to install all packages from a uv lockfile into a virtualenv."""
65

66
    lockfile: LoadedLockfile
5✔
67
    python: PythonExecutable
5✔
68

69

70
@dataclass(frozen=True)
5✔
71
class FindLinksFromUvLockfileRequest:
5✔
72
    """Request to export a subset of a uv lockfile as a find-links HTML digest for pex."""
73

74
    lockfile: LoadedLockfile
5✔
75
    req_strings: tuple[str, ...]
5✔
76

77

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

82
    cache_name: ClassVar[str] = "venv_cache"
5✔
83
    cache_dir: ClassVar[str] = f".cache/{cache_name}/"
5✔
84

85
    venv_path_suffix: str
5✔
86

87
    def relpath(self) -> str:
5✔
88
        # The path to the venv in any sandbox that has the venv_cache append-only cache.
UNCOV
89
        return os.path.join(self.cache_dir, self.venv_path_suffix)
×
90

91
    @classmethod
5✔
92
    def append_only_caches(cls) -> FrozenDict[str, str]:
5✔
UNCOV
93
        return FrozenDict({cls.cache_name: cls.cache_dir})
×
94

95

96
@dataclass(frozen=True)
5✔
97
class UvEnvironment:
5✔
98
    env: FrozenDict[str, str]
5✔
99

100

101
@rule
5✔
102
async def get_uv_environment(
5✔
103
    subprocess_env_vars: SubprocessEnvironmentVars,
104
    uv_env_aware: Uv.EnvironmentAware,
105
    python_native_code: PythonNativeCodeSubsystem.EnvironmentAware,
106
) -> UvEnvironment:
UNCOV
107
    path = os.pathsep.join(uv_env_aware.path)
×
UNCOV
108
    subprocess_env_dict = dict(subprocess_env_vars.vars)
×
109

110
    extra_env = await environment_vars_subset(
×
111
        EnvironmentVarsRequest(uv_env_aware.extra_env_vars), **implicitly()
112
    )
113

UNCOV
114
    if "PATH" in subprocess_env_dict:
×
UNCOV
115
        path = os.pathsep.join([path, subprocess_env_dict.pop("PATH")])
×
UNCOV
116
    return UvEnvironment(
×
117
        env=FrozenDict(
118
            {
119
                **extra_env,
120
                "PATH": path,
121
                **subprocess_env_dict,
122
                **python_native_code.subprocess_env_vars,
123
            }
124
        )
125
    )
126

127

128
# A utility function to generate a transient, minimal pyproject.toml for uv to interact with.
129
# The synthetic project name (pants-lockfile-for-*) must not collide with any real requirement.
130
# uv will include this project as a virtual package in the lockfile, and we set package = false,
131
# so it won't try to install it.
132
def generate_pyproject_toml(
5✔
133
    resolve: str,
134
    ics: InterpreterConstraints,
135
    reqs: Iterable[str],
136
    subset_group: Iterable[str] | None = None,
137
) -> str:
UNCOV
138
    def escape_double_quotes(s: str) -> str:
×
NEW
139
        return s.replace('"', '\\"')
×
140

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

UNCOV
144
    result = dedent(
×
145
        """
146
        [project]
147
        name = "pants-lockfile-for-{resolve}"
148
        version = "0.0.0"
149
        requires-python = "{requires_python}"
150
        dependencies = [
151
        {deps_lines}
152
        ]
153

154
        [tool.uv]
155
        package = false
156
        """
157
    ).format(resolve=resolve, requires_python=requires_python, deps_lines=deps_lines)
158

NEW
159
    if subset_group is not None:
×
NEW
160
        group_lines = "\n".join(
×
161
            f'    "{escape_double_quotes(r)}",' for r in sorted(subset_group)
162
        )
NEW
163
        result += dedent(
×
164
            """
165
            [dependency-groups]
166
            pants-subset = [
167
            {group_lines}
168
            ]
169
            """
170
        ).format(group_lines=group_lines)
171

UNCOV
172
    return result
×
173

174

175
@rule
5✔
176
async def create_venv_repository_from_uv_lockfile(
5✔
177
    request: VenvFromUvLockfileRequest,
178
    downloaded_uv: DownloadedUv,
179
    uv_env: UvEnvironment,
180
    bash_binary: BashBinary,
181
    realpath_binary: RealpathBinary,
182
    buildroot: BuildRoot,
183
) -> VenvRepository:
184
    """Install all packages from a uv lockfile into a virtualenv."""
UNCOV
185
    if request.lockfile.lockfile_format != LockfileFormat.UV:
×
UNCOV
186
        raise ValueError(f"Expected a uv lockfile, got {request.lockfile.lockfile_format}")
×
UNCOV
187
    if request.lockfile.metadata is None:
×
UNCOV
188
        raise ValueError(
×
189
            softwrap(
190
                f"""
191
                Cannot install from uv lockfile {request.lockfile.lockfile_path}: metadata is
192
                missing. uv lockfiles must have a separate metadata file. Please regenerate
193
                the lockfile by running `{bin_name()} generate-lockfiles`.
194
                """
195
            )
196
        )
UNCOV
197
    metadata: PythonLockfileMetadataV8 = cast(PythonLockfileMetadataV8, request.lockfile.metadata)
×
198

NEW
199
    pyproject_content = generate_pyproject_toml(
×
200
        metadata.resolve,
201
        metadata.valid_for_interpreter_constraints,
202
        tuple(str(req) for req in metadata.requirements),
203
    )
204

UNCOV
205
    uv_config_digest, uv_lock_contents = await concurrently(
×
206
        create_digest(
207
            CreateDigest(
208
                (
209
                    FileContent("pyproject.toml", pyproject_content.encode()),
210
                    # Nothing to put in config right now, but we need it to be present.
211
                    FileContent("uv.toml", b""),
212
                )
213
            )
214
        ),
215
        get_digest_contents(request.lockfile.lockfile_digest),
216
    )
UNCOV
217
    uv_lock_digest = await create_digest(
×
218
        CreateDigest([FileContent("uv.lock", uv_lock_contents[0].content)])
219
    )
220

UNCOV
221
    input_digest = await merge_digests(
×
222
        MergeDigests(
223
            (
224
                downloaded_uv.digest,
225
                uv_config_digest,
226
                uv_lock_digest,
227
            )
228
        )
229
    )
230

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

NEW
241
    uv_cmd = shlex.join(
×
242
        (
243
            *downloaded_uv.args(),
244
            "sync",
245
            "--frozen",
246
            "--no-install-project",
247
            # TODO: extras can conflict, so we might need to be more selective.
248
            "--all-extras",
249
            "--no-progress",
250
            "--python",
251
            request.python.path,
252
        )
253
    )
254
    # We use `realpath` to resolve the named cache symlink to an absolute path in whatever
255
    # environment this process runs in. This gives uv a stable absolute path for the venv
256
    # so that any entry point scripts it creates exec a valid path that doesn't reference
257
    # the sandbox.
NEW
258
    script = dedent(
×
259
        f"""\
260
        cache_root="$({realpath_binary.path} {shlex.quote(venv_repository.cache_dir)})"
261
        UV_PROJECT_ENVIRONMENT="${{cache_root}}/{venv_repository.venv_path_suffix}" {uv_cmd}
262
        """
263
    )
NEW
264
    await execute_process_or_raise(
×
265
        **implicitly(
266
            Process(
267
                argv=(bash_binary.path, "-c", script),
268
                input_digest=input_digest,
269
                env=uv_env.env,
270
                append_only_caches={
271
                    **downloaded_uv.append_only_caches(),
272
                    **venv_repository.append_only_caches(),
273
                },
274
                level=LogLevel.INFO,
275
                description=f"Create venv from uv lockfile at {request.lockfile.lockfile_path}",
276
                # TODO: We might need to set cache_scope=ProcessCacheScope.PER_SESSION if
277
                #  running in a non-local environment (e.g., a docker environment), as we're
278
                #  less sure that a venv created by a previous run still exists. For example
279
                #  the docker container might have been recreated. However this impacts performance
280
                #  in the overwhelmingly common local run case. Alternatively we can look at
281
                #  creating the uv venv in the same Process as whatever is consuming it, which
282
                #  would ensure availability even in a completely ephemeral remote environment
283
                #  where we can't guarantee that the venv exists even in the same Pants run.
284
                #  But that is a more general issue and a much bigger change.
285
            )
286
        )
287
    )
288

UNCOV
289
    return venv_repository
×
290

291

292
def _pylock_to_find_links_html(pylock_content: str) -> str:
5✔
UNCOV
293
    data = tomllib.loads(pylock_content)
×
UNCOV
294
    links = []
×
UNCOV
295
    for pkg in data.get("packages", []):
×
UNCOV
296
        for wheel in pkg.get("wheels", []):
×
UNCOV
297
            url = wheel["url"]
×
UNCOV
298
            sha256 = wheel["hashes"]["sha256"]
×
UNCOV
299
            filename = url.rsplit("/", 1)[-1]
×
UNCOV
300
            links.append(f'<a href="{url}#sha256={sha256}">{filename}</a>')
×
UNCOV
301
        if "sdist" in pkg:
×
UNCOV
302
            sdist = pkg["sdist"]
×
UNCOV
303
            url = sdist["url"]
×
UNCOV
304
            sha256 = sdist["hashes"]["sha256"]
×
NEW
305
            filename = url.rsplit("/", 1)[-1]
×
UNCOV
306
            links.append(f'<a href="{url}#sha256={sha256}">{filename}</a>')
×
UNCOV
307
    return "<html><body>\n" + "\n".join(links) + "\n</body></html>\n"
×
308

309

310
@rule
5✔
311
async def create_find_links_from_uv_lockfile(
5✔
312
    request: FindLinksFromUvLockfileRequest,
313
    downloaded_uv: DownloadedUv,
314
    uv_env: UvEnvironment,
315
) -> Digest:
316
    """Export a subset of a uv lockfile as a find-links HTML digest for pex."""
UNCOV
317
    if request.lockfile.lockfile_format != LockfileFormat.UV:
×
UNCOV
318
        raise ValueError(f"Expected a uv lockfile, got {request.lockfile.lockfile_format}")
×
UNCOV
319
    if request.lockfile.metadata is None:
×
UNCOV
320
        raise ValueError(
×
321
            softwrap(
322
                f"""
323
                Cannot export from uv lockfile {request.lockfile.lockfile_path}: metadata is
324
                missing. uv lockfiles must have a separate metadata file. Please regenerate
325
                the lockfile by running `{bin_name()} generate-lockfiles`.
326
                """
327
            )
328
        )
UNCOV
329
    metadata: PythonLockfileMetadataV8 = cast(PythonLockfileMetadataV8, request.lockfile.metadata)
×
330

UNCOV
331
    pyproject_content = generate_pyproject_toml(
×
332
        metadata.resolve,
333
        metadata.valid_for_interpreter_constraints,
334
        tuple(str(req) for req in metadata.requirements),
335
        subset_group=request.req_strings,
336
    )
337

UNCOV
338
    uv_config_digest, uv_lock_contents = await concurrently(
×
339
        create_digest(
340
            CreateDigest(
341
                (
342
                    FileContent("pyproject.toml", pyproject_content.encode()),
343
                    FileContent("uv.toml", b""),
344
                )
345
            )
346
        ),
347
        get_digest_contents(request.lockfile.lockfile_digest),
348
    )
UNCOV
349
    uv_lock_digest = await create_digest(
×
350
        CreateDigest([FileContent("uv.lock", uv_lock_contents[0].content)])
351
    )
UNCOV
352
    input_digest = await merge_digests(
×
353
        MergeDigests((downloaded_uv.digest, uv_config_digest, uv_lock_digest))
354
    )
355

UNCOV
356
    result: ProcessResult = await execute_process_or_raise(
×
357
        **implicitly(
358
            Process(
359
                argv=(
360
                    *downloaded_uv.args(),
361
                    "export",
362
                    "--only-group",
363
                    "pants-subset",
364
                    "--frozen",
365
                    "--format",
366
                    "pylock.toml",
367
                    "--no-emit-project",
368
                    "--no-progress",
369
                ),
370
                input_digest=input_digest,
371
                env=uv_env.env,
372
                description=f"Export subset from uv lockfile at {request.lockfile.lockfile_path}",
373
                level=LogLevel.DEBUG,
374
            )
375
        )
376
    )
377

UNCOV
378
    html = _pylock_to_find_links_html(result.stdout.decode())
×
UNCOV
379
    return await create_digest(CreateDigest([FileContent("find_links.html", html.encode())]))
×
380

381

382
def rules():
5✔
383
    return [
5✔
384
        *collect_rules(),
385
        *uv_subsystem.rules(),
386
        *system_binaries.rules(),
387
    ]
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc