25405181773

Committed 05 May 2026 10:12PM UTC coverage: 92.884% (-0.03%) from 92.911%

Build # 25405181773

Build Type

Pull #23320

github

Committed by

web-flow

Commit Message

Merge aa9992c41 into 736246ca9

Pull Request Pull Request #23320: Support for uv lockfiles as an alternative to pex lockfiles (cherry-pick of #23302)

Coverage Stats

443 of 505 new or added lines in 23 files covered. (87.72%)

11 existing lines in 1 file now uncovered.

92029 of 99080 relevant lines covered (92.88%)

4.05 hits per line

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

61.9

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

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

from __future__ import annotations

import itertools
import json
import logging
import tomllib
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any

from packaging.version import Version, parse

from pants.backend.python.util_rules.lockfile_metadata import LockfileFormat
from pants.backend.python.util_rules.pex_requirements import (
    LoadedLockfileRequest,
    Lockfile,
    load_lockfile,
    strip_comments_from_pex_json_lockfile,
)
from pants.base.exceptions import EngineError
from pants.core.goals.generate_lockfiles import LockfileDiff, LockfilePackages, PackageName
from pants.engine.fs import Digest
from pants.engine.intrinsics import get_digest_contents
from pants.engine.rules import implicitly
from pants.util.frozendict import FrozenDict

logger = logging.getLogger(__name__)


@dataclass(frozen=True, order=True)
class PythonRequirementVersion:
    _parsed: Version

    @classmethod
    def parse(cls, version: str) -> PythonRequirementVersion:
        return cls(parse(version))

    def __str__(self) -> str:
        return str(self._parsed)

    def __getattr__(self, key: str) -> Any:
        return getattr(self._parsed, key)


def _pex_lockfile_requirements(
    lockfile_data: Mapping[str, Any] | None, path: str | None = None
) -> LockfilePackages:
    if not lockfile_data:
        return LockfilePackages({})

    try:
        # Setup generators
        locked_resolves = (
            (
                (PackageName(r["project_name"]), PythonRequirementVersion.parse(r["version"]))
                for r in resolve["locked_requirements"]
            )
            for resolve in lockfile_data["locked_resolves"]
        )
        requirements = dict(itertools.chain.from_iterable(locked_resolves))
    except KeyError as e:
        if path:
            logger.warning(f"{path}: Failed to parse lockfile: {e}")

        requirements = {}

    return LockfilePackages(requirements)


def _uv_lockfile_requirements(
    lockfile_data: Mapping[str, Any] | None, path: str | None = None
) -> LockfilePackages:
    if not lockfile_data:
        return LockfilePackages({})

    requirements = {}
    for pkg in lockfile_data.get("package", []):
        try:
            name = pkg["name"]
            version = pkg.get("version")
            # Skip the synthetic virtual package, it's not interesting in diffs.
            if version is not None and not name.startswith("pants-lockfile-for-"):
                requirements[PackageName(name)] = PythonRequirementVersion.parse(version)
        except Exception as e:
            if path:
                logger.warning(f"{path}: Failed to parse package entry in lockfile: {e}")

    return LockfilePackages(requirements)


def _parse_lockfile_packages(
    content: bytes, lockfile_format: LockfileFormat, path: str | None = None
) -> LockfilePackages:
    """Parse the packages from lockfile content according to its format."""
    try:
        match lockfile_format:
            case LockfileFormat.PEX:
                # strip_comments_from_pex_json_lockfile is idempotent, so safe to call on
                # already-stripped content (e.g. when content was loaded via load_lockfile).
                stripped = strip_comments_from_pex_json_lockfile(content)
                data = FrozenDict.deep_freeze(json.loads(stripped))
                return _pex_lockfile_requirements(data, path)
            case LockfileFormat.UV:
                data = FrozenDict.deep_freeze(tomllib.loads(content.decode()))
                return _uv_lockfile_requirements(data, path)
            case LockfileFormat.CONSTRAINTS_DEPRECATED:
                # These can't meaningfully be diffed.
                return LockfilePackages({})
            case _:
                raise ValueError(f"Unrecognized lockfile format: {lockfile_format}")
    except Exception as e:
        if path:
            logger.debug(f"{path}: Failed to parse lockfile contents: {e}")
        return LockfilePackages({})


async def _generate_lockfile_diff(
    digest: Digest, resolve_name: str, path: str, new_format: LockfileFormat
) -> LockfileDiff:
    """Generate a diff between the newly generated lockfile and the existing one on disk.

    Handles all combinations of old vs. new and pex vs. uv lockfile formats.
    """
    new_digest_contents = await get_digest_contents(digest)
    new_content = next(c for c in new_digest_contents if c.path == path).content
    new_packages = _parse_lockfile_packages(new_content, new_format, path)

    old_packages = LockfilePackages({})
    try:
        loaded = await load_lockfile(
            LoadedLockfileRequest(
                Lockfile(
                    url=path,
                    url_description_of_origin="existing lockfile",
                    resolve_name=resolve_name,
                )
            ),
            **implicitly(),
        )
        old_content_entries = await get_digest_contents(loaded.lockfile_digest)
        old_content = next(iter(old_content_entries)).content
        old_packages = _parse_lockfile_packages(old_content, loaded.lockfile_format, path)
    except EngineError:
        # May fail if the file doesn't exist, which is expected the first time a new lockfile
        # is generated.
        pass

    return LockfileDiff.create(
        path=path,
        resolve_name=resolve_name,
        old=old_packages,
        new=new_packages,
    )

1	# Copyright 2022 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 itertools	12✔
7	import json	12✔
8	import logging	12✔
9	import tomllib	12✔
10	from collections.abc import Mapping	12✔
11	from dataclasses import dataclass	12✔
12	from typing import Any	12✔
13
14	from packaging.version import Version, parse	12✔
15
16	from pants.backend.python.util_rules.lockfile_metadata import LockfileFormat	12✔
17	from pants.backend.python.util_rules.pex_requirements import (	12✔
18	LoadedLockfileRequest,
19	Lockfile,
20	load_lockfile,
21	strip_comments_from_pex_json_lockfile,
22	)
23	from pants.base.exceptions import EngineError	12✔
24	from pants.core.goals.generate_lockfiles import LockfileDiff, LockfilePackages, PackageName	12✔
25	from pants.engine.fs import Digest	12✔
26	from pants.engine.intrinsics import get_digest_contents	12✔
27	from pants.engine.rules import implicitly	12✔
28	from pants.util.frozendict import FrozenDict	12✔
29
30	logger = logging.getLogger(__name__)	12✔
31
32
33	@dataclass(frozen=True, order=True)	12✔
34	class PythonRequirementVersion:	12✔
35	_parsed: Version	12✔
36
37	@classmethod	12✔
38	def parse(cls, version: str) -> PythonRequirementVersion:	12✔
39	return cls(parse(version))	1✔
40
41	def __str__(self) -> str:	12✔
42	return str(self._parsed)	1✔
43
44	def __getattr__(self, key: str) -> Any:	12✔
45	return getattr(self._parsed, key)	×
46
47
48	def _pex_lockfile_requirements(	12✔
49	lockfile_data: Mapping[str, Any] \| None, path: str \| None = None
50	) -> LockfilePackages:
51	if not lockfile_data:	1✔
52	return LockfilePackages({})	×
53
54	try:	1✔
55	# Setup generators
56	locked_resolves = (	1✔
57	(
58	(PackageName(r["project_name"]), PythonRequirementVersion.parse(r["version"]))
59	for r in resolve["locked_requirements"]
60	)
61	for resolve in lockfile_data["locked_resolves"]
62	)
63	requirements = dict(itertools.chain.from_iterable(locked_resolves))	1✔
64	except KeyError as e:	×
65	if path:	×
66	logger.warning(f"{path}: Failed to parse lockfile: {e}")	×
67
68	requirements = {}	×
69
70	return LockfilePackages(requirements)	1✔
71
72
73	def _uv_lockfile_requirements(	12✔
74	lockfile_data: Mapping[str, Any] \| None, path: str \| None = None
75	) -> LockfilePackages:
NEW 76	if not lockfile_data:	×
NEW 77	return LockfilePackages({})	×
78
NEW 79	requirements = {}	×
NEW 80	for pkg in lockfile_data.get("package", []):	×
NEW 81	try:	×
NEW 82	name = pkg["name"]	×
NEW 83	version = pkg.get("version")	×
84	# Skip the synthetic virtual package, it's not interesting in diffs.
NEW 85	if version is not None and not name.startswith("pants-lockfile-for-"):	×
NEW 86	requirements[PackageName(name)] = PythonRequirementVersion.parse(version)	×
NEW 87	except Exception as e:	×
NEW 88	if path:	×
NEW 89	logger.warning(f"{path}: Failed to parse package entry in lockfile: {e}")	×
90
NEW 91	return LockfilePackages(requirements)	×
92
93
94	def _parse_lockfile_packages(	12✔
95	content: bytes, lockfile_format: LockfileFormat, path: str \| None = None
96	) -> LockfilePackages:
97	"""Parse the packages from lockfile content according to its format."""
98	try:	1✔
99	match lockfile_format:	1✔
100	case LockfileFormat.PEX:	1✔
101	# strip_comments_from_pex_json_lockfile is idempotent, so safe to call on
102	# already-stripped content (e.g. when content was loaded via load_lockfile).
103	stripped = strip_comments_from_pex_json_lockfile(content)	1✔
104	data = FrozenDict.deep_freeze(json.loads(stripped))	1✔
105	return _pex_lockfile_requirements(data, path)	1✔
NEW 106	case LockfileFormat.UV:	×
NEW 107	data = FrozenDict.deep_freeze(tomllib.loads(content.decode()))	×
NEW 108	return _uv_lockfile_requirements(data, path)	×
NEW 109	case LockfileFormat.CONSTRAINTS_DEPRECATED:	×
110	# These can't meaningfully be diffed.
NEW 111	return LockfilePackages({})	×
NEW 112	case _:	×
NEW 113	raise ValueError(f"Unrecognized lockfile format: {lockfile_format}")	×
NEW 114	except Exception as e:	×
NEW 115	if path:	×
NEW 116	logger.debug(f"{path}: Failed to parse lockfile contents: {e}")	×
NEW 117	return LockfilePackages({})	×
118
119
120	async def _generate_lockfile_diff(	12✔
121	digest: Digest, resolve_name: str, path: str, new_format: LockfileFormat
122	) -> LockfileDiff:
123	"""Generate a diff between the newly generated lockfile and the existing one on disk.
124
125	Handles all combinations of old vs. new and pex vs. uv lockfile formats.
126	"""
127	new_digest_contents = await get_digest_contents(digest)	1✔
128	new_content = next(c for c in new_digest_contents if c.path == path).content	1✔
129	new_packages = _parse_lockfile_packages(new_content, new_format, path)	1✔
130
131	old_packages = LockfilePackages({})	1✔
132	try:	1✔
133	loaded = await load_lockfile(	1✔
134	LoadedLockfileRequest(
135	Lockfile(
136	url=path,
137	url_description_of_origin="existing lockfile",
138	resolve_name=resolve_name,
139	)
140	),
141	**implicitly(),
142	)
143	old_content_entries = await get_digest_contents(loaded.lockfile_digest)	1✔
144	old_content = next(iter(old_content_entries)).content	1✔
145	old_packages = _parse_lockfile_packages(old_content, loaded.lockfile_format, path)	1✔
NEW 146	except EngineError:	×
147	# May fail if the file doesn't exist, which is expected the first time a new lockfile
148	# is generated.
NEW 149	pass	×
150
151	return LockfileDiff.create(	1✔
152	path=path,
153	resolve_name=resolve_name,
154	old=old_packages,
155	new=new_packages,
156	)

pantsbuild / pants / 25405181773

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