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

pantsbuild / pants / 18143316655

30 Sep 2025 09:00PM UTC coverage: 80.263% (-0.01%) from 80.275%
18143316655

push

github

web-flow
Write Python lockfile metadata to separate files (#22713)

Currently we tack it on as a header to the lockfile, which
makes the lockfile unusable when working directly with Pex
without first manually editing it to remove the header.

Instead, we now (optionally) write to a separate metadata
sibling file. 

We always unconditionally try and read the metadata file,
falling back to the header if it doesn't exist. This will
allow us to regenerate the embedded lockfiles without
worrying about whether the user has the new metadata
files enabled.

42 of 87 new or added lines in 7 files covered. (48.28%)

1 existing line in 1 file now uncovered.

77226 of 96216 relevant lines covered (80.26%)

3.37 hits per line

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

90.77
/src/python/pants/core/util_rules/lockfile_metadata.py
1
# Copyright 2021 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 json
12✔
8
from collections.abc import Callable, Iterable
12✔
9
from dataclasses import dataclass
12✔
10
from enum import Enum
12✔
11
from typing import Any, ClassVar, Generic, TypeVar, cast
12✔
12

13
from pants.util.docutil import bin_name
12✔
14
from pants.util.ordered_set import FrozenOrderedSet
12✔
15
from pants.util.strutil import softwrap
12✔
16

17
BEGIN_LOCKFILE_HEADER = "--- BEGIN PANTS LOCKFILE METADATA: DO NOT EDIT OR REMOVE ---"
12✔
18
END_LOCKFILE_HEADER = "--- END PANTS LOCKFILE METADATA ---"
12✔
19

20

21
class LockfileScope(Enum):
12✔
22
    JVM = "jvm"
12✔
23
    PYTHON = "python"
12✔
24

25

26
_LockfileMetadataSubclass = TypeVar("_LockfileMetadataSubclass", bound="LockfileMetadata")
12✔
27
# N.B. the value type is `type[_LockfileMetadataSubclass]`
28
_concrete_metadata_classes: dict[tuple[LockfileScope, int], type] = {}
12✔
29

30

31
# Registrar types (pre-declaring to avoid repetition)
32
RegisterClassForVersion = Callable[[type["LockfileMetadata"]], type["LockfileMetadata"]]
12✔
33

34

35
def lockfile_metadata_registrar(scope: LockfileScope) -> Callable[[int], RegisterClassForVersion]:
12✔
36
    """Decorator factory -- returns a decorator that can be used to register Lockfile metadata
37
    version subclasses belonging to a specific `LockfileScope`."""
38

39
    def _lockfile_metadata_version(
12✔
40
        version: int,
41
    ) -> RegisterClassForVersion:
42
        """Decorator to register a Lockfile metadata version subclass with a given version number.
43

44
        The class must be a frozen dataclass
45
        """
46

47
        def _dec(cls: type[LockfileMetadata]) -> type[LockfileMetadata]:
12✔
48
            # Only frozen dataclasses may be registered as lockfile metadata:
49
            cls_dataclass_params = getattr(cls, "__dataclass_params__", None)
12✔
50
            if not cls_dataclass_params or not cls_dataclass_params.frozen:
12✔
51
                raise ValueError(
×
52
                    softwrap(
53
                        """
54
                        Classes registered with `_lockfile_metadata_version` may only be
55
                        frozen dataclasses
56
                        """
57
                    )
58
                )
59
            _concrete_metadata_classes[(scope, version)] = cls
12✔
60
            return cls
12✔
61

62
        return _dec
12✔
63

64
    return _lockfile_metadata_version
12✔
65

66

67
class InvalidLockfileError(Exception):
12✔
68
    pass
12✔
69

70

71
class NoLockfileMetadataBlock(InvalidLockfileError):
12✔
72
    pass
12✔
73

74

75
@dataclass(frozen=True)
12✔
76
class LockfileMetadata:
12✔
77
    """Base class for metadata that is attached to a given lockfile.
78

79
    This class provides the external API for serializing, deserializing, and validating the
80
    contents of individual lockfiles. New versions of metadata implement a concrete subclass and
81
    provide deserialization and validation logic, along with specialist serialization logic.
82

83
    To construct an instance of the most recent concrete subclass, call `LockfileMetadata.new()`.
84
    """
85

86
    scope: ClassVar[LockfileScope]
12✔
87

88
    @classmethod
12✔
89
    def from_lockfile(
12✔
90
        cls: type[_LockfileMetadataSubclass],
91
        lockfile: bytes,
92
        lockfile_path: str | None = None,
93
        resolve_name: str | None = None,
94
        *,
95
        delimeter: str,
96
    ) -> _LockfileMetadataSubclass:
97
        """Parse and return the metadata from the lockfile's header.
98

99
        This shouldn't be called on `LockfileMetadata`, but rather on the "base class" for your
100
        metadata class. See the existing callers for an example.
101
        """
102
        assert cls is not LockfileMetadata, "Call me on a subclass!"
10✔
103
        in_metadata_block = False
10✔
104
        metadata_lines = []
10✔
105
        for line in lockfile.splitlines():
10✔
106
            if line == f"{delimeter} {BEGIN_LOCKFILE_HEADER}".encode():
10✔
107
                in_metadata_block = True
10✔
108
            elif line == f"{delimeter} {END_LOCKFILE_HEADER}".encode():
10✔
109
                break
10✔
110
            elif in_metadata_block:
10✔
111
                metadata_lines.append(line[len(delimeter) + 1 :])
10✔
112

113
        error_suffix = softwrap(
10✔
114
            f"""
115
            To resolve this error, you will need to regenerate the lockfile by running
116
            `{bin_name()} generate-lockfiles
117
            """
118
        )
119
        if resolve_name:
10✔
120
            error_suffix += f" --resolve={resolve_name}"
2✔
121
        error_suffix += "`."
10✔
122

123
        if lockfile_path is not None and resolve_name is not None:
10✔
124
            lockfile_description = f"the lockfile `{lockfile_path}` for `{resolve_name}`"
×
125
        elif lockfile_path is not None:
10✔
126
            lockfile_description = f"the lockfile `{lockfile_path}`"
×
127
        elif resolve_name is not None:
10✔
128
            lockfile_description = f"the lockfile for `{resolve_name}`"
2✔
129
        else:
130
            lockfile_description = "this lockfile"
10✔
131

132
        if not metadata_lines:
10✔
133
            raise NoLockfileMetadataBlock(
1✔
134
                f"Could not find a Pants metadata block in {lockfile_description}. {error_suffix}"
135
            )
136
        try:
10✔
137
            json_dict = json.loads(b"\n".join(metadata_lines))
10✔
138
        except json.decoder.JSONDecodeError:
1✔
139
            raise InvalidLockfileError(
1✔
140
                softwrap(
141
                    f"""
142
                    Metadata header in {lockfile_description} is not a valid JSON string and can't
143
                    be decoded.
144
                    """
145
                )
146
                + error_suffix
147
            )
148
        return cls.from_json_dict(json_dict, lockfile_description, error_suffix)
10✔
149

150
    @classmethod
12✔
151
    def from_json_dict(
12✔
152
        cls: type[_LockfileMetadataSubclass],
153
        json_dict: dict[Any, Any],
154
        lockfile_description: str,
155
        error_suffix: str,
156
    ) -> _LockfileMetadataSubclass:
157
        version = json_dict.get("version", 1)
10✔
158
        concrete_class = _concrete_metadata_classes[(cls.scope, version)]
10✔
159

160
        assert issubclass(concrete_class, cls)
10✔
161
        assert concrete_class.scope == cls.scope, (
10✔
162
            "The class used to call `from_json_dict` has a different scope than what was "
163
            f"expected given the metadata. Expected '{cls.scope}', got '{concrete_class.scope}'",
164
        )
165

166
        return concrete_class._from_json_dict(json_dict, lockfile_description, error_suffix)
10✔
167

168
    @classmethod
12✔
169
    def _from_json_dict(
12✔
170
        cls: type[_LockfileMetadataSubclass],
171
        json_dict: dict[Any, Any],
172
        lockfile_description: str,
173
        error_suffix: str,
174
    ) -> _LockfileMetadataSubclass:
175
        """Construct a `LockfileMetadata` subclass from the supplied JSON dict.
176

177
        *** Not implemented. Subclasses should override. ***
178

179

180
        `lockfile_description` is a detailed, human-readable description of the lockfile, which can
181
        be read by the user to figure out which lockfile is broken in case of an error.
182

183
        `error_suffix` is a string describing how to fix the lockfile.
184
        """
185

186
        raise NotImplementedError(
×
187
            "`LockfileMetadata._from_json_dict` should not be directly called."
188
        )
189

190
    def to_json(self, with_description: str | None = None) -> str:
12✔
191
        metadata_dict = self.__render_header_dict()
5✔
192
        if with_description is not None:
5✔
NEW
193
            metadata_dict["description"] = with_description
×
194
        return json.dumps(metadata_dict, ensure_ascii=True, indent=2)
5✔
195

196
    def add_header_to_lockfile(
12✔
197
        self, lockfile: bytes, *, delimeter: str, regenerate_command: str
198
    ) -> bytes:
199
        metadata_json = self.to_json()
5✔
200
        metadata_as_a_comment = "\n".join(f"{delimeter} {l}" for l in metadata_json.splitlines())
5✔
201

202
        regenerate_command_bytes = "\n".join(
5✔
203
            [
204
                f"{delimeter} This lockfile was autogenerated by Pants. To regenerate, run:",
205
                delimeter,
206
                f"{delimeter}    {regenerate_command}",
207
            ]
208
        ).encode()
209
        header = "\n".join(
5✔
210
            [
211
                f"{delimeter} {BEGIN_LOCKFILE_HEADER}",
212
                metadata_as_a_comment,
213
                f"{delimeter} {END_LOCKFILE_HEADER}",
214
            ]
215
        ).encode("ascii")
216

217
        return b"%b\n%b\n%b\n\n%b" % (
5✔
218
            regenerate_command_bytes,
219
            delimeter.encode(),
220
            header,
221
            lockfile,
222
        )
223

224
    def __render_header_dict(self) -> dict[Any, Any]:
12✔
225
        """Produce a dictionary to be serialized into the lockfile header.
226

227
        Each class should implement a class method called `additional_header_attrs`, which returns a
228
        `dict` containing the metadata attributes that should be stored in the lockfile.
229
        """
230

231
        attrs: dict[Any, tuple[Any, type]] = {}  # attr name -> (value, where we first saw it)
5✔
232
        for cls in reversed(self.__class__.__mro__[:-1]):
5✔
233
            new_attrs = cast(LockfileMetadata, cls).additional_header_attrs(self)
5✔
234
            for attr in new_attrs:
5✔
235
                if attr in attrs and attrs[attr][0] != new_attrs[attr]:
5✔
236
                    raise AssertionError(
×
237
                        softwrap(
238
                            f"""
239
                            Lockfile header attribute `{attr}` was returned by both
240
                            `{attrs[attr][1]}` and `{cls}`, returning different values. If these
241
                            classes return the same attribute, they must also return the same
242
                            value.
243
                            """
244
                        )
245
                    )
246
                attrs[attr] = new_attrs[attr], cls
5✔
247

248
        return {key: val[0] for key, val in attrs.items()}
5✔
249

250
    @classmethod
12✔
251
    def additional_header_attrs(cls, instance: LockfileMetadata) -> dict[Any, Any]:
12✔
252
        return {"version": instance.metadata_version()}
5✔
253

254
    def metadata_version(self):
12✔
255
        """Returns the version number for this metadata class, or raises an exception.
256

257
        To avoid raising an exception, ensure the subclass is decorated with
258
        `lockfile_metadata_version`
259
        """
260
        for (scope, ver), cls in _concrete_metadata_classes.items():
5✔
261
            # Note that we do exact version matches so that authors can subclass earlier versions.
262
            if type(self) is cls:
5✔
263
                return ver
5✔
264
        raise ValueError("Trying to serialize an unregistered `LockfileMetadata` subclass.")
×
265

266

267
def calculate_invalidation_digest(requirements: Iterable[str]) -> str:
12✔
268
    """Returns an invalidation digest for the given requirements."""
269
    m = hashlib.sha256()
1✔
270
    inputs = {
1✔
271
        # `FrozenOrderedSet` deduplicates while keeping ordering, which speeds up the sorting if
272
        # the input was already sorted.
273
        "requirements": sorted(FrozenOrderedSet(requirements)),
274
    }
275
    m.update(json.dumps(inputs).encode("utf-8"))
1✔
276
    return m.hexdigest()
1✔
277

278

279
T = TypeVar("T")
12✔
280

281

282
class LockfileMetadataValidation(Generic[T]):
12✔
283
    """Boolean-like value which additionally carries reasons why a validation failed."""
284

285
    failure_reasons: set[T]
12✔
286

287
    def __init__(self, failure_reasons: Iterable[T] = ()):
12✔
288
        self.failure_reasons = set(failure_reasons)
10✔
289

290
    def __bool__(self):
12✔
291
        return not self.failure_reasons
10✔
292

293

294
def _get_metadata(
12✔
295
    metadata: dict[Any, Any],
296
    lockfile_description: str,
297
    error_suffix: str,
298
) -> Callable[[str, type[T], Callable[[Any], T] | None], T]:
299
    """Returns a function that will get a given key from the `metadata` dict, and optionally do some
300
    verification and post-processing to return a value of the correct type."""
301

302
    def get_metadata(key: str, type_: type[T], coerce: Callable[[Any], T] | None) -> T:
10✔
303
        val: Any
304
        try:
10✔
305
            val = metadata[key]
10✔
306
        except KeyError:
×
307
            raise InvalidLockfileError(
×
308
                softwrap(
309
                    f"""
310
                    Required key `{key}` is not present in metadata header for
311
                    {lockfile_description}. {error_suffix}
312
                    """
313
                )
314
            )
315

316
        if not coerce:
10✔
317
            if isinstance(val, type_):
1✔
318
                return val
1✔
319

320
            raise InvalidLockfileError(
×
321
                softwrap(
322
                    f"""
323
                    Metadata value `{key}` in {lockfile_description} must
324
                    be a {type(type_).__name__}. {error_suffix}
325
                    """
326
                )
327
            )
328
        else:
329
            try:
10✔
330
                return coerce(val)
10✔
331
            except Exception:
×
332
                raise InvalidLockfileError(
×
333
                    softwrap(
334
                        f"""
335
                        Metadata value `{key}` in {lockfile_description} must be able to
336
                        be converted to a {type(type_).__name__}. {error_suffix}
337
                        """
338
                    )
339
                )
340

341
    return get_metadata
10✔
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

© 2025 Coveralls, Inc