19250292619

Committed 11 Nov 2025 12:09AM UTC coverage: 77.865% (-2.4%) from 80.298%

Build # 19250292619

Build Type

push

github

Committed by

web-flow

Commit Message

flag non-runnable targets used with `code_quality_tool` (#22875)

Run Details

2 of 5 new or added lines in 2 files covered. (40.0%)

1487 existing lines in 72 files now uncovered.

71448 of 91759 relevant lines covered (77.86%)

3.22 hits per line

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

92.19

/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	11✔
5
6	from collections.abc import Iterable, Mapping, Sequence	11✔
7	from dataclasses import dataclass	11✔
8	from datetime import timedelta	11✔
9	from enum import Enum	11✔
10	from typing import TYPE_CHECKING, Union	11✔
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	11✔
14	from pants.engine.collection import Collection	11✔
15	from pants.engine.engine_aware import SideEffecting	11✔
16	from pants.engine.internals.native_engine import EMPTY_DIGEST as EMPTY_DIGEST # noqa: F401	11✔
17	from pants.engine.internals.native_engine import ( # noqa: F401	11✔
18	EMPTY_FILE_DIGEST as EMPTY_FILE_DIGEST,
19	)
20	from pants.engine.internals.native_engine import EMPTY_SNAPSHOT as EMPTY_SNAPSHOT # noqa: F401	11✔
21	from pants.engine.internals.native_engine import AddPrefix as AddPrefix	11✔
22	from pants.engine.internals.native_engine import Digest as Digest	11✔
23	from pants.engine.internals.native_engine import FileDigest as FileDigest	11✔
24	from pants.engine.internals.native_engine import MergeDigests as MergeDigests	11✔
25	from pants.engine.internals.native_engine import PathMetadata, PathNamespace	11✔
26	from pants.engine.internals.native_engine import RemovePrefix as RemovePrefix	11✔
27	from pants.engine.internals.native_engine import Snapshot as Snapshot	11✔
28	from pants.util.frozendict import FrozenDict	11✔
29
30	if TYPE_CHECKING:
31	from pants.engine.internals.scheduler import SchedulerSession
32
33
34	@dataclass(frozen=True)	11✔
35	class Paths:	11✔
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, ...]	11✔
43	dirs: tuple[str, ...]	11✔
44
45
46	@dataclass(frozen=True)	11✔
47	class FileContent:	11✔
48	"""The content of a file."""
49
50	path: str	11✔
51	content: bytes	11✔
52	is_executable: bool = False	11✔
53
54	def __post_init__(self):	11✔
55	if not isinstance(self.content, bytes):	11✔
UNCOV 56	raise TypeError(	×
57	f"Expected 'content' to be bytes, but got {type(self.content).__name__}"
58	)
59
60	def __repr__(self) -> str:	11✔
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)	11✔
68	class FileEntry:	11✔
69	"""An indirect reference to the content of a file by digest."""
70
71	path: str	11✔
72	file_digest: FileDigest	11✔
73	is_executable: bool = False	11✔
74
75
76	@dataclass(frozen=True)	11✔
77	class SymlinkEntry:	11✔
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	11✔
91	target: str	11✔
92
93
94	@dataclass(frozen=True)	11✔
95	class Directory:	11✔
96	"""The path to a directory."""
97
98	path: str	11✔
99
100	def __repr__(self) -> str:	11✔
101	return f"Directory({repr(self.path)})"	×
102
103
104	class DigestContents(Collection[FileContent]):	11✔
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]]):	11✔
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]]):	11✔
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):	11✔
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"	11✔
143	all_match = "all_match"	11✔
144
145
146	@dataclass(frozen=True)	11✔
147	class PathGlobs:	11✔
148	globs: tuple[str, ...]	11✔
149	glob_match_error_behavior: GlobMatchErrorBehavior	11✔
150	conjunction: GlobExpansionConjunction	11✔
151	description_of_origin: str \| None	11✔
152
153	def __init__(	11✔
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.
176	object.__setattr__(self, "globs", tuple(sorted(globs)))	11✔
177	object.__setattr__(self, "glob_match_error_behavior", glob_match_error_behavior)	11✔
178	object.__setattr__(self, "conjunction", conjunction)	11✔
179	object.__setattr__(self, "description_of_origin", description_of_origin)	11✔
180	self.__post_init__()	11✔
181
182	def __post_init__(self) -> None:	11✔
183	if self.glob_match_error_behavior == GlobMatchErrorBehavior.ignore:	11✔
184	if self.description_of_origin:	11✔
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:
194	if not self.description_of_origin:	1✔
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)	11✔
202	class PathGlobsAndRoot:	11✔
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	11✔
213	root: str	11✔
214	digest_hint: Digest \| None = None	11✔
215
216
217	@dataclass(frozen=True)	11✔
218	class DigestSubset:	11✔
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	11✔
227	globs: PathGlobs	11✔
228
229
230	@dataclass(frozen=True)	11✔
231	class DownloadFile:	11✔
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	11✔
240	expected_digest: FileDigest	11✔
241
242
243	@dataclass(frozen=True)	11✔
244	class NativeDownloadFile:	11✔
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	11✔
257	expected_digest: FileDigest	11✔
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]	11✔
261
262	retry_error_duration: timedelta	11✔
263	max_attempts: int	11✔
264
265	def __init__(	11✔
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)	11✔
281	class Workspace(SideEffecting):	11✔
282	"""A handle for operations that mutate the local filesystem."""
283
284	_scheduler: SchedulerSession	11✔
285	_enforce_effects: bool = True	11✔
286
287	def write_digest(	11✔
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	"""
303	if side_effecting:	2✔
304	self.side_effected()	2✔
305	self._scheduler.write_digest(digest, path_prefix=path_prefix, clear_paths=clear_paths)	2✔
306
307
308	@dataclass(frozen=True)	11✔
309	class SpecsPaths(Paths):	11✔
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)	11✔
318	class SnapshotDiff:	11✔
319	our_unique_files: tuple[str, ...] = ()	11✔
320	our_unique_dirs: tuple[str, ...] = ()	11✔
321	their_unique_files: tuple[str, ...] = ()	11✔
322	their_unique_dirs: tuple[str, ...] = ()	11✔
323	changed_files: tuple[str, ...] = ()	11✔
324
325	@classmethod	11✔
326	def from_snapshots(cls, ours: Snapshot, theirs: Snapshot) -> SnapshotDiff:	11✔
327	return cls(*ours._diff(theirs))	1✔
328
329
330	@dataclass(frozen=True)	11✔
331	class PathMetadataRequest:	11✔
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	11✔
338	namespace: PathNamespace = PathNamespace.WORKSPACE	11✔
339
340
341	@dataclass(frozen=True)	11✔
342	class PathMetadataResult:	11✔
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	11✔
350
351
352	def rules():	11✔
353	# Avoids an import cycle.
354	from pants.engine.rules import QueryRule	11✔
355
356	return (	11✔
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 / 19250292619

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