19015773527

Committed 02 Nov 2025 05:33PM UTC coverage: 17.872% (-62.4%) from 80.3%

Build # 19015773527

Build Type

Pull #22816

github

Committed by

web-flow

Commit Message

Merge a12d75757 into 6c024e162

Pull Request Pull Request #22816: Update Pants internal Python to 3.14

Run Details

4 of 5 new or added lines in 3 files covered. (80.0%)

28452 existing lines in 683 files now uncovered.

9831 of 55007 relevant lines covered (17.87%)

0.18 hits per line

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

80.47

/src/python/pants/engine/fs.py

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

from __future__ import annotations

from collections.abc import Iterable, Mapping, Sequence
from dataclasses import dataclass
from datetime import timedelta
from enum import Enum
from typing import TYPE_CHECKING, Union

# Note: several of these types are re-exported as the public API of `engine/fs.py`.
from pants.base.glob_match_error_behavior import GlobMatchErrorBehavior as GlobMatchErrorBehavior
from pants.engine.collection import Collection
from pants.engine.engine_aware import SideEffecting
from pants.engine.internals.native_engine import EMPTY_DIGEST as EMPTY_DIGEST  # noqa: F401
from pants.engine.internals.native_engine import (  # noqa: F401
    EMPTY_FILE_DIGEST as EMPTY_FILE_DIGEST,
)
from pants.engine.internals.native_engine import EMPTY_SNAPSHOT as EMPTY_SNAPSHOT  # noqa: F401
from pants.engine.internals.native_engine import AddPrefix as AddPrefix
from pants.engine.internals.native_engine import Digest as Digest
from pants.engine.internals.native_engine import FileDigest as FileDigest
from pants.engine.internals.native_engine import MergeDigests as MergeDigests
from pants.engine.internals.native_engine import PathMetadata, PathNamespace
from pants.engine.internals.native_engine import RemovePrefix as RemovePrefix
from pants.engine.internals.native_engine import Snapshot as Snapshot
from pants.util.frozendict import FrozenDict

if TYPE_CHECKING:
    from pants.engine.internals.scheduler import SchedulerSession


@dataclass(frozen=True)
class Paths:
    """A Paths object is a collection of sorted file paths and dir paths.

    Paths is like a Snapshot, but has the performance optimization that it does not digest the files
    or save them to the LMDB store.
    """

    files: tuple[str, ...]
    dirs: tuple[str, ...]


@dataclass(frozen=True)
class FileContent:
    """The content of a file."""

    path: str
    content: bytes
    is_executable: bool = False

    def __post_init__(self):
        if not isinstance(self.content, bytes):
            raise TypeError(
                f"Expected 'content' to be bytes, but got {type(self.content).__name__}"
            )

    def __repr__(self) -> str:
        return (
            f"FileContent(path={self.path}, content=(len:{len(self.content)}), "
            f"is_executable={self.is_executable})"
        )


@dataclass(frozen=True)
class FileEntry:
    """An indirect reference to the content of a file by digest."""

    path: str
    file_digest: FileDigest
    is_executable: bool = False


@dataclass(frozen=True)
class SymlinkEntry:
    """A symlink pointing to a target path.

    For the symlink target:
        - uses a forward slash `/` path separator.
        - can be relative to the parent directory of the symlink or can be an absolute path starting with `/`.
        - Allows `..` components anywhere in the path (as logical canonicalization may lead to
            different behavior in the presence of directory symlinks).

    See also the REAPI for a SymlinkNode:
    https://github.com/bazelbuild/remote-apis/blob/aa29b91f336b9be2c5370297210b67a6654c0b72/build/bazel/remote/execution/v2/remote_execution.proto#L882
    """

    path: str
    target: str


@dataclass(frozen=True)
class Directory:
    """The path to a directory."""

    path: str

    def __repr__(self) -> str:
        return f"Directory({repr(self.path)})"


class DigestContents(Collection[FileContent]):
    """The file contents of a Digest.

    Although the contents of the Digest are not memoized across `@rules` or across runs (each
    request for `DigestContents` will load the file content from disk), this API should still
    generally only be used for small inputs, since concurrency might mean that very many `@rule`s
    are holding `DigestContents` simultaneously.
    """


class DigestEntries(Collection[Union[FileEntry, SymlinkEntry, Directory]]):
    """The indirect file contents of a Digest.

    DigestEntries is a collection of FileEntry/SymlinkEntry/Directory instances representing,
    respectively, actual files, actual symlinks, and empty directories present in the Digest.
    """


class CreateDigest(Collection[Union[FileContent, FileEntry, SymlinkEntry, Directory]]):
    """A request to create a Digest with the input FileContent/FileEntry/SymlinkEntry/Directory
    values.

    The engine will create any parent directories necessary, e.g. `FileContent('a/b/c.txt')` will
    result in `a/`, `a/b`, and `a/b/c.txt` being created. You only need to use `Directory` to
    create an empty directory.

    This does _not_ actually materialize the digest to the build root. You must use
    `engine.fs.Workspace` in a `@goal_rule` to save the resulting digest to disk.
    """


class GlobExpansionConjunction(Enum):
    """Describe whether to require that only some or all glob strings match in a target's sources.

    NB: this object is interpreted from within Snapshot::lift_path_globs() -- that method will need to
    be aware of any changes to this object's definition.
    """

    any_match = "any_match"
    all_match = "all_match"


@dataclass(frozen=True)
class PathGlobs:
    globs: tuple[str, ...]
    glob_match_error_behavior: GlobMatchErrorBehavior
    conjunction: GlobExpansionConjunction
    description_of_origin: str | None

    def __init__(
        self,
        globs: Iterable[str],
        glob_match_error_behavior: GlobMatchErrorBehavior = GlobMatchErrorBehavior.ignore,
        conjunction: GlobExpansionConjunction = GlobExpansionConjunction.any_match,
        description_of_origin: str | None = None,
    ) -> None:
        """A request to find files given a set of globs.

        The syntax supported is roughly Git's glob syntax. Use `*` for globs, `**` for recursive
        globs, and `!` for ignores.

        :param globs: globs to match, e.g. `foo.txt` or `**/*.txt`. To exclude something, prefix it
            with `!`, e.g. `!ignore.py`.
        :param glob_match_error_behavior: whether to warn or error upon match failures
        :param conjunction: whether all `globs` must match or only at least one must match
        :param description_of_origin: a human-friendly description of where this PathGlobs request
            is coming from, used to improve the error message for unmatched globs. For example,
            this might be the text string "the option `--isort-config`".
        """

        # NB: this object is interpreted from within Snapshot::lift_path_globs() -- that method
        # will need to be aware of any changes to this object's definition.
        object.__setattr__(self, "globs", tuple(sorted(globs)))
        object.__setattr__(self, "glob_match_error_behavior", glob_match_error_behavior)
        object.__setattr__(self, "conjunction", conjunction)
        object.__setattr__(self, "description_of_origin", description_of_origin)
        self.__post_init__()

    def __post_init__(self) -> None:
        if self.glob_match_error_behavior == GlobMatchErrorBehavior.ignore:
            if self.description_of_origin:
                raise ValueError(
                    "You provided a `description_of_origin` value when `glob_match_error_behavior` "
                    "is set to `ignore`. The `ignore` value means that the engine will never "
                    "generate an error when the globs are generated, so `description_of_origin` "
                    "won't end up ever being used. Please either change "
                    "`glob_match_error_behavior` to `warn` or `error`, or remove "
                    "`description_of_origin`."
                )
        else:
            if not self.description_of_origin:
                raise ValueError(
                    "Please provide a `description_of_origin` so that the error message is more "
                    "helpful to users when their globs fail to match."
                )


@dataclass(frozen=True)
class PathGlobsAndRoot:
    """A set of PathGlobs to capture relative to some root (which may exist outside of the
    buildroot).

    If the `digest_hint` is set, it must be the Digest that we would expect to get if we were to
    expand and Digest the globs. The hint is an optimization that allows for bypassing filesystem
    operations in cases where the expected Digest is known, and the content for the Digest is
    already stored.
    """

    path_globs: PathGlobs
    root: str
    digest_hint: Digest | None = None


@dataclass(frozen=True)
class DigestSubset:
    """A request to get a subset of a digest.

    The digest will be traversed symlink-oblivious to match the provided globs. If you require a
    symlink-aware subset, you can access the digest's DigestEntries, filter them out, and create a
    new digest.
    """

    digest: Digest
    globs: PathGlobs


@dataclass(frozen=True)
class DownloadFile:
    """Retrieve the contents of a file via an HTTP GET request or directly for local file:// URLs.

    To compute the `expected_digest`, manually download the file, then run `shasum -a 256` to
    compute the fingerprint and `wc -c` to compute the expected length of the downloaded file in
    bytes.
    """

    url: str
    expected_digest: FileDigest


@dataclass(frozen=True)
class NativeDownloadFile:
    """Retrieve the contents of a file via an HTTP GET request or directly for local file:// URLs.

    This request is handled directly by the native engine without any additional coercion by plugins,
    and therefore should only be used in cases where the URL is known to be publicly accessible.
    Otherwise, callers should use `DownloadFile`.

    The auth_headers are part of this nodes' cache key for memoization (changing a header invalidates
    prior results) but are not part of the underlying cache key for the local/remote cache (changing
    a header won't re-download a file if the file was previously downloaded).
    """

    url: str
    expected_digest: FileDigest
    # NB: This mapping can be of any arbitrary headers, but should be limited to those required for
    # authorization.
    auth_headers: FrozenDict[str, str]

    retry_error_duration: timedelta
    max_attempts: int

    def __init__(
        self,
        url: str,
        expected_digest: FileDigest,
        auth_headers: Mapping[str, str] | None = None,
        retry_delay_duration: timedelta = timedelta(milliseconds=10),
        max_attempts: int = 4,
    ) -> None:
        object.__setattr__(self, "url", url)
        object.__setattr__(self, "expected_digest", expected_digest)
        object.__setattr__(self, "auth_headers", FrozenDict(auth_headers or {}))
        object.__setattr__(self, "retry_error_duration", retry_delay_duration)
        object.__setattr__(self, "max_attempts", max_attempts)


@dataclass(frozen=True)
class Workspace(SideEffecting):
    """A handle for operations that mutate the local filesystem."""

    _scheduler: SchedulerSession
    _enforce_effects: bool = True

    def write_digest(
        self,
        digest: Digest,
        *,
        path_prefix: str | None = None,
        clear_paths: Sequence[str] = (),
        side_effecting: bool = True,
    ) -> None:
        """Write a digest to disk, relative to the build root.

        You should not use this in a `for` loop due to slow performance. Instead, first merge
        digests and then write the single merged digest.

        As an advanced use-case, if the digest is known to be written to a temporary or idempotent
        location, side_effecting=False may be passed to avoid tracking this write as a side effect.
        """
        if side_effecting:
            self.side_effected()
        self._scheduler.write_digest(digest, path_prefix=path_prefix, clear_paths=clear_paths)


@dataclass(frozen=True)
class SpecsPaths(Paths):
    """All files matched by command line specs.

    `@goal_rule`s may request this when they only need source files to operate and do not need any
    target information. This allows running on files with no owning targets.
    """


@dataclass(frozen=True)
class SnapshotDiff:
    our_unique_files: tuple[str, ...] = ()
    our_unique_dirs: tuple[str, ...] = ()
    their_unique_files: tuple[str, ...] = ()
    their_unique_dirs: tuple[str, ...] = ()
    changed_files: tuple[str, ...] = ()

    @classmethod
    def from_snapshots(cls, ours: Snapshot, theirs: Snapshot) -> SnapshotDiff:
        return cls(*ours._diff(theirs))


@dataclass(frozen=True)
class PathMetadataRequest:
    """Request the full metadata of a single path in the filesystem.

    Note: This API is symlink-aware and will distinguish between symlinks and regular files.
    """

    path: str
    namespace: PathNamespace = PathNamespace.WORKSPACE


@dataclass(frozen=True)
class PathMetadataResult:
    """Result of requesting the metadata for a path in the filesystem.

    The `metadata` field will contain the metadata for the requested path, or else `None` if the
    path does not exist.
    """

    metadata: PathMetadata | None


def rules():
    # Avoids an import cycle.
    from pants.engine.rules import QueryRule

    return (
        QueryRule(Digest, (CreateDigest,)),
        QueryRule(Digest, (PathGlobs,)),
        QueryRule(Digest, (AddPrefix,)),
        QueryRule(Digest, (RemovePrefix,)),
        QueryRule(Digest, (NativeDownloadFile,)),
        QueryRule(Digest, (MergeDigests,)),
        QueryRule(Digest, (DigestSubset,)),
        QueryRule(DigestContents, (Digest,)),
        QueryRule(Snapshot, (Digest,)),
        QueryRule(Paths, (PathGlobs,)),
    )

1	# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	1✔
5
6	from collections.abc import Iterable, Mapping, Sequence	1✔
7	from dataclasses import dataclass	1✔
8	from datetime import timedelta	1✔
9	from enum import Enum	1✔
10	from typing import TYPE_CHECKING, Union	1✔
11
12	# Note: several of these types are re-exported as the public API of `engine/fs.py`.
13	from pants.base.glob_match_error_behavior import GlobMatchErrorBehavior as GlobMatchErrorBehavior	1✔
14	from pants.engine.collection import Collection	1✔
15	from pants.engine.engine_aware import SideEffecting	1✔
16	from pants.engine.internals.native_engine import EMPTY_DIGEST as EMPTY_DIGEST # noqa: F401	1✔
17	from pants.engine.internals.native_engine import ( # noqa: F401	1✔
18	EMPTY_FILE_DIGEST as EMPTY_FILE_DIGEST,
19	)
20	from pants.engine.internals.native_engine import EMPTY_SNAPSHOT as EMPTY_SNAPSHOT # noqa: F401	1✔
21	from pants.engine.internals.native_engine import AddPrefix as AddPrefix	1✔
22	from pants.engine.internals.native_engine import Digest as Digest	1✔
23	from pants.engine.internals.native_engine import FileDigest as FileDigest	1✔
24	from pants.engine.internals.native_engine import MergeDigests as MergeDigests	1✔
25	from pants.engine.internals.native_engine import PathMetadata, PathNamespace	1✔
26	from pants.engine.internals.native_engine import RemovePrefix as RemovePrefix	1✔
27	from pants.engine.internals.native_engine import Snapshot as Snapshot	1✔
28	from pants.util.frozendict import FrozenDict	1✔
29
30	if TYPE_CHECKING:
31	from pants.engine.internals.scheduler import SchedulerSession
32
33
34	@dataclass(frozen=True)	1✔
35	class Paths:	1✔
36	"""A Paths object is a collection of sorted file paths and dir paths.
37
38	Paths is like a Snapshot, but has the performance optimization that it does not digest the files
39	or save them to the LMDB store.
40	"""
41
42	files: tuple[str, ...]	1✔
43	dirs: tuple[str, ...]	1✔
44
45
46	@dataclass(frozen=True)	1✔
47	class FileContent:	1✔
48	"""The content of a file."""
49
50	path: str	1✔
51	content: bytes	1✔
52	is_executable: bool = False	1✔
53
54	def __post_init__(self):	1✔
UNCOV 55	if not isinstance(self.content, bytes):	×
UNCOV 56	raise TypeError(	×
57	f"Expected 'content' to be bytes, but got {type(self.content).__name__}"
58	)
59
60	def __repr__(self) -> str:	1✔
61	return (	×
62	f"FileContent(path={self.path}, content=(len:{len(self.content)}), "
63	f"is_executable={self.is_executable})"
64	)
65
66
67	@dataclass(frozen=True)	1✔
68	class FileEntry:	1✔
69	"""An indirect reference to the content of a file by digest."""
70
71	path: str	1✔
72	file_digest: FileDigest	1✔
73	is_executable: bool = False	1✔
74
75
76	@dataclass(frozen=True)	1✔
77	class SymlinkEntry:	1✔
78	"""A symlink pointing to a target path.
79
80	For the symlink target:
81	- uses a forward slash `/` path separator.
82	- can be relative to the parent directory of the symlink or can be an absolute path starting with `/`.
83	- Allows `..` components anywhere in the path (as logical canonicalization may lead to
84	different behavior in the presence of directory symlinks).
85
86	See also the REAPI for a SymlinkNode:
87	https://github.com/bazelbuild/remote-apis/blob/aa29b91f336b9be2c5370297210b67a6654c0b72/build/bazel/remote/execution/v2/remote_execution.proto#L882
88	"""
89
90	path: str	1✔
91	target: str	1✔
92
93
94	@dataclass(frozen=True)	1✔
95	class Directory:	1✔
96	"""The path to a directory."""
97
98	path: str	1✔
99
100	def __repr__(self) -> str:	1✔
101	return f"Directory({repr(self.path)})"	×
102
103
104	class DigestContents(Collection[FileContent]):	1✔
105	"""The file contents of a Digest.
106
107	Although the contents of the Digest are not memoized across `@rules` or across runs (each
108	request for `DigestContents` will load the file content from disk), this API should still
109	generally only be used for small inputs, since concurrency might mean that very many `@rule`s
110	are holding `DigestContents` simultaneously.
111	"""
112
113
114	class DigestEntries(Collection[Union[FileEntry, SymlinkEntry, Directory]]):	1✔
115	"""The indirect file contents of a Digest.
116
117	DigestEntries is a collection of FileEntry/SymlinkEntry/Directory instances representing,
118	respectively, actual files, actual symlinks, and empty directories present in the Digest.
119	"""
120
121
122	class CreateDigest(Collection[Union[FileContent, FileEntry, SymlinkEntry, Directory]]):	1✔
123	"""A request to create a Digest with the input FileContent/FileEntry/SymlinkEntry/Directory
124	values.
125
126	The engine will create any parent directories necessary, e.g. `FileContent('a/b/c.txt')` will
127	result in `a/`, `a/b`, and `a/b/c.txt` being created. You only need to use `Directory` to
128	create an empty directory.
129
130	This does _not_ actually materialize the digest to the build root. You must use
131	`engine.fs.Workspace` in a `@goal_rule` to save the resulting digest to disk.
132	"""
133
134
135	class GlobExpansionConjunction(Enum):	1✔
136	"""Describe whether to require that only some or all glob strings match in a target's sources.
137
138	NB: this object is interpreted from within Snapshot::lift_path_globs() -- that method will need to
139	be aware of any changes to this object's definition.
140	"""
141
142	any_match = "any_match"	1✔
143	all_match = "all_match"	1✔
144
145
146	@dataclass(frozen=True)	1✔
147	class PathGlobs:	1✔
148	globs: tuple[str, ...]	1✔
149	glob_match_error_behavior: GlobMatchErrorBehavior	1✔
150	conjunction: GlobExpansionConjunction	1✔
151	description_of_origin: str \| None	1✔
152
153	def __init__(	1✔
154	self,
155	globs: Iterable[str],
156	glob_match_error_behavior: GlobMatchErrorBehavior = GlobMatchErrorBehavior.ignore,
157	conjunction: GlobExpansionConjunction = GlobExpansionConjunction.any_match,
158	description_of_origin: str \| None = None,
159	) -> None:
160	"""A request to find files given a set of globs.
161
162	The syntax supported is roughly Git's glob syntax. Use `` for globs, `*` for recursive
163	globs, and `!` for ignores.
164
165	:param globs: globs to match, e.g. `foo.txt` or `*/.txt`. To exclude something, prefix it
166	with `!`, e.g. `!ignore.py`.
167	:param glob_match_error_behavior: whether to warn or error upon match failures
168	:param conjunction: whether all `globs` must match or only at least one must match
169	:param description_of_origin: a human-friendly description of where this PathGlobs request
170	is coming from, used to improve the error message for unmatched globs. For example,
171	this might be the text string "the option `--isort-config`".
172	"""
173
174	# NB: this object is interpreted from within Snapshot::lift_path_globs() -- that method
175	# will need to be aware of any changes to this object's definition.
UNCOV 176	object.__setattr__(self, "globs", tuple(sorted(globs)))	×
UNCOV 177	object.__setattr__(self, "glob_match_error_behavior", glob_match_error_behavior)	×
UNCOV 178	object.__setattr__(self, "conjunction", conjunction)	×
UNCOV 179	object.__setattr__(self, "description_of_origin", description_of_origin)	×
UNCOV 180	self.__post_init__()	×
181
182	def __post_init__(self) -> None:	1✔
UNCOV 183	if self.glob_match_error_behavior == GlobMatchErrorBehavior.ignore:	×
UNCOV 184	if self.description_of_origin:	×
185	raise ValueError(	×
186	"You provided a `description_of_origin` value when `glob_match_error_behavior` "
187	"is set to `ignore`. The `ignore` value means that the engine will never "
188	"generate an error when the globs are generated, so `description_of_origin` "
189	"won't end up ever being used. Please either change "
190	"`glob_match_error_behavior` to `warn` or `error`, or remove "
191	"`description_of_origin`."
192	)
193	else:
UNCOV 194	if not self.description_of_origin:	×
195	raise ValueError(	×
196	"Please provide a `description_of_origin` so that the error message is more "
197	"helpful to users when their globs fail to match."
198	)
199
200
201	@dataclass(frozen=True)	1✔
202	class PathGlobsAndRoot:	1✔
203	"""A set of PathGlobs to capture relative to some root (which may exist outside of the
204	buildroot).
205
206	If the `digest_hint` is set, it must be the Digest that we would expect to get if we were to
207	expand and Digest the globs. The hint is an optimization that allows for bypassing filesystem
208	operations in cases where the expected Digest is known, and the content for the Digest is
209	already stored.
210	"""
211
212	path_globs: PathGlobs	1✔
213	root: str	1✔
214	digest_hint: Digest \| None = None	1✔
215
216
217	@dataclass(frozen=True)	1✔
218	class DigestSubset:	1✔
219	"""A request to get a subset of a digest.
220
221	The digest will be traversed symlink-oblivious to match the provided globs. If you require a
222	symlink-aware subset, you can access the digest's DigestEntries, filter them out, and create a
223	new digest.
224	"""
225
226	digest: Digest	1✔
227	globs: PathGlobs	1✔
228
229
230	@dataclass(frozen=True)	1✔
231	class DownloadFile:	1✔
232	"""Retrieve the contents of a file via an HTTP GET request or directly for local file:// URLs.
233
234	To compute the `expected_digest`, manually download the file, then run `shasum -a 256` to
235	compute the fingerprint and `wc -c` to compute the expected length of the downloaded file in
236	bytes.
237	"""
238
239	url: str	1✔
240	expected_digest: FileDigest	1✔
241
242
243	@dataclass(frozen=True)	1✔
244	class NativeDownloadFile:	1✔
245	"""Retrieve the contents of a file via an HTTP GET request or directly for local file:// URLs.
246
247	This request is handled directly by the native engine without any additional coercion by plugins,
248	and therefore should only be used in cases where the URL is known to be publicly accessible.
249	Otherwise, callers should use `DownloadFile`.
250
251	The auth_headers are part of this nodes' cache key for memoization (changing a header invalidates
252	prior results) but are not part of the underlying cache key for the local/remote cache (changing
253	a header won't re-download a file if the file was previously downloaded).
254	"""
255
256	url: str	1✔
257	expected_digest: FileDigest	1✔
258	# NB: This mapping can be of any arbitrary headers, but should be limited to those required for
259	# authorization.
260	auth_headers: FrozenDict[str, str]	1✔
261
262	retry_error_duration: timedelta	1✔
263	max_attempts: int	1✔
264
265	def __init__(	1✔
266	self,
267	url: str,
268	expected_digest: FileDigest,
269	auth_headers: Mapping[str, str] \| None = None,
270	retry_delay_duration: timedelta = timedelta(milliseconds=10),
271	max_attempts: int = 4,
272	) -> None:
UNCOV 273	object.__setattr__(self, "url", url)	×
UNCOV 274	object.__setattr__(self, "expected_digest", expected_digest)	×
UNCOV 275	object.__setattr__(self, "auth_headers", FrozenDict(auth_headers or {}))	×
UNCOV 276	object.__setattr__(self, "retry_error_duration", retry_delay_duration)	×
UNCOV 277	object.__setattr__(self, "max_attempts", max_attempts)	×
278
279
280	@dataclass(frozen=True)	1✔
281	class Workspace(SideEffecting):	1✔
282	"""A handle for operations that mutate the local filesystem."""
283
284	_scheduler: SchedulerSession	1✔
285	_enforce_effects: bool = True	1✔
286
287	def write_digest(	1✔
288	self,
289	digest: Digest,
290	*,
291	path_prefix: str \| None = None,
292	clear_paths: Sequence[str] = (),
293	side_effecting: bool = True,
294	) -> None:
295	"""Write a digest to disk, relative to the build root.
296
297	You should not use this in a `for` loop due to slow performance. Instead, first merge
298	digests and then write the single merged digest.
299
300	As an advanced use-case, if the digest is known to be written to a temporary or idempotent
301	location, side_effecting=False may be passed to avoid tracking this write as a side effect.
302	"""
UNCOV 303	if side_effecting:	×
UNCOV 304	self.side_effected()	×
UNCOV 305	self._scheduler.write_digest(digest, path_prefix=path_prefix, clear_paths=clear_paths)	×
306
307
308	@dataclass(frozen=True)	1✔
309	class SpecsPaths(Paths):	1✔
310	"""All files matched by command line specs.
311
312	`@goal_rule`s may request this when they only need source files to operate and do not need any
313	target information. This allows running on files with no owning targets.
314	"""
315
316
317	@dataclass(frozen=True)	1✔
318	class SnapshotDiff:	1✔
319	our_unique_files: tuple[str, ...] = ()	1✔
320	our_unique_dirs: tuple[str, ...] = ()	1✔
321	their_unique_files: tuple[str, ...] = ()	1✔
322	their_unique_dirs: tuple[str, ...] = ()	1✔
323	changed_files: tuple[str, ...] = ()	1✔
324
325	@classmethod	1✔
326	def from_snapshots(cls, ours: Snapshot, theirs: Snapshot) -> SnapshotDiff:	1✔
UNCOV 327	return cls(*ours._diff(theirs))	×
328
329
330	@dataclass(frozen=True)	1✔
331	class PathMetadataRequest:	1✔
332	"""Request the full metadata of a single path in the filesystem.
333
334	Note: This API is symlink-aware and will distinguish between symlinks and regular files.
335	"""
336
337	path: str	1✔
338	namespace: PathNamespace = PathNamespace.WORKSPACE	1✔
339
340
341	@dataclass(frozen=True)	1✔
342	class PathMetadataResult:	1✔
343	"""Result of requesting the metadata for a path in the filesystem.
344
345	The `metadata` field will contain the metadata for the requested path, or else `None` if the
346	path does not exist.
347	"""
348
349	metadata: PathMetadata \| None	1✔
350
351
352	def rules():	1✔
353	# Avoids an import cycle.
UNCOV 354	from pants.engine.rules import QueryRule	×
355
UNCOV 356	return (	×
357	QueryRule(Digest, (CreateDigest,)),
358	QueryRule(Digest, (PathGlobs,)),
359	QueryRule(Digest, (AddPrefix,)),
360	QueryRule(Digest, (RemovePrefix,)),
361	QueryRule(Digest, (NativeDownloadFile,)),
362	QueryRule(Digest, (MergeDigests,)),
363	QueryRule(Digest, (DigestSubset,)),
364	QueryRule(DigestContents, (Digest,)),
365	QueryRule(Snapshot, (Digest,)),
366	QueryRule(Paths, (PathGlobs,)),
367	)

pantsbuild / pants / 19015773527

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