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

pantsbuild / pants / 23818810901

31 Mar 2026 08:51PM UTC coverage: 52.374% (-40.5%) from 92.907%
23818810901

Pull #23204

github

web-flow
Merge 89fac2a7e into 0c78ceb96
Pull Request #23204: Port ScalarField, AsyncFieldMixin and friends to rust

9 of 13 new or added lines in 2 files covered. (69.23%)

23031 existing lines in 605 files now uncovered.

31644 of 60419 relevant lines covered (52.37%)

0.52 hits per line

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

79.64
/src/python/pants/engine/target.py
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
import collections.abc
1✔
7
import dataclasses
1✔
8
import enum
1✔
9
import glob as glob_stdlib
1✔
10
import itertools
1✔
11
import logging
1✔
12
import os.path
1✔
13
import textwrap
1✔
14
import zlib
1✔
15
from abc import ABC, ABCMeta, abstractmethod
1✔
16
from collections import deque
1✔
17
from collections.abc import Callable, Iterable, Iterator, KeysView, Mapping, Sequence
1✔
18
from dataclasses import dataclass
1✔
19
from enum import Enum
1✔
20
from operator import attrgetter
1✔
21
from pathlib import PurePath
1✔
22
from typing import (
1✔
23
    AbstractSet,
24
    Any,
25
    ClassVar,
26
    Generic,
27
    Protocol,
28
    TypeVar,
29
    cast,
30
    final,
31
    get_type_hints,
32
)
33

34
from pants.base.deprecated import warn_or_error
1✔
35
from pants.engine.addresses import Address, Addresses, UnparsedAddressInputs, assert_single_address
1✔
36
from pants.engine.collection import Collection
1✔
37
from pants.engine.engine_aware import EngineAwareParameter
1✔
38
from pants.engine.environment import EnvironmentName
1✔
39
from pants.engine.fs import (
1✔
40
    GlobExpansionConjunction,
41
    GlobMatchErrorBehavior,
42
    PathGlobs,
43
    Paths,
44
    Snapshot,
45
)
46
from pants.engine.internals.dep_rules import (
1✔
47
    DependencyRuleActionDeniedError,
48
    DependencyRuleApplication,
49
)
50
from pants.engine.internals.native_engine import NO_VALUE as NO_VALUE  # noqa: F401
1✔
51
from pants.engine.internals.native_engine import AsyncFieldMixin as AsyncFieldMixin
1✔
52
from pants.engine.internals.native_engine import Field as Field
1✔
53
from pants.engine.internals.native_engine import ScalarField as ScalarField
1✔
54
from pants.engine.internals.native_engine import SequenceField as SequenceField  # noqa: F401
1✔
55
from pants.engine.internals.native_engine import StringField as StringField
1✔
56
from pants.engine.internals.native_engine import StringSequenceField as StringSequenceField
1✔
57
from pants.engine.internals.target_adaptor import SourceBlock, SourceBlocks  # noqa: F401
1✔
58
from pants.engine.rules import rule
1✔
59
from pants.engine.unions import UnionMembership, UnionRule, distinct_union_type_per_subclass, union
1✔
60
from pants.option.bootstrap_options import UnmatchedBuildFileGlobs
1✔
61
from pants.source.filespec import Filespec, FilespecMatcher
1✔
62
from pants.util.collections import ensure_str_list
1✔
63
from pants.util.dirutil import fast_relpath
1✔
64
from pants.util.docutil import bin_name, doc_url
1✔
65
from pants.util.frozendict import FrozenDict
1✔
66
from pants.util.memo import memoized_classproperty, memoized_method, memoized_property
1✔
67
from pants.util.ordered_set import FrozenOrderedSet
1✔
68
from pants.util.strutil import bullet_list, help_text, pluralize, softwrap
1✔
69

70
logger = logging.getLogger(__name__)
1✔
71

72
# -----------------------------------------------------------------------------------------------
73
# Core Field abstractions
74
# -----------------------------------------------------------------------------------------------
75

76
# Type alias to express the intent that the type should be immutable and hashable. There's nothing
77
# to actually enforce this, outside of convention. Maybe we could develop a MyPy plugin?
78
ImmutableValue = Any
1✔
79

80

81
@union
1✔
82
@dataclass(frozen=True)
1✔
83
class FieldDefaultFactoryRequest:
1✔
84
    """Registers a dynamic default for a Field.
85

86
    See `FieldDefaults`.
87
    """
88

89
    field_type: ClassVar[type[Field]]
1✔
90

91

92
# TODO: Workaround for https://github.com/python/mypy/issues/5485, because we cannot directly use
93
# a Callable.
94
class FieldDefaultFactory(Protocol):
1✔
95
    def __call__(self, field: Field) -> Any:
1✔
96
        pass
×
97

98

99
@dataclass(frozen=True)
1✔
100
class FieldDefaultFactoryResult:
1✔
101
    """A wrapper for a function which computes the default value of a Field."""
102

103
    default_factory: FieldDefaultFactory
1✔
104

105

106
@dataclass(frozen=True)
1✔
107
class FieldDefaults:
1✔
108
    """Generic Field default values. To install a default, see `FieldDefaultFactoryRequest`.
109

110
    TODO: This is to work around the fact that Field value defaulting cannot have arbitrary
111
    subsystem requirements, and so e.g. `JvmResolveField` and `PythonResolveField` have methods
112
    which compute the true value of the field given a subsystem argument. Consumers need to
113
    be type aware, and `@rules` cannot have dynamic requirements.
114

115
    Additionally, `__defaults__` should mean that computed default Field values should become
116
    more rare: i.e. `JvmResolveField` and `PythonResolveField` could potentially move to
117
    hardcoded default values which users override with `__defaults__` if they'd like to change
118
    the default resolve names.
119

120
    See https://github.com/pantsbuild/pants/issues/12934 about potentially allowing unions
121
    (including Field registrations) to have `@rule_helper` methods, which would allow the
122
    computation of an AsyncField to directly require a subsystem. Since
123
    https://github.com/pantsbuild/pants/pull/17947 rules may use any methods as rule helpers without
124
    special decoration so this should now be possible to implement.
125
    """
126

127
    _factories: FrozenDict[type[Field], FieldDefaultFactory]
1✔
128

129
    @memoized_method
1✔
130
    def factory(self, field_type: type[Field]) -> FieldDefaultFactory:
1✔
131
        """Looks up a Field default factory in a subclass-aware way."""
UNCOV
132
        factory = self._factories.get(field_type, None)
×
UNCOV
133
        if factory is not None:
×
UNCOV
134
            return factory
×
135

UNCOV
136
        for ft, factory in self._factories.items():
×
UNCOV
137
            if issubclass(field_type, ft):
×
UNCOV
138
                return factory
×
139

UNCOV
140
        return lambda f: f.value
×
141

142
    def value_or_default(self, field: Field) -> Any:
1✔
UNCOV
143
        return (self.factory(type(field)))(field)
×
144

145

146
# -----------------------------------------------------------------------------------------------
147
# Core Target abstractions
148
# -----------------------------------------------------------------------------------------------
149

150

151
# NB: This TypeVar is what allows `Target.get()` to properly work with MyPy so that MyPy knows
152
# the precise Field returned.
153
_F = TypeVar("_F", bound=Field)
1✔
154

155

156
@dataclass(frozen=True)
1✔
157
class Target:
1✔
158
    """A Target represents an addressable set of metadata.
159

160
    Set the `help` class property with a description, which will be used in `./pants help`. For the
161
    best rendering, use soft wrapping (e.g. implicit string concatenation) within paragraphs, but
162
    hard wrapping (`\n`) to separate distinct paragraphs and/or lists.
163
    """
164

165
    # Subclasses must define these
166
    alias: ClassVar[str]
1✔
167
    core_fields: ClassVar[tuple[type[Field], ...]]
1✔
168
    help: ClassVar[str | Callable[[], str]]
1✔
169

170
    removal_version: ClassVar[str | None] = None
1✔
171
    removal_hint: ClassVar[str | None] = None
1✔
172

173
    deprecated_alias: ClassVar[str | None] = None
1✔
174
    deprecated_alias_removal_version: ClassVar[str | None] = None
1✔
175

176
    # These get calculated in the constructor
177
    address: Address
1✔
178
    field_values: FrozenDict[type[Field], Field]
1✔
179
    residence_dir: str
1✔
180
    name_explicitly_set: bool
1✔
181
    description_of_origin: str
1✔
182
    origin_sources_blocks: FrozenDict[str, SourceBlocks]
1✔
183

184
    @final
1✔
185
    def __init__(
1✔
186
        self,
187
        unhydrated_values: Mapping[str, Any],
188
        address: Address,
189
        # NB: `union_membership` is only optional to facilitate tests. In production, we should
190
        # always provide this parameter. This should be safe to do because production code should
191
        # rarely directly instantiate Targets and should instead use the engine to request them.
192
        union_membership: UnionMembership | None = None,
193
        *,
194
        name_explicitly_set: bool = True,
195
        residence_dir: str | None = None,
196
        ignore_unrecognized_fields: bool = False,
197
        description_of_origin: str | None = None,
198
        origin_sources_blocks: FrozenDict[str, SourceBlocks] = FrozenDict(),
199
    ) -> None:
200
        """Create a target.
201

202
        :param unhydrated_values: A mapping of field aliases to their raw values. Any left off
203
            fields will either use their default or error if required=True.
204
        :param address: How to uniquely identify this target.
205
        :param union_membership: Used to determine plugin fields. This must be set in production!
206
        :param residence_dir: Where this target "lives". If unspecified, will be the `spec_path`
207
            of the `address`, i.e. where the target was either explicitly defined or where its
208
            target generator was explicitly defined. Target generators can, however, set this to
209
            the directory where the generated target provides metadata for. For example, a
210
            file-based target like `python_source` should set this to the parent directory of
211
            its file. A file-less target like `go_third_party_package` should keep the default of
212
            `address.spec_path`. This field impacts how command line specs work, so that globs
213
            like `dir:` know whether to match the target or not.
214
        :param ignore_unrecognized_fields: Don't error if fields are not recognized. This is only
215
            intended for when Pants is bootstrapping itself.
216
        :param description_of_origin: Where this target was declared, such as a path to BUILD file
217
            and line number.
218
        """
219
        if self.removal_version and not address.is_generated_target:
1✔
220
            if not self.removal_hint:
×
221
                raise ValueError(
×
222
                    f"You specified `removal_version` for {self.__class__}, but not "
223
                    "the class property `removal_hint`."
224
                )
225
            warn_or_error(
×
226
                self.removal_version,
227
                entity=f"the {repr(self.alias)} target type",
228
                hint=f"Using the `{self.alias}` target type for {address}. {self.removal_hint}",
229
            )
230

231
        if origin_sources_blocks:
1✔
UNCOV
232
            _validate_origin_sources_blocks(origin_sources_blocks)
×
233

234
        object.__setattr__(
1✔
235
            self, "residence_dir", residence_dir if residence_dir is not None else address.spec_path
236
        )
237
        object.__setattr__(self, "address", address)
1✔
238
        object.__setattr__(
1✔
239
            self, "description_of_origin", description_of_origin or self.residence_dir
240
        )
241
        object.__setattr__(self, "origin_sources_blocks", origin_sources_blocks)
1✔
242
        object.__setattr__(self, "name_explicitly_set", name_explicitly_set)
1✔
243
        try:
1✔
244
            object.__setattr__(
1✔
245
                self,
246
                "field_values",
247
                self._calculate_field_values(
248
                    unhydrated_values,
249
                    address,
250
                    union_membership,
251
                    ignore_unrecognized_fields=ignore_unrecognized_fields,
252
                ),
253
            )
254

255
            self.validate()
1✔
256
        except Exception as e:
1✔
257
            raise InvalidTargetException(
1✔
258
                str(e), description_of_origin=self.description_of_origin
259
            ) from e
260

261
    @final
1✔
262
    def _calculate_field_values(
1✔
263
        self,
264
        unhydrated_values: Mapping[str, Any],
265
        address: Address,
266
        # See `__init__`.
267
        union_membership: UnionMembership | None,
268
        *,
269
        ignore_unrecognized_fields: bool,
270
    ) -> FrozenDict[type[Field], Field]:
271
        all_field_types = self.class_field_types(union_membership)
1✔
272
        field_values = {}
1✔
273
        aliases_to_field_types = self._get_field_aliases_to_field_types(all_field_types)
1✔
274

275
        for alias, value in unhydrated_values.items():
1✔
276
            if alias not in aliases_to_field_types:
1✔
277
                if ignore_unrecognized_fields:
1✔
UNCOV
278
                    continue
×
279
                valid_aliases = set(aliases_to_field_types.keys())
1✔
280
                if isinstance(self, TargetGenerator):
1✔
281
                    # Even though moved_fields don't live on the target generator, they are valid
282
                    # for users to specify. It's intentional that these are only used for
283
                    # `InvalidFieldException` and are not stored as normal fields with
284
                    # `aliases_to_field_types`.
UNCOV
285
                    for field_type in self.moved_fields:
×
UNCOV
286
                        valid_aliases.add(field_type.alias)
×
UNCOV
287
                        if field_type.deprecated_alias is not None:
×
UNCOV
288
                            valid_aliases.add(field_type.deprecated_alias)
×
289
                raise InvalidFieldException(
1✔
290
                    f"Unrecognized field `{alias}={value}` in target {address}. Valid fields for "
291
                    f"the target type `{self.alias}`: {sorted(valid_aliases)}.",
292
                )
293
            field_type = aliases_to_field_types[alias]
1✔
294
            field_values[field_type] = field_type(value, address)
1✔
295

296
        # For undefined fields, mark the raw value as missing.
297
        for field_type in all_field_types:
1✔
298
            if field_type in field_values:
1✔
299
                continue
1✔
300
            field_values[field_type] = field_type(NO_VALUE, address)
1✔
301
        return FrozenDict(
1✔
302
            sorted(
303
                field_values.items(),
304
                key=lambda field_type_to_val_pair: field_type_to_val_pair[0].alias,
305
            )
306
        )
307

308
    @final
1✔
309
    @classmethod
1✔
310
    def _get_field_aliases_to_field_types(
1✔
311
        cls, field_types: Iterable[type[Field]]
312
    ) -> dict[str, type[Field]]:
313
        aliases_to_field_types = {}
1✔
314
        for field_type in field_types:
1✔
315
            aliases_to_field_types[field_type.alias] = field_type
1✔
316
            if field_type.deprecated_alias is not None:
1✔
UNCOV
317
                aliases_to_field_types[field_type.deprecated_alias] = field_type
×
318
        return aliases_to_field_types
1✔
319

320
    @final
1✔
321
    @property
1✔
322
    def field_types(self) -> KeysView[type[Field]]:
1✔
323
        return self.field_values.keys()
1✔
324

325
    @distinct_union_type_per_subclass
1✔
326
    class PluginField:
1✔
327
        pass
1✔
328

329
    def __repr__(self) -> str:
1✔
330
        fields = ", ".join(str(field) for field in self.field_values.values())
×
331
        return (
×
332
            f"{self.__class__}("
333
            f"address={self.address}, "
334
            f"alias={self.alias!r}, "
335
            f"residence_dir={self.residence_dir!r}, "
336
            f"origin={self.description_of_origin}, "
337
            f"{fields})"
338
        )
339

340
    def __str__(self) -> str:
1✔
UNCOV
341
        fields = ", ".join(str(field) for field in self.field_values.values())
×
UNCOV
342
        address = f'address="{self.address}"{", " if fields else ""}'
×
UNCOV
343
        return f"{self.alias}({address}{fields})"
×
344

345
    def __hash__(self) -> int:
1✔
346
        return hash((self.__class__, self.address, self.residence_dir, self.field_values))
1✔
347

348
    def __eq__(self, other: Target | Any) -> bool:
1✔
349
        if not isinstance(other, Target):
1✔
UNCOV
350
            return NotImplemented
×
351
        return (self.__class__, self.address, self.residence_dir, self.field_values) == (
1✔
352
            other.__class__,
353
            other.address,
354
            other.residence_dir,
355
            other.field_values,
356
        )
357

358
    def __lt__(self, other: Any) -> bool:
1✔
UNCOV
359
        if not isinstance(other, Target):
×
360
            return NotImplemented
×
UNCOV
361
        return self.address < other.address
×
362

363
    def __gt__(self, other: Any) -> bool:
1✔
364
        if not isinstance(other, Target):
×
365
            return NotImplemented
×
366
        return self.address > other.address
×
367

368
    @classmethod
1✔
369
    @memoized_method
1✔
370
    def _find_plugin_fields(cls, union_membership: UnionMembership) -> tuple[type[Field], ...]:
1✔
371
        result: set[type[Field]] = set()
1✔
372
        classes = [cls]
1✔
373
        while classes:
1✔
374
            cls = classes.pop()
1✔
375
            classes.extend(cls.__bases__)
1✔
376
            if issubclass(cls, Target):
1✔
377
                result.update(cast("set[type[Field]]", union_membership.get(cls.PluginField)))
1✔
378

379
        return tuple(sorted(result, key=attrgetter("alias")))
1✔
380

381
    @final
1✔
382
    @classmethod
1✔
383
    def _find_registered_field_subclass(
1✔
384
        cls, requested_field: type[_F], *, registered_fields: Iterable[type[Field]]
385
    ) -> type[_F] | None:
386
        """Check if the Target has registered a subclass of the requested Field.
387

388
        This is necessary to allow targets to override the functionality of common fields. For
389
        example, you could subclass `Tags` to define `CustomTags` with a different default. At the
390
        same time, we still want to be able to call `tgt.get(Tags)`, in addition to
391
        `tgt.get(CustomTags)`.
392
        """
393
        subclass = next(
1✔
394
            (
395
                registered_field
396
                for registered_field in registered_fields
397
                if issubclass(registered_field, requested_field)
398
            ),
399
            None,
400
        )
401
        return subclass
1✔
402

403
    @final
1✔
404
    def _maybe_get(self, field: type[_F]) -> _F | None:
1✔
405
        result = self.field_values.get(field, None)
1✔
406
        if result is not None:
1✔
407
            return cast(_F, result)
1✔
408
        field_subclass = self._find_registered_field_subclass(
1✔
409
            field, registered_fields=self.field_values.keys()
410
        )
411
        if field_subclass is not None:
1✔
412
            return cast(_F, self.field_values[field_subclass])
1✔
413
        return None
1✔
414

415
    @final
1✔
416
    def __getitem__(self, field: type[_F]) -> _F:
1✔
417
        """Get the requested `Field` instance belonging to this target.
418

419
        If the `Field` is not registered on this `Target` type, this method will raise a
420
        `KeyError`. To avoid this, you should first call `tgt.has_field()` or `tgt.has_fields()`
421
        to ensure that the field is registered, or, alternatively, use `Target.get()`.
422

423
        See the docstring for `Target.get()` for how this method handles subclasses of the
424
        requested Field and for tips on how to use the returned value.
425
        """
426
        result = self._maybe_get(field)
1✔
427
        if result is not None:
1✔
428
            return result
1✔
UNCOV
429
        raise KeyError(
×
430
            f"The target `{self}` does not have a field `{field.__name__}`. Before calling "
431
            f"`my_tgt[{field.__name__}]`, call `my_tgt.has_field({field.__name__})` to "
432
            f"filter out any irrelevant Targets or call `my_tgt.get({field.__name__})` to use the "
433
            f"default Field value."
434
        )
435

436
    @final
1✔
437
    def get(self, field: type[_F], *, default_raw_value: Any | None = None) -> _F:
1✔
438
        """Get the requested `Field` instance belonging to this target.
439

440
        This will return an instance of the requested field type, e.g. an instance of
441
        `InterpreterConstraints`, `SourcesField`, `EntryPoint`, etc. Usually, you will want to
442
        grab the `Field`'s inner value, e.g. `tgt.get(Compatibility).value`. (For async fields like
443
        `SourcesField`, you may need to hydrate the value.).
444

445
        This works with subclasses of `Field`. For example, if you subclass `Tags`
446
        to define a custom subclass `CustomTags`, both `tgt.get(Tags)` and
447
        `tgt.get(CustomTags)` will return the same `CustomTags` instance.
448

449
        If the `Field` is not registered on this `Target` type, this will return an instance of
450
        the requested Field by using `default_raw_value` to create the instance. Alternatively,
451
        first call `tgt.has_field()` or `tgt.has_fields()` to ensure that the field is registered,
452
        or, alternatively, use indexing (e.g. `tgt[Compatibility]`) to raise a KeyError when the
453
        field is not registered.
454
        """
455
        result = self._maybe_get(field)
1✔
456
        if result is not None:
1✔
457
            return result
1✔
458
        return field(default_raw_value, self.address)
1✔
459

460
    @final
1✔
461
    @classmethod
1✔
462
    def _has_fields(
1✔
463
        cls, fields: Iterable[type[Field]], *, registered_fields: AbstractSet[type[Field]]
464
    ) -> bool:
465
        unrecognized_fields = [field for field in fields if field not in registered_fields]
1✔
466
        if not unrecognized_fields:
1✔
467
            return True
1✔
468
        for unrecognized_field in unrecognized_fields:
1✔
469
            maybe_subclass = cls._find_registered_field_subclass(
1✔
470
                unrecognized_field, registered_fields=registered_fields
471
            )
472
            if maybe_subclass is None:
1✔
473
                return False
1✔
474
        return True
1✔
475

476
    @final
1✔
477
    def has_field(self, field: type[Field]) -> bool:
1✔
478
        """Check that this target has registered the requested field.
479

480
        This works with subclasses of `Field`. For example, if you subclass `Tags` to define a
481
        custom subclass `CustomTags`, both `tgt.has_field(Tags)` and
482
        `python_tgt.has_field(CustomTags)` will return True.
483
        """
484
        return self.has_fields([field])
1✔
485

486
    @final
1✔
487
    def has_fields(self, fields: Iterable[type[Field]]) -> bool:
1✔
488
        """Check that this target has registered all of the requested fields.
489

490
        This works with subclasses of `Field`. For example, if you subclass `Tags` to define a
491
        custom subclass `CustomTags`, both `tgt.has_fields([Tags])` and
492
        `python_tgt.has_fields([CustomTags])` will return True.
493
        """
494
        return self._has_fields(fields, registered_fields=self.field_values.keys())
1✔
495

496
    @final
1✔
497
    @classmethod
1✔
498
    @memoized_method
1✔
499
    def class_field_types(
1✔
500
        cls, union_membership: UnionMembership | None
501
    ) -> FrozenOrderedSet[type[Field]]:
502
        """Return all registered Fields belonging to this target type.
503

504
        You can also use the instance property `tgt.field_types` to avoid having to pass the
505
        parameter UnionMembership.
506
        """
507
        if union_membership is None:
1✔
508
            return FrozenOrderedSet(cls.core_fields)
1✔
509
        else:
510
            return FrozenOrderedSet((*cls.core_fields, *cls._find_plugin_fields(union_membership)))
1✔
511

512
    @final
1✔
513
    @classmethod
1✔
514
    def class_has_field(cls, field: type[Field], union_membership: UnionMembership) -> bool:
1✔
515
        """Behaves like `Target.has_field()`, but works as a classmethod rather than an instance
516
        method."""
UNCOV
517
        return cls.class_has_fields([field], union_membership)
×
518

519
    @final
1✔
520
    @classmethod
1✔
521
    def class_has_fields(
1✔
522
        cls, fields: Iterable[type[Field]], union_membership: UnionMembership
523
    ) -> bool:
524
        """Behaves like `Target.has_fields()`, but works as a classmethod rather than an instance
525
        method."""
UNCOV
526
        return cls._has_fields(fields, registered_fields=cls.class_field_types(union_membership))
×
527

528
    @final
1✔
529
    @classmethod
1✔
530
    def class_get_field(cls, field: type[_F], union_membership: UnionMembership) -> type[_F]:
1✔
531
        """Get the requested Field type registered with this target type.
532

533
        This will error if the field is not registered, so you should call Target.class_has_field()
534
        first.
535
        """
UNCOV
536
        class_fields = cls.class_field_types(union_membership)
×
UNCOV
537
        result = next(
×
538
            (
539
                registered_field
540
                for registered_field in class_fields
541
                if issubclass(registered_field, field)
542
            ),
543
            None,
544
        )
UNCOV
545
        if result is None:
×
UNCOV
546
            raise KeyError(
×
547
                f"The target type `{cls.alias}` does not have a field `{field.__name__}`. Before "
548
                f"calling `TargetType.class_get_field({field.__name__})`, call "
549
                f"`TargetType.class_has_field({field.__name__})`."
550
            )
UNCOV
551
        return result
×
552

553
    @classmethod
1✔
554
    def register_plugin_field(cls, field: type[Field]) -> UnionRule:
1✔
555
        """Register a new field on the target type.
556

557
        In the `rules()` register.py entry-point, include
558
        `MyTarget.register_plugin_field(NewField)`. This will register `NewField` as a first-class
559
        citizen. Plugins can use this new field like any other.
560
        """
561
        return UnionRule(cls.PluginField, field)
1✔
562

563
    def validate(self) -> None:
1✔
564
        """Validate the target, such as checking for mutually exclusive fields.
565

566
        N.B.: The validation should only be of properties intrinsic to the associated files in any
567
        context. If the validation only makes sense for certain goals acting on targets; those
568
        validations should be done in the associated rules.
569
        """
570

571

572
def _validate_origin_sources_blocks(origin_sources_blocks: FrozenDict[str, SourceBlocks]) -> None:
1✔
UNCOV
573
    if not isinstance(origin_sources_blocks, FrozenDict):
×
UNCOV
574
        raise ValueError(
×
575
            f"Expected `origin_sources_blocks` to be of type `FrozenDict`, got {type(origin_sources_blocks)=} {origin_sources_blocks=}"
576
        )
UNCOV
577
    for blocks in origin_sources_blocks.values():
×
UNCOV
578
        if not isinstance(blocks, SourceBlocks):
×
UNCOV
579
            raise ValueError(
×
580
                f"Expected `origin_sources_blocks` to be a `FrozenDict` with values of type `SourceBlocks`, got values of {type(blocks)=} {blocks=}"
581
            )
UNCOV
582
        for block in blocks:
×
UNCOV
583
            if not isinstance(block, SourceBlock):
×
584
                raise ValueError(
×
585
                    f"Expected `origin_sources_blocks` to be a `FrozenDict` with values of type `SourceBlocks`, got values of {type(blocks)=} {blocks=}"
586
                )
587

588

589
@dataclass(frozen=True)
1✔
590
class WrappedTargetRequest:
1✔
591
    """Used with `WrappedTarget` to get the Target corresponding to an address.
592

593
    `description_of_origin` is used for error messages when the address does not actually exist. If
594
    you are confident this cannot happen, set the string to something like `<infallible>`.
595
    """
596

597
    address: Address
1✔
598
    description_of_origin: str = dataclasses.field(hash=False, compare=False)
1✔
599

600

601
@dataclass(frozen=True)
1✔
602
class WrappedTarget:
1✔
603
    """A light wrapper to encapsulate all the distinct `Target` subclasses into a single type.
604

605
    This is necessary when using a single target in a rule because the engine expects exact types
606
    and does not work with subtypes.
607
    """
608

609
    target: Target
1✔
610

611

612
class Targets(Collection[Target]):
1✔
613
    """A heterogeneous collection of instances of Target subclasses.
614

615
    While every element will be a subclass of `Target`, there may be many different `Target` types
616
    in this collection, e.g. some `FileTarget` and some `PythonTestTarget`.
617

618
    Often, you will want to filter out the relevant targets by looking at what fields they have
619
    registered, e.g.:
620

621
        valid_tgts = [tgt for tgt in tgts if tgt.has_fields([Compatibility, PythonSources])]
622

623
    You should not check the Target's actual type because this breaks custom target types;
624
    for example, prefer `tgt.has_field(PythonTestsSourcesField)` to
625
    `isinstance(tgt, PythonTestsTarget)`.
626
    """
627

628
    def expect_single(self) -> Target:
1✔
UNCOV
629
        assert_single_address([tgt.address for tgt in self])
×
UNCOV
630
        return self[0]
×
631

632

633
# This distinct type is necessary because of https://github.com/pantsbuild/pants/issues/14977.
634
#
635
# NB: We still proactively apply filtering inside `AddressSpecs` and `FilesystemSpecs`, which is
636
# earlier in the rule pipeline of `RawSpecs -> Addresses -> UnexpandedTargets -> Targets ->
637
# FilteredTargets`. That is necessary so that project-introspection goals like `list` which don't
638
# use `FilteredTargets` still have filtering applied.
639
class FilteredTargets(Collection[Target]):
1✔
640
    """A heterogeneous collection of Target instances that have been filtered with the global
641
    options `--tag` and `--exclude-target-regexp`.
642

643
    Outside of the extra filtering, this type is identical to `Targets`, including its handling of
644
    target generators.
645
    """
646

647
    def expect_single(self) -> Target:
1✔
648
        assert_single_address([tgt.address for tgt in self])
×
649
        return self[0]
×
650

651

652
class UnexpandedTargets(Collection[Target]):
1✔
653
    """Like `Targets`, but will not replace target generators with their generated targets (e.g.
654
    replace `python_sources` "BUILD targets" with generated `python_source` "file targets")."""
655

656
    def expect_single(self) -> Target:
1✔
657
        assert_single_address([tgt.address for tgt in self])
×
658
        return self[0]
×
659

660

661
class DepsTraversalBehavior(Enum):
1✔
662
    """The return value for ShouldTraverseDepsPredicate.
663

664
    NB: This only indicates whether to traverse the deps of a target;
665
    It does not control the inclusion of the target itself (though that
666
    might be added in the future). By the time the predicate is called,
667
    the target itself was already included.
668
    """
669

670
    INCLUDE = "include"
1✔
671
    EXCLUDE = "exclude"
1✔
672

673

674
@dataclass(frozen=True)
1✔
675
class ShouldTraverseDepsPredicate(metaclass=ABCMeta):
1✔
676
    """This callable determines whether to traverse through deps of a given Target + Field.
677

678
    This is a callable dataclass instead of a function to avoid issues with hashing closures.
679
    Only the id is used when hashing a function; any closure vars are NOT accounted for.
680

681
    NB: When subclassing a dataclass (like this one), you do not need to add the @dataclass
682
    decorator unless the subclass has additional fields. The @dataclass decorator only inspects
683
    the current class (not any parents from __mro__) to determine if any methods were explicitly
684
    defined. So, any typically-generated methods explicitly added on this parent class would NOT
685
    be inherited by @dataclass decorated subclasses. To avoid these issues, this parent class
686
    uses __post_init__ and relies on the generated __init__ and __hash__ methods.
687
    """
688

689
    # NB: This _callable field ensures that __call__ is included in the __hash__ method generated by @dataclass.
690
    # That is extremely important because two predicates with different implementations but the same data
691
    # (or no data) need to have different hashes and compare unequal.
692
    _callable: Callable[
1✔
693
        [Any, Target, Dependencies | SpecialCasedDependencies], DepsTraversalBehavior
694
    ] = dataclasses.field(init=False, repr=False)
695

696
    def __post_init__(self):
1✔
697
        object.__setattr__(self, "_callable", type(self).__call__)
1✔
698

699
    @abstractmethod
1✔
700
    def __call__(
1✔
701
        self, target: Target, field: Dependencies | SpecialCasedDependencies
702
    ) -> DepsTraversalBehavior:
703
        """This predicate decides when to INCLUDE or EXCLUDE the target's field's deps."""
704

705

706
class TraverseIfDependenciesField(ShouldTraverseDepsPredicate):
1✔
707
    """This is the default ShouldTraverseDepsPredicate implementation.
708

709
    This skips resolving dependencies for fields (like SpecialCasedDependencies) that are not
710
    subclasses of Dependencies.
711
    """
712

713
    def __call__(
1✔
714
        self, target: Target, field: Dependencies | SpecialCasedDependencies
715
    ) -> DepsTraversalBehavior:
716
        if isinstance(field, Dependencies):
1✔
717
            return DepsTraversalBehavior.INCLUDE
1✔
718
        return DepsTraversalBehavior.EXCLUDE
1✔
719

720

721
class AlwaysTraverseDeps(ShouldTraverseDepsPredicate):
1✔
722
    """A predicate to use when a request needs all deps.
723

724
    This includes deps from fields like SpecialCasedDependencies which are ignored in most cases.
725
    """
726

727
    def __call__(
1✔
728
        self, target: Target, field: Dependencies | SpecialCasedDependencies
729
    ) -> DepsTraversalBehavior:
UNCOV
730
        return DepsTraversalBehavior.INCLUDE
×
731

732

733
class CoarsenedTarget(EngineAwareParameter):
1✔
734
    def __init__(self, members: Iterable[Target], dependencies: Iterable[CoarsenedTarget]) -> None:
1✔
735
        """A set of Targets which cyclically reach one another, and are thus indivisible.
736

737
        Instances of this class form a structure-shared DAG, and so a hashcode is pre-computed for the
738
        recursive portion.
739

740
        :param members: The members of the cycle.
741
        :param dependencies: The deduped direct (not transitive) dependencies of all Targets in
742
            the cycle. Dependencies between members of the cycle are excluded.
743
        """
744
        self.members = FrozenOrderedSet(members)
1✔
745
        self.dependencies = FrozenOrderedSet(dependencies)
1✔
746
        self._hashcode = hash((self.members, self.dependencies))
1✔
747

748
    def debug_hint(self) -> str:
1✔
749
        return str(self)
×
750

751
    def metadata(self) -> dict[str, Any]:
1✔
752
        return {"addresses": [t.address.spec for t in self.members]}
×
753

754
    @property
1✔
755
    def representative(self) -> Target:
1✔
756
        """A stable "representative" target in the cycle."""
UNCOV
757
        return next(iter(self.members))
×
758

759
    def bullet_list(self) -> str:
1✔
760
        """The addresses and type aliases of all members of the cycle."""
UNCOV
761
        return bullet_list(sorted(f"{t.address.spec}\t({type(t).alias})" for t in self.members))
×
762

763
    def closure(self, visited: set[CoarsenedTarget] | None = None) -> Iterator[Target]:
1✔
764
        """All Targets reachable from this root."""
765
        return (t for ct in self.coarsened_closure(visited) for t in ct.members)
1✔
766

767
    def coarsened_closure(
1✔
768
        self, visited: set[CoarsenedTarget] | None = None
769
    ) -> Iterator[CoarsenedTarget]:
770
        """All CoarsenedTargets reachable from this root."""
771

772
        visited = set() if visited is None else visited
1✔
773
        queue = deque([self])
1✔
774
        while queue:
1✔
775
            ct = queue.popleft()
1✔
776
            if ct in visited:
1✔
UNCOV
777
                continue
×
778
            visited.add(ct)
1✔
779
            yield ct
1✔
780
            queue.extend(ct.dependencies)
1✔
781

782
    def __hash__(self) -> int:
1✔
783
        return self._hashcode
1✔
784

785
    def _eq_helper(self, other: CoarsenedTarget, equal_items: set[tuple[int, int]]) -> bool:
1✔
786
        key = (id(self), id(other))
1✔
787
        if key[0] == key[1] or key in equal_items:
1✔
UNCOV
788
            return True
×
789

790
        is_eq = (
1✔
791
            self._hashcode == other._hashcode
792
            and self.members == other.members
793
            and len(self.dependencies) == len(other.dependencies)
794
            and all(
795
                l._eq_helper(r, equal_items) for l, r in zip(self.dependencies, other.dependencies)
796
            )
797
        )
798

799
        # NB: We only track equal items because any non-equal item will cause the entire
800
        # operation to shortcircuit.
801
        if is_eq:
1✔
802
            equal_items.add(key)
1✔
803
        return is_eq
1✔
804

805
    def __eq__(self, other: Any) -> bool:
1✔
UNCOV
806
        if not isinstance(other, CoarsenedTarget):
×
807
            return NotImplemented
×
UNCOV
808
        return self._eq_helper(other, set())
×
809

810
    def __str__(self) -> str:
1✔
UNCOV
811
        if len(self.members) > 1:
×
UNCOV
812
            others = len(self.members) - 1
×
UNCOV
813
            return f"{self.representative.address.spec} (and {others} more)"
×
UNCOV
814
        return self.representative.address.spec
×
815

816
    def __repr__(self) -> str:
1✔
817
        return f"{self.__class__.__name__}({str(self)})"
×
818

819

820
class CoarsenedTargets(Collection[CoarsenedTarget]):
1✔
821
    """The CoarsenedTarget roots of a transitive graph walk for some addresses.
822

823
    To collect all reachable CoarsenedTarget members, use `def closure`.
824
    """
825

826
    def by_address(self) -> dict[Address, CoarsenedTarget]:
1✔
827
        """Compute a mapping from Address to containing CoarsenedTarget."""
828
        return {t.address: ct for ct in self for t in ct.members}
1✔
829

830
    def closure(self) -> Iterator[Target]:
1✔
831
        """All Targets reachable from these CoarsenedTarget roots."""
832
        visited: set[CoarsenedTarget] = set()
1✔
833
        return (t for root in self for t in root.closure(visited))
1✔
834

835
    def coarsened_closure(self) -> Iterator[CoarsenedTarget]:
1✔
836
        """All CoarsenedTargets reachable from these CoarsenedTarget roots."""
UNCOV
837
        visited: set[CoarsenedTarget] = set()
×
UNCOV
838
        return (ct for root in self for ct in root.coarsened_closure(visited))
×
839

840
    def __eq__(self, other: Any) -> bool:
1✔
841
        if not isinstance(other, CoarsenedTargets):
1✔
842
            return NotImplemented
×
843
        equal_items: set[tuple[int, int]] = set()
1✔
844
        return len(self) == len(other) and all(
1✔
845
            l._eq_helper(r, equal_items) for l, r in zip(self, other)
846
        )
847

848
    def __hash__(self):
1✔
849
        return super().__hash__()
1✔
850

851

852
@dataclass(frozen=True)
1✔
853
class CoarsenedTargetsRequest:
1✔
854
    """A request to get CoarsenedTargets for input roots."""
855

856
    roots: tuple[Address, ...]
1✔
857
    expanded_targets: bool
1✔
858
    should_traverse_deps_predicate: ShouldTraverseDepsPredicate
1✔
859

860
    def __init__(
1✔
861
        self,
862
        roots: Iterable[Address],
863
        *,
864
        expanded_targets: bool = False,
865
        should_traverse_deps_predicate: ShouldTraverseDepsPredicate = TraverseIfDependenciesField(),
866
    ) -> None:
867
        object.__setattr__(self, "roots", tuple(roots))
1✔
868
        object.__setattr__(self, "expanded_targets", expanded_targets)
1✔
869
        object.__setattr__(self, "should_traverse_deps_predicate", should_traverse_deps_predicate)
1✔
870

871

872
@dataclass(frozen=True)
1✔
873
class TransitiveTargets:
1✔
874
    """A set of Target roots, and their transitive, flattened, de-duped dependencies.
875

876
    If a target root is a dependency of another target root, then it will show up both in `roots`
877
    and in `dependencies`.
878
    """
879

880
    roots: tuple[Target, ...]
1✔
881
    dependencies: FrozenOrderedSet[Target]
1✔
882

883
    @memoized_property
1✔
884
    def closure(self) -> FrozenOrderedSet[Target]:
1✔
885
        """The roots and the dependencies combined."""
886
        return FrozenOrderedSet([*self.roots, *self.dependencies])
1✔
887

888

889
@dataclass(frozen=True)
1✔
890
class TransitiveTargetsRequest:
1✔
891
    """A request to get the transitive dependencies of the input roots."""
892

893
    roots: tuple[Address, ...]
1✔
894
    should_traverse_deps_predicate: ShouldTraverseDepsPredicate
1✔
895

896
    def __init__(
1✔
897
        self,
898
        roots: Iterable[Address],
899
        *,
900
        should_traverse_deps_predicate: ShouldTraverseDepsPredicate = TraverseIfDependenciesField(),
901
    ) -> None:
902
        object.__setattr__(self, "roots", tuple(roots))
1✔
903
        object.__setattr__(self, "should_traverse_deps_predicate", should_traverse_deps_predicate)
1✔
904

905

906
@dataclass(frozen=True)
1✔
907
class RegisteredTargetTypes:
1✔
908
    aliases_to_types: FrozenDict[str, type[Target]]
1✔
909

910
    def __init__(self, aliases_to_types: Mapping[str, type[Target]]) -> None:
1✔
911
        object.__setattr__(self, "aliases_to_types", FrozenDict(aliases_to_types))
1✔
912

913
    @classmethod
1✔
914
    def create(cls, target_types: Iterable[type[Target]]) -> RegisteredTargetTypes:
1✔
915
        result = {}
1✔
916
        for target_type in sorted(target_types, key=lambda tt: tt.alias):
1✔
917
            result[target_type.alias] = target_type
1✔
918
            if target_type.deprecated_alias is not None:
1✔
UNCOV
919
                result[target_type.deprecated_alias] = target_type
×
920
        return cls(result)
1✔
921

922
    @property
1✔
923
    def aliases(self) -> FrozenOrderedSet[str]:
1✔
924
        return FrozenOrderedSet(self.aliases_to_types.keys())
1✔
925

926
    @property
1✔
927
    def types(self) -> FrozenOrderedSet[type[Target]]:
1✔
UNCOV
928
        return FrozenOrderedSet(self.aliases_to_types.values())
×
929

930

931
class AllTargets(Collection[Target]):
1✔
932
    """All targets in the project, but with target generators replaced by their generated targets,
933
    unlike `AllUnexpandedTargets`."""
934

935

936
class AllUnexpandedTargets(Collection[Target]):
1✔
937
    """All targets in the project, including generated targets.
938

939
    This should generally be avoided because it is relatively expensive to compute and is frequently
940
    invalidated, but it can be necessary for things like dependency inference to build a global
941
    mapping of imports to targets.
942
    """
943

944

945
# -----------------------------------------------------------------------------------------------
946
# Target generation
947
# -----------------------------------------------------------------------------------------------
948

949

950
class TargetGenerator(Target):
1✔
951
    """A Target type which generates other Targets via installed `@rule` logic.
952

953
    To act as a generator, a Target type should subclass this base class and install generation
954
    `@rule`s which consume a corresponding GenerateTargetsRequest subclass to produce
955
    GeneratedTargets.
956
    """
957

958
    # The generated Target class.
959
    #
960
    # If this is not provided, consider checking for the default values that applies to the target
961
    # types being generated manually. The applicable defaults are available on the `AddressFamily`
962
    # which you can get using:
963
    #
964
    #    family = await ensure_address_family(**implicitly(AddressFamilyDir(address.spec_path)))
965
    #    target_defaults = family.defaults.get(MyTarget.alias, {})
966
    generated_target_cls: ClassVar[type[Target]]
1✔
967

968
    # Fields which have their values copied from the generator Target to the generated Target.
969
    #
970
    # Must be a subset of `core_fields`.
971
    #
972
    # Fields should be copied from the generator to the generated when their semantic meaning is
973
    # the same for both Target types, and when it is valuable for them to be introspected on
974
    # either the generator or generated target (such as by `peek`, or in `filter`).
975
    copied_fields: ClassVar[tuple[type[Field], ...]]
1✔
976

977
    # Fields which are specified to instances of the generator Target, but which are propagated
978
    # to generated Targets rather than being stored on the generator Target.
979
    #
980
    # Must be disjoint from `core_fields`.
981
    #
982
    # Only Fields which are moved to the generated Target are allowed to be `parametrize`d. But
983
    # it can also be the case that a Field only makes sense semantically when it is applied to
984
    # the generated Target (for example, for an individual file), and the generator Target is just
985
    # acting as a convenient place for them to be specified.
986
    moved_fields: ClassVar[tuple[type[Field], ...]]
1✔
987

988
    @distinct_union_type_per_subclass
1✔
989
    class MovedPluginField:
1✔
990
        """A plugin field that should be moved into the generated targets."""
991

992
    def validate(self) -> None:
1✔
993
        super().validate()
1✔
994

995
        copied_dependencies_field_types = [
1✔
996
            field_type.__name__
997
            for field_type in type(self).copied_fields
998
            if issubclass(field_type, Dependencies)
999
        ]
1000
        if copied_dependencies_field_types:
1✔
1001
            raise InvalidTargetException(
×
1002
                f"Using a `Dependencies` field subclass ({copied_dependencies_field_types}) as a "
1003
                "`TargetGenerator.copied_field`. `Dependencies` fields should be "
1004
                "`TargetGenerator.moved_field`s, to avoid redundant graph edges."
1005
            )
1006

1007
    @classmethod
1✔
1008
    def register_plugin_field(cls, field: type[Field], *, as_moved_field=False) -> UnionRule:
1✔
1009
        if as_moved_field:
1✔
1010
            return UnionRule(cls.MovedPluginField, field)
1✔
1011
        else:
1012
            return super().register_plugin_field(field)
1✔
1013

1014
    @classmethod
1✔
1015
    @memoized_method
1✔
1016
    def _find_plugin_fields(cls, union_membership: UnionMembership) -> tuple[type[Field], ...]:
1✔
1017
        return (
1✔
1018
            *cls._find_copied_plugin_fields(union_membership),
1019
            *cls._find_moved_plugin_fields(union_membership),
1020
        )
1021

1022
    @final
1✔
1023
    @classmethod
1✔
1024
    @memoized_method
1✔
1025
    def _find_moved_plugin_fields(
1✔
1026
        cls, union_membership: UnionMembership
1027
    ) -> tuple[type[Field], ...]:
1028
        result: set[type[Field]] = set()
1✔
1029
        classes = [cls]
1✔
1030
        while classes:
1✔
1031
            cls = classes.pop()
1✔
1032
            classes.extend(cls.__bases__)
1✔
1033
            if issubclass(cls, TargetGenerator):
1✔
1034
                result.update(cast("set[type[Field]]", union_membership.get(cls.MovedPluginField)))
1✔
1035

1036
        return tuple(result)
1✔
1037

1038
    @final
1✔
1039
    @classmethod
1✔
1040
    @memoized_method
1✔
1041
    def _find_copied_plugin_fields(
1✔
1042
        cls, union_membership: UnionMembership
1043
    ) -> tuple[type[Field], ...]:
1044
        return super()._find_plugin_fields(union_membership)
1✔
1045

1046

1047
class TargetFilesGenerator(TargetGenerator):
1✔
1048
    """A TargetGenerator which generates a Target per file matched by the generator.
1049

1050
    Unlike TargetGenerator, no additional `@rules` are required to be installed, because generation
1051
    is implemented declaratively. But an optional `settings_request_cls` can be declared to
1052
    dynamically control some settings of generation.
1053
    """
1054

1055
    settings_request_cls: ClassVar[type[TargetFilesGeneratorSettingsRequest] | None] = None
1✔
1056

1057
    def validate(self) -> None:
1✔
1058
        super().validate()
1✔
1059

1060
        if self.has_field(MultipleSourcesField) and not self[MultipleSourcesField].value:
1✔
1061
            raise InvalidTargetException(
×
1062
                f"The `{self.alias}` target generator at {self.address} has an empty "
1063
                f"`{self[MultipleSourcesField].alias}` field; so it will not generate any targets. "
1064
                "If its purpose is to act as an alias for its dependencies, then it should be "
1065
                "declared as a `target(..)` generic target instead. If it is unused, then it "
1066
                "should be removed."
1067
            )
1068

1069

1070
@union(in_scope_types=[EnvironmentName])
1✔
1071
class TargetFilesGeneratorSettingsRequest:
1✔
1072
    """An optional union to provide dynamic settings for a `TargetFilesGenerator`.
1073

1074
    See `TargetFilesGenerator`.
1075
    """
1076

1077

1078
@dataclass
1✔
1079
class TargetFilesGeneratorSettings:
1✔
1080
    # Set `add_dependencies_on_all_siblings` to True so that each file-level target depends on all
1081
    # other generated targets from the target generator. This is useful if both are true:
1082
    #
1083
    # a) file-level targets usually need their siblings to be present to work. Most target types
1084
    #   (Python, Java, Shell, etc) meet this, except for `files` and `resources` which have no
1085
    #   concept of "imports"
1086
    # b) dependency inference cannot infer dependencies on sibling files.
1087
    #
1088
    # Otherwise, set `add_dependencies_on_all_siblings` to `False` so that dependencies are
1089
    # finer-grained.
1090
    add_dependencies_on_all_siblings: bool = False
1✔
1091

1092

1093
_TargetGenerator = TypeVar("_TargetGenerator", bound=TargetGenerator)
1✔
1094

1095

1096
@union(in_scope_types=[EnvironmentName])
1✔
1097
@dataclass(frozen=True)
1✔
1098
class GenerateTargetsRequest(Generic[_TargetGenerator]):
1✔
1099
    generate_from: ClassVar[type[_TargetGenerator]]
1✔
1100

1101
    # The TargetGenerator instance to generate targets for.
1102
    generator: _TargetGenerator
1✔
1103
    # The base Address to generate for. Note that due to parametrization, this may not
1104
    # always be the Address of the underlying target.
1105
    template_address: Address
1✔
1106
    # The `TargetGenerator.moved_field/copied_field` Field values that the generator
1107
    # should generate targets with.
1108
    template: Mapping[str, Any] = dataclasses.field(hash=False)
1✔
1109
    # Per-generated-Target overrides, with an additional `template_address` to be applied. The
1110
    # per-instance Address might not match the base `template_address` if parametrization was
1111
    # applied within overrides.
1112
    overrides: Mapping[str, Mapping[Address, Mapping[str, Any]]] = dataclasses.field(hash=False)
1✔
1113

1114
    def require_unparametrized_overrides(self) -> dict[str, Mapping[str, Any]]:
1✔
1115
        """Flattens overrides for `GenerateTargetsRequest` impls which don't support `parametrize`.
1116

1117
        If `parametrize` has been used in overrides, this will raise an error indicating that that is
1118
        not yet supported for the generator target type.
1119

1120
        TODO: https://github.com/pantsbuild/pants/issues/14430 covers porting implementations and
1121
        removing this method.
1122
        """
1123
        if any(len(templates) != 1 for templates in self.overrides.values()):
1✔
1124
            raise ValueError(
×
1125
                f"Target generators of type `{self.generate_from.alias}` (defined at "
1126
                f"`{self.generator.address}`) do not (yet) support use of the `parametrize(..)` "
1127
                f"builtin in their `{OverridesField.alias}=` field."
1128
            )
1129
        return {name: next(iter(templates.values())) for name, templates in self.overrides.items()}
1✔
1130

1131

1132
class GeneratedTargets(FrozenDict[Address, Target]):
1✔
1133
    """A mapping of the address of generated targets to the targets themselves."""
1134

1135
    def __new__(cls, generator: Target, generated_targets: Iterable[Target]) -> GeneratedTargets:
1✔
1136
        expected_spec_path = generator.address.spec_path
1✔
1137
        expected_tgt_name = generator.address.target_name
1✔
1138
        mapping = {}
1✔
1139
        for tgt in sorted(generated_targets, key=lambda t: t.address):
1✔
1140
            if tgt.address.spec_path != expected_spec_path:
1✔
UNCOV
1141
                raise InvalidGeneratedTargetException(
×
1142
                    "All generated targets must have the same `Address.spec_path` as their "
1143
                    f"target generator. Expected {generator.address.spec_path}, but got "
1144
                    f"{tgt.address.spec_path} for target generated from {generator.address}: {tgt}"
1145
                    "\n\nConsider using `request.generator.address.create_generated()`."
1146
                )
1147
            if tgt.address.target_name != expected_tgt_name:
1✔
UNCOV
1148
                raise InvalidGeneratedTargetException(
×
1149
                    "All generated targets must have the same `Address.target_name` as their "
1150
                    f"target generator. Expected {generator.address.target_name}, but got "
1151
                    f"{tgt.address.target_name} for target generated from {generator.address}: "
1152
                    f"{tgt}\n\n"
1153
                    "Consider using `request.generator.address.create_generated()`."
1154
                )
1155
            if not tgt.address.is_generated_target:
1✔
1156
                raise InvalidGeneratedTargetException(
×
1157
                    "All generated targets must set `Address.generator_name` or "
1158
                    "`Address.relative_file_path`. Invalid for target generated from "
1159
                    f"{generator.address}: {tgt}\n\n"
1160
                    "Consider using `request.generator.address.create_generated()`."
1161
                )
1162
            mapping[tgt.address] = tgt
1✔
1163
        return super().__new__(cls, mapping)
1✔
1164

1165

1166
@rule(polymorphic=True)
1✔
1167
async def generate_targets(req: GenerateTargetsRequest) -> GeneratedTargets:
1✔
1168
    raise NotImplementedError()
×
1169

1170

1171
class TargetTypesToGenerateTargetsRequests(
1✔
1172
    FrozenDict[type[TargetGenerator], type[GenerateTargetsRequest]]
1173
):
1174
    def is_generator(self, tgt: Target) -> bool:
1✔
1175
        """Does this target type generate other targets?"""
1176
        return isinstance(tgt, TargetGenerator) and bool(self.request_for(type(tgt)))
1✔
1177

1178
    def request_for(self, tgt_cls: type[TargetGenerator]) -> type[GenerateTargetsRequest] | None:
1✔
1179
        """Return the request type for the given Target, or None."""
1180
        if issubclass(tgt_cls, TargetFilesGenerator):
1✔
1181
            return self.get(TargetFilesGenerator)
1✔
1182
        return self.get(tgt_cls)
1✔
1183

1184

1185
def _generate_file_level_targets(
1✔
1186
    generated_target_cls: type[Target],
1187
    generator: Target,
1188
    paths: Sequence[str],
1189
    template_address: Address,
1190
    template: Mapping[str, Any],
1191
    overrides: Mapping[str, Mapping[Address, Mapping[str, Any]]],
1192
    # NB: Should only ever be set to `None` in tests.
1193
    union_membership: UnionMembership | None,
1194
    *,
1195
    add_dependencies_on_all_siblings: bool,
1196
) -> GeneratedTargets:
1197
    """Generate one new target for each path, using the same fields as the generator target except
1198
    for the `sources` field only referring to the path and using a new address.
1199

1200
    Set `add_dependencies_on_all_siblings` to True so that each file-level target depends on all
1201
    other generated targets from the target generator. This is useful if both are true:
1202

1203
        a) file-level targets usually need their siblings to be present to work. Most target types
1204
          (Python, Java, Shell, etc) meet this, except for `files` and `resources` which have no
1205
          concept of "imports"
1206
        b) dependency inference cannot infer dependencies on sibling files.
1207

1208
    Otherwise, set `add_dependencies_on_all_siblings` to `False` so that dependencies are
1209
    finer-grained.
1210

1211
    `overrides` allows changing the fields for particular targets. It expects the full file path
1212
     as the key.
1213
    """
1214

1215
    # Paths will have already been globbed, so they should be escaped. See
1216
    # https://github.com/pantsbuild/pants/issues/15381.
1217
    paths = [glob_stdlib.escape(path) for path in paths]
1✔
1218

1219
    normalized_overrides = dict(overrides or {})
1✔
1220

1221
    all_generated_items: list[tuple[Address, str, dict[str, Any]]] = []
1✔
1222
    for fp in paths:
1✔
1223
        relativized_fp = fast_relpath(fp, template_address.spec_path)
1✔
1224

1225
        generated_overrides = normalized_overrides.pop(fp, None)
1✔
1226
        if generated_overrides is None:
1✔
1227
            # No overrides apply.
1228
            all_generated_items.append(
1✔
1229
                (template_address.create_file(relativized_fp), fp, dict(template))
1230
            )
1231
        else:
1232
            # At least one override applies. Generate a target per set of fields.
UNCOV
1233
            all_generated_items.extend(
×
1234
                (
1235
                    overridden_address.create_file(relativized_fp),
1236
                    fp,
1237
                    {**template, **override_fields},
1238
                )
1239
                for overridden_address, override_fields in generated_overrides.items()
1240
            )
1241

1242
    # TODO: Parametrization in overrides will result in some unusual internal dependencies when
1243
    # `add_dependencies_on_all_siblings`. Similar to inference, `add_dependencies_on_all_siblings`
1244
    # should probably be field value aware.
1245
    all_generated_address_specs = (
1✔
1246
        FrozenOrderedSet(addr.spec for addr, _, _ in all_generated_items)
1247
        if add_dependencies_on_all_siblings
1248
        else FrozenOrderedSet()
1249
    )
1250

1251
    def gen_tgt(address: Address, full_fp: str, generated_target_fields: dict[str, Any]) -> Target:
1✔
1252
        if add_dependencies_on_all_siblings:
1✔
UNCOV
1253
            if union_membership and not generated_target_cls.class_has_field(
×
1254
                Dependencies, union_membership
1255
            ):
1256
                raise AssertionError(
×
1257
                    f"The {type(generator).__name__} target class generates "
1258
                    f"{generated_target_cls.__name__} targets, which do not "
1259
                    f"have a `{Dependencies.alias}` field, and thus cannot "
1260
                    "`add_dependencies_on_all_siblings`."
1261
                )
UNCOV
1262
            original_deps = generated_target_fields.get(Dependencies.alias, ())
×
UNCOV
1263
            generated_target_fields[Dependencies.alias] = tuple(original_deps) + tuple(
×
1264
                all_generated_address_specs - {address.spec}
1265
            )
1266

1267
        generated_target_fields[SingleSourceField.alias] = fast_relpath(full_fp, address.spec_path)
1✔
1268
        return generated_target_cls(
1✔
1269
            generated_target_fields,
1270
            address,
1271
            union_membership=union_membership,
1272
            residence_dir=os.path.dirname(full_fp),
1273
        )
1274

1275
    result = tuple(
1✔
1276
        gen_tgt(address, full_fp, fields) for address, full_fp, fields in all_generated_items
1277
    )
1278

1279
    if normalized_overrides:
1✔
UNCOV
1280
        unused_relative_paths = sorted(
×
1281
            fast_relpath(fp, template_address.spec_path) for fp in normalized_overrides
1282
        )
UNCOV
1283
        all_valid_relative_paths = sorted(
×
1284
            cast(str, tgt.address.relative_file_path or tgt.address.generated_name)
1285
            for tgt in result
1286
        )
UNCOV
1287
        raise InvalidFieldException(
×
1288
            f"Unused file paths in the `overrides` field for {template_address}: "
1289
            f"{sorted(unused_relative_paths)}"
1290
            f"\n\nDid you mean one of these valid paths?\n\n"
1291
            f"{all_valid_relative_paths}"
1292
        )
1293

1294
    return GeneratedTargets(generator, result)
1✔
1295

1296

1297
# -----------------------------------------------------------------------------------------------
1298
# FieldSet
1299
# -----------------------------------------------------------------------------------------------
1300
def _get_field_set_fields_from_target(
1✔
1301
    field_set: type[FieldSet], target: Target
1302
) -> dict[str, Field]:
1303
    return {
1✔
1304
        dataclass_field_name: (
1305
            target[field_cls] if field_cls in field_set.required_fields else target.get(field_cls)
1306
        )
1307
        for dataclass_field_name, field_cls in field_set.fields.items()
1308
    }
1309

1310

1311
_FS = TypeVar("_FS", bound="FieldSet")
1✔
1312

1313

1314
@dataclass(frozen=True)
1✔
1315
class FieldSet(EngineAwareParameter, metaclass=ABCMeta):
1✔
1316
    """An ad hoc set of fields from a target which are used by rules.
1317

1318
    Subclasses should declare all the fields they consume as dataclass attributes. They should also
1319
    indicate which of these are required, rather than optional, through the class property
1320
    `required_fields`. When a field is optional, the default constructor for the field will be used
1321
    for any targets that do not have that field registered.
1322

1323
    Subclasses must set `@dataclass(frozen=True)` for their declared fields to be recognized.
1324

1325
    You can optionally implement the classmethod `opt_out` so that targets have a
1326
    mechanism to not match with the FieldSet even if they have the `required_fields` registered.
1327

1328
    For example:
1329

1330
        @dataclass(frozen=True)
1331
        class FortranTestFieldSet(FieldSet):
1332
            required_fields = (FortranSources,)
1333

1334
            sources: FortranSources
1335
            fortran_version: FortranVersion
1336

1337
            @classmethod
1338
            def opt_out(cls, tgt: Target) -> bool:
1339
                return tgt.get(MaybeSkipFortranTestsField).value
1340

1341
    This field set may then be created from a `Target` through the `is_applicable()` and `create()`
1342
    class methods:
1343

1344
        field_sets = [
1345
            FortranTestFieldSet.create(tgt) for tgt in targets
1346
            if FortranTestFieldSet.is_applicable(tgt)
1347
        ]
1348

1349
    FieldSets are consumed like any normal dataclass:
1350

1351
        print(field_set.address)
1352
        print(field_set.sources)
1353
    """
1354

1355
    required_fields: ClassVar[tuple[type[Field], ...]]
1✔
1356

1357
    address: Address
1✔
1358

1359
    @classmethod
1✔
1360
    def opt_out(cls, tgt: Target) -> bool:
1✔
1361
        """If `True`, the target will not match with the field set, even if it has the FieldSet's
1362
        `required_fields`.
1363

1364
        Note: this method is not intended to categorically opt out a target type from a
1365
        FieldSet, i.e. to always opt out based solely on the target type. While it is possible to
1366
        do, some error messages will incorrectly suggest that that target is compatible with the
1367
        FieldSet. Instead, if you need this feature, please ask us to implement it. See
1368
        https://github.com/pantsbuild/pants/pull/12002 for discussion.
1369
        """
1370
        return False
1✔
1371

1372
    @final
1✔
1373
    @classmethod
1✔
1374
    def is_applicable(cls, tgt: Target) -> bool:
1✔
1375
        return tgt.has_fields(cls.required_fields) and not cls.opt_out(tgt)
1✔
1376

1377
    @final
1✔
1378
    @classmethod
1✔
1379
    def applicable_target_types(
1✔
1380
        cls, target_types: Iterable[type[Target]], union_membership: UnionMembership
1381
    ) -> tuple[type[Target], ...]:
UNCOV
1382
        return tuple(
×
1383
            tgt_type
1384
            for tgt_type in target_types
1385
            if tgt_type.class_has_fields(cls.required_fields, union_membership)
1386
        )
1387

1388
    @final
1✔
1389
    @classmethod
1✔
1390
    def create(cls: type[_FS], tgt: Target) -> _FS:
1✔
1391
        return cls(address=tgt.address, **_get_field_set_fields_from_target(cls, tgt))
1✔
1392

1393
    @final
1✔
1394
    @memoized_classproperty
1✔
1395
    def fields(cls) -> FrozenDict[str, type[Field]]:
1✔
1396
        return FrozenDict(
1✔
1397
            (
1398
                (name, field_type)
1399
                for name, field_type in get_type_hints(cls).items()
1400
                if isinstance(field_type, type) and issubclass(field_type, Field)
1401
            )
1402
        )
1403

1404
    def debug_hint(self) -> str:
1✔
1405
        return self.address.spec
1✔
1406

1407
    def metadata(self) -> dict[str, Any]:
1✔
1408
        return {"address": self.address.spec}
1✔
1409

1410
    def __repr__(self) -> str:
1✔
1411
        # We use a short repr() because this often shows up in stack traces. We don't need any of
1412
        # the field information because we can ask a user to send us their BUILD file.
1413
        return f"{self.__class__.__name__}(address={self.address})"
×
1414

1415

1416
@dataclass(frozen=True)
1✔
1417
class TargetRootsToFieldSets(Generic[_FS]):
1✔
1418
    mapping: FrozenDict[Target, tuple[_FS, ...]]
1✔
1419

1420
    def __init__(self, mapping: Mapping[Target, Iterable[_FS]]) -> None:
1✔
UNCOV
1421
        object.__setattr__(
×
1422
            self,
1423
            "mapping",
1424
            FrozenDict({tgt: tuple(field_sets) for tgt, field_sets in mapping.items()}),
1425
        )
1426

1427
    @memoized_property
1✔
1428
    def field_sets(self) -> tuple[_FS, ...]:
1✔
UNCOV
1429
        return tuple(
×
1430
            itertools.chain.from_iterable(
1431
                field_sets_per_target for field_sets_per_target in self.mapping.values()
1432
            )
1433
        )
1434

1435
    @memoized_property
1✔
1436
    def targets(self) -> tuple[Target, ...]:
1✔
UNCOV
1437
        return tuple(self.mapping.keys())
×
1438

1439

1440
class NoApplicableTargetsBehavior(Enum):
1✔
1441
    ignore = "ignore"
1✔
1442
    warn = "warn"
1✔
1443
    error = "error"
1✔
1444

1445

1446
def parse_shard_spec(shard_spec: str, origin: str = "") -> tuple[int, int]:
1✔
UNCOV
1447
    def invalid():
×
UNCOV
1448
        origin_str = f" from {origin}" if origin else ""
×
UNCOV
1449
        return ValueError(
×
1450
            f"Invalid shard specification {shard_spec}{origin_str}. Use a string of the form "
1451
            '"k/N" where k and N are integers, and 0 <= k < N .'
1452
        )
1453

UNCOV
1454
    if not shard_spec:
×
UNCOV
1455
        return 0, -1
×
UNCOV
1456
    shard_str, _, num_shards_str = shard_spec.partition("/")
×
UNCOV
1457
    try:
×
UNCOV
1458
        shard, num_shards = int(shard_str), int(num_shards_str)
×
UNCOV
1459
    except ValueError:
×
UNCOV
1460
        raise invalid()
×
UNCOV
1461
    if shard < 0 or shard >= num_shards:
×
UNCOV
1462
        raise invalid()
×
UNCOV
1463
    return shard, num_shards
×
1464

1465

1466
def get_shard(key: str, num_shards: int) -> int:
1✔
1467
    # Note: hash() is not guaranteed to be stable across processes, and adler32 is not
1468
    # well-distributed for small strings, so we use crc32. It's faster to compute than
1469
    # a cryptographic hash, which would be overkill.
UNCOV
1470
    return zlib.crc32(key.encode()) % num_shards
×
1471

1472

1473
@dataclass(frozen=True)
1✔
1474
class TargetRootsToFieldSetsRequest(Generic[_FS]):
1✔
1475
    field_set_superclass: type[_FS]
1✔
1476
    goal_description: str
1✔
1477
    no_applicable_targets_behavior: NoApplicableTargetsBehavior
1✔
1478
    shard: int
1✔
1479
    num_shards: int
1✔
1480

1481
    def __init__(
1✔
1482
        self,
1483
        field_set_superclass: type[_FS],
1484
        *,
1485
        goal_description: str,
1486
        no_applicable_targets_behavior: NoApplicableTargetsBehavior,
1487
        shard: int = 0,
1488
        num_shards: int = -1,
1489
    ) -> None:
UNCOV
1490
        object.__setattr__(self, "field_set_superclass", field_set_superclass)
×
UNCOV
1491
        object.__setattr__(self, "goal_description", goal_description)
×
UNCOV
1492
        object.__setattr__(self, "no_applicable_targets_behavior", no_applicable_targets_behavior)
×
UNCOV
1493
        object.__setattr__(self, "shard", shard)
×
UNCOV
1494
        object.__setattr__(self, "num_shards", num_shards)
×
1495

1496
    def is_in_shard(self, key: str) -> bool:
1✔
1497
        return get_shard(key, self.num_shards) == self.shard
×
1498

1499

1500
@dataclass(frozen=True)
1✔
1501
class FieldSetsPerTarget(Generic[_FS]):
1✔
1502
    # One tuple of FieldSet instances per input target.
1503
    collection: tuple[tuple[_FS, ...], ...]
1✔
1504

1505
    def __init__(self, collection: Iterable[Iterable[_FS]]):
1✔
1506
        object.__setattr__(self, "collection", tuple(tuple(iterable) for iterable in collection))
1✔
1507

1508
    @memoized_property
1✔
1509
    def field_sets(self) -> tuple[_FS, ...]:
1✔
1510
        return tuple(itertools.chain.from_iterable(self.collection))
1✔
1511

1512

1513
@dataclass(frozen=True)
1✔
1514
class FieldSetsPerTargetRequest(Generic[_FS]):
1✔
1515
    field_set_superclass: type[_FS]
1✔
1516
    targets: tuple[Target, ...]
1✔
1517

1518
    def __init__(self, field_set_superclass: type[_FS], targets: Iterable[Target]):
1✔
1519
        object.__setattr__(self, "field_set_superclass", field_set_superclass)
1✔
1520
        object.__setattr__(self, "targets", tuple(targets))
1✔
1521

1522

1523
# -----------------------------------------------------------------------------------------------
1524
# Exception messages
1525
# -----------------------------------------------------------------------------------------------
1526

1527

1528
class InvalidTargetException(Exception):
1✔
1529
    """Use when there's an issue with the target, e.g. mutually exclusive fields set.
1530

1531
    Suggested template:
1532

1533
         f"The `{alias!r}` target {address} ..."
1534
    """
1535

1536
    def __init__(self, message: Any, *, description_of_origin: str | None = None) -> None:
1✔
1537
        self.description_of_origin = description_of_origin
1✔
1538
        super().__init__(message)
1✔
1539

1540
    def __str__(self) -> str:
1✔
1541
        if not self.description_of_origin:
1✔
UNCOV
1542
            return super().__str__()
×
1543
        return f"{self.description_of_origin}: {super().__str__()}"
1✔
1544

1545
    def __repr__(self) -> str:
1✔
UNCOV
1546
        if not self.description_of_origin:
×
UNCOV
1547
            return super().__repr__()
×
1548
        return f"{self.description_of_origin}: {super().__repr__()}"
×
1549

1550

1551
class InvalidGeneratedTargetException(InvalidTargetException):
1✔
1552
    pass
1✔
1553

1554

1555
class InvalidFieldException(Exception):
1✔
1556
    """Use when there's an issue with a particular field.
1557

1558
    Suggested template:
1559

1560
         f"The {alias!r} field in target {address} must ..., but ..."
1561
    """
1562

1563
    def __init__(self, message: Any, *, description_of_origin: str | None = None) -> None:
1✔
1564
        self.description_of_origin = description_of_origin
1✔
1565
        super().__init__(message)
1✔
1566

1567
    def __str__(self) -> str:
1✔
1568
        if not self.description_of_origin:
1✔
1569
            return super().__str__()
1✔
1570
        return f"{self.description_of_origin}: {super().__str__()}"
×
1571

1572
    def __repr__(self) -> str:
1✔
1573
        if not self.description_of_origin:
×
1574
            return super().__repr__()
×
1575
        return f"{self.description_of_origin}: {super().__repr__()}"
×
1576

1577

1578
class InvalidFieldTypeException(InvalidFieldException):
1✔
1579
    """This is used to ensure that the field's value conforms with the expected type for the field,
1580
    e.g. `a boolean` or `a string` or `an iterable of strings and integers`."""
1581

1582
    def __init__(
1✔
1583
        self,
1584
        address: Address,
1585
        field_alias: str,
1586
        raw_value: Any | None,
1587
        *,
1588
        expected_type: str,
1589
        description_of_origin: str | None = None,
1590
    ) -> None:
1591
        raw_type = f"with type `{type(raw_value).__name__}`"
1✔
1592
        super().__init__(
1✔
1593
            f"The {repr(field_alias)} field in target {address} must be {expected_type}, but was "
1594
            f"`{repr(raw_value)}` {raw_type}.",
1595
            description_of_origin=description_of_origin,
1596
        )
1597

1598

1599
class InvalidFieldMemberTypeException(InvalidFieldException):
1✔
1600
    # based on InvalidFieldTypeException
1601
    def __init__(
1✔
1602
        self,
1603
        address: Address,
1604
        field_alias: str,
1605
        raw_value: Any | None,
1606
        *,
1607
        expected_type: str,
1608
        at_index: int,
1609
        wrong_element: Any,
1610
        description_of_origin: str | None = None,
1611
    ) -> None:
UNCOV
1612
        super().__init__(
×
1613
            softwrap(
1614
                f"""
1615
                The {repr(field_alias)} field in target {address} must be an iterable with
1616
                elements that have type {expected_type}. Encountered the element `{wrong_element}`
1617
                of type {type(wrong_element)} instead of {expected_type} at index {at_index}:
1618
                `{repr(raw_value)}`
1619
                """
1620
            ),
1621
            description_of_origin=description_of_origin,
1622
        )
1623

1624

1625
class RequiredFieldMissingException(InvalidFieldException):
1✔
1626
    def __init__(
1✔
1627
        self, address: Address, field_alias: str, *, description_of_origin: str | None = None
1628
    ) -> None:
1629
        super().__init__(
×
1630
            f"The {repr(field_alias)} field in target {address} must be defined.",
1631
            description_of_origin=description_of_origin,
1632
        )
1633

1634

1635
class InvalidFieldChoiceException(InvalidFieldException):
1✔
1636
    def __init__(
1✔
1637
        self,
1638
        address: Address,
1639
        field_alias: str,
1640
        raw_value: Any | None,
1641
        *,
1642
        valid_choices: Iterable[Any],
1643
        description_of_origin: str | None = None,
1644
    ) -> None:
UNCOV
1645
        super().__init__(
×
1646
            f"Values for the {repr(field_alias)} field in target {address} must be one of "
1647
            f"{sorted(valid_choices)}, but {repr(raw_value)} was provided.",
1648
            description_of_origin=description_of_origin,
1649
        )
1650

1651

1652
class UnrecognizedTargetTypeException(InvalidTargetException):
1✔
1653
    def __init__(
1✔
1654
        self,
1655
        target_type: str,
1656
        registered_target_types: RegisteredTargetTypes,
1657
        address: Address | None = None,
1658
        description_of_origin: str | None = None,
1659
    ) -> None:
UNCOV
1660
        for_address = f" for address {address}" if address else ""
×
UNCOV
1661
        super().__init__(
×
1662
            softwrap(
1663
                f"""
1664
                Target type {target_type!r} is not registered{for_address}.
1665

1666
                All valid target types: {sorted(registered_target_types.aliases)}
1667

1668
                (If {target_type!r} is a custom target type, refer to
1669
                {doc_url("docs/writing-plugins/the-target-api/concepts")} for getting it registered with Pants.)
1670

1671
                """
1672
            ),
1673
            description_of_origin=description_of_origin,
1674
        )
1675

1676

1677
# -----------------------------------------------------------------------------------------------
1678
# Field templates
1679
# -----------------------------------------------------------------------------------------------
1680

1681
T = TypeVar("T")
1✔
1682

1683

1684
class BoolField(ScalarField[bool]):
1✔
1685
    """A field whose value is a boolean.
1686

1687
    Subclasses must either set `default: bool` or `required = True` so that the value is always
1688
    defined.
1689
    """
1690

1691
    expected_type = bool
1✔
1692
    expected_type_description = "a boolean"
1✔
1693
    value: bool
1✔
1694
    default: ClassVar[bool]
1✔
1695

1696

1697
class TriBoolField(ScalarField[bool]):
1✔
1698
    """A field whose value is a boolean or None, which is meant to represent a tri-state."""
1699

1700
    expected_type = bool
1✔
1701
    expected_type_description = "a boolean or None"
1✔
1702

1703

1704
class ValidNumbers(Enum):
1✔
1705
    """What range of numbers are allowed for IntField and FloatField."""
1706

1707
    positive_only = enum.auto()
1✔
1708
    positive_and_zero = enum.auto()
1✔
1709
    all = enum.auto()
1✔
1710

1711
    def validate(self, num: float | int | None, alias: str, address: Address) -> None:
1✔
1712
        if num is None or self == self.all:
1✔
1713
            return
1✔
1714
        if self == self.positive_and_zero:
1✔
1715
            if num < 0:
1✔
UNCOV
1716
                raise InvalidFieldException(
×
1717
                    f"The {repr(alias)} field in target {address} must be greater than or equal to "
1718
                    f"zero, but was set to `{num}`."
1719
                )
1720
            return
1✔
1721
        if num <= 0:
1✔
UNCOV
1722
            raise InvalidFieldException(
×
1723
                f"The {repr(alias)} field in target {address} must be greater than zero, but was "
1724
                f"set to `{num}`."
1725
            )
1726

1727

1728
class IntField(ScalarField[int]):
1✔
1729
    expected_type = int
1✔
1730
    expected_type_description = "an integer"
1✔
1731
    valid_numbers: ClassVar[ValidNumbers] = ValidNumbers.all
1✔
1732

1733
    @classmethod
1✔
1734
    def compute_value(cls, raw_value: int | None, address: Address) -> int | None:
1✔
1735
        value_or_default = super().compute_value(raw_value, address)
1✔
1736
        cls.valid_numbers.validate(value_or_default, cls.alias, address)
1✔
1737
        return value_or_default
1✔
1738

1739

1740
class FloatField(ScalarField[float]):
1✔
1741
    expected_type = float
1✔
1742
    expected_type_description = "a float"
1✔
1743
    valid_numbers: ClassVar[ValidNumbers] = ValidNumbers.all
1✔
1744

1745
    @classmethod
1✔
1746
    def compute_value(cls, raw_value: float | None, address: Address) -> float | None:
1✔
UNCOV
1747
        value_or_default = super().compute_value(raw_value, address)
×
UNCOV
1748
        cls.valid_numbers.validate(value_or_default, cls.alias, address)
×
UNCOV
1749
        return value_or_default
×
1750

1751

1752
class TupleSequenceField(Generic[T], Field):
1✔
1753
    # this cannot be a SequenceField as compute_value's use of ensure_list
1754
    # does not work with expected_element_type=tuple when the value itself
1755
    # is already a tuple.
1756
    expected_element_type: ClassVar[type]
1✔
1757
    expected_element_count: ClassVar[int]  # -1 for unlimited
1✔
1758
    expected_type_description: ClassVar[str]
1✔
1759
    expected_element_type_description: ClassVar[str]
1✔
1760

1761
    value: tuple[tuple[T, ...], ...] | None
1✔
1762
    default: ClassVar[tuple[tuple[T, ...], ...] | None] = None
1✔
1763

1764
    @classmethod
1✔
1765
    def compute_value(
1✔
1766
        cls, raw_value: Iterable[Iterable[T]] | None, address: Address
1767
    ) -> tuple[tuple[T, ...], ...] | None:
1768
        value_or_default = super().compute_value(raw_value, address)
1✔
1769
        if value_or_default is None:
1✔
UNCOV
1770
            return value_or_default
×
1771
        if isinstance(value_or_default, str) or not isinstance(
1✔
1772
            value_or_default, collections.abc.Iterable
1773
        ):
UNCOV
1774
            raise InvalidFieldTypeException(
×
1775
                address,
1776
                cls.alias,
1777
                raw_value,
1778
                expected_type=cls.expected_type_description,
1779
            )
1780

1781
        def invalid_member_exception(
1✔
1782
            at_index: int, wrong_element: Any
1783
        ) -> InvalidFieldMemberTypeException:
UNCOV
1784
            return InvalidFieldMemberTypeException(
×
1785
                address,
1786
                cls.alias,
1787
                raw_value,
1788
                expected_type=cls.expected_element_type_description,
1789
                wrong_element=wrong_element,
1790
                at_index=at_index,
1791
            )
1792

1793
        validated: list[tuple[T, ...]] = []
1✔
1794
        for i, x in enumerate(value_or_default):
1✔
1795
            if isinstance(x, str) or not isinstance(x, collections.abc.Iterable):
1✔
UNCOV
1796
                raise invalid_member_exception(i, x)
×
1797
            element = tuple(x)
1✔
1798
            if cls.expected_element_count >= 0 and cls.expected_element_count != len(element):
1✔
1799
                raise invalid_member_exception(i, x)
×
1800
            for s in element:
1✔
1801
                if not isinstance(s, cls.expected_element_type):
1✔
UNCOV
1802
                    raise invalid_member_exception(i, x)
×
1803
            validated.append(cast(tuple[T, ...], element))
1✔
1804

1805
        return tuple(validated)
1✔
1806

1807

1808
class DictStringToStringField(Field):
1✔
1809
    value: FrozenDict[str, str] | None
1✔
1810
    default: ClassVar[FrozenDict[str, str] | None] = None
1✔
1811

1812
    @classmethod
1✔
1813
    def compute_value(
1✔
1814
        cls, raw_value: dict[str, str] | None, address: Address
1815
    ) -> FrozenDict[str, str] | None:
1816
        value_or_default = super().compute_value(raw_value, address)
1✔
1817
        if value_or_default is None:
1✔
1818
            return None
1✔
1819
        invalid_type_exception = InvalidFieldTypeException(
1✔
1820
            address, cls.alias, raw_value, expected_type="a dictionary of string -> string"
1821
        )
1822
        if not isinstance(value_or_default, collections.abc.Mapping):
1✔
UNCOV
1823
            raise invalid_type_exception
×
1824
        if not all(isinstance(k, str) and isinstance(v, str) for k, v in value_or_default.items()):
1✔
UNCOV
1825
            raise invalid_type_exception
×
1826
        return FrozenDict(value_or_default)
1✔
1827

1828

1829
class ListOfDictStringToStringField(Field):
1✔
1830
    value: tuple[FrozenDict[str, str]] | None
1✔
1831
    default: ClassVar[list[FrozenDict[str, str]] | None] = None
1✔
1832

1833
    @classmethod
1✔
1834
    def compute_value(
1✔
1835
        cls, raw_value: list[dict[str, str]] | None, address: Address
1836
    ) -> tuple[FrozenDict[str, str], ...] | None:
1837
        value_or_default = super().compute_value(raw_value, address)
1✔
1838
        if value_or_default is None:
1✔
1839
            return None
1✔
UNCOV
1840
        invalid_type_exception = InvalidFieldTypeException(
×
1841
            address,
1842
            cls.alias,
1843
            raw_value,
1844
            expected_type="a list of dictionaries (or a single dictionary) of string -> string",
1845
        )
1846

1847
        # Also support passing in a single dictionary by wrapping it
UNCOV
1848
        if not isinstance(value_or_default, (list, tuple)):
×
UNCOV
1849
            value_or_default = [value_or_default]
×
1850

UNCOV
1851
        result_lst: list[FrozenDict[str, str]] = []
×
UNCOV
1852
        for item in value_or_default:
×
UNCOV
1853
            if not isinstance(item, collections.abc.Mapping):
×
UNCOV
1854
                raise invalid_type_exception
×
UNCOV
1855
            if not all(isinstance(k, str) and isinstance(v, str) for k, v in item.items()):
×
UNCOV
1856
                raise invalid_type_exception
×
UNCOV
1857
            result_lst.append(FrozenDict(item))
×
1858

UNCOV
1859
        return tuple(result_lst)
×
1860

1861

1862
class NestedDictStringToStringField(Field):
1✔
1863
    value: FrozenDict[str, FrozenDict[str, str]] | None
1✔
1864
    default: ClassVar[FrozenDict[str, FrozenDict[str, str]] | None] = None
1✔
1865

1866
    @classmethod
1✔
1867
    def compute_value(
1✔
1868
        cls, raw_value: dict[str, dict[str, str]] | None, address: Address
1869
    ) -> FrozenDict[str, FrozenDict[str, str]] | None:
UNCOV
1870
        value_or_default = super().compute_value(raw_value, address)
×
UNCOV
1871
        if value_or_default is None:
×
UNCOV
1872
            return None
×
UNCOV
1873
        invalid_type_exception = InvalidFieldTypeException(
×
1874
            address,
1875
            cls.alias,
1876
            raw_value,
1877
            expected_type="dict[str, dict[str, str]]",
1878
        )
UNCOV
1879
        if not isinstance(value_or_default, collections.abc.Mapping):
×
UNCOV
1880
            raise invalid_type_exception
×
UNCOV
1881
        for key, nested_value in value_or_default.items():
×
UNCOV
1882
            if not isinstance(key, str) or not isinstance(nested_value, collections.abc.Mapping):
×
UNCOV
1883
                raise invalid_type_exception
×
UNCOV
1884
            if not all(isinstance(k, str) and isinstance(v, str) for k, v in nested_value.items()):
×
1885
                raise invalid_type_exception
×
UNCOV
1886
        return FrozenDict(
×
1887
            {key: FrozenDict(nested_value) for key, nested_value in value_or_default.items()}
1888
        )
1889

1890

1891
class DictStringToStringSequenceField(Field):
1✔
1892
    value: FrozenDict[str, tuple[str, ...]] | None
1✔
1893
    default: ClassVar[FrozenDict[str, tuple[str, ...]] | None] = None
1✔
1894

1895
    @classmethod
1✔
1896
    def compute_value(
1✔
1897
        cls, raw_value: dict[str, Iterable[str]] | None, address: Address
1898
    ) -> FrozenDict[str, tuple[str, ...]] | None:
1899
        value_or_default = super().compute_value(raw_value, address)
1✔
1900
        if value_or_default is None:
1✔
1901
            return None
1✔
1902
        invalid_type_exception = InvalidFieldTypeException(
1✔
1903
            address,
1904
            cls.alias,
1905
            raw_value,
1906
            expected_type="a dictionary of string -> an iterable of strings",
1907
        )
1908
        if not isinstance(value_or_default, collections.abc.Mapping):
1✔
UNCOV
1909
            raise invalid_type_exception
×
1910
        result = {}
1✔
1911
        for k, v in value_or_default.items():
1✔
1912
            if not isinstance(k, str):
1✔
UNCOV
1913
                raise invalid_type_exception
×
1914
            try:
1✔
1915
                result[k] = tuple(ensure_str_list(v))
1✔
UNCOV
1916
            except ValueError:
×
UNCOV
1917
                raise invalid_type_exception
×
1918
        return FrozenDict(result)
1✔
1919

1920

1921
# -----------------------------------------------------------------------------------------------
1922
# Sources and codegen
1923
# -----------------------------------------------------------------------------------------------
1924

1925

1926
class SourcesField(AsyncFieldMixin, Field):
1✔
1927
    """A field for the sources that a target owns.
1928

1929
    When defining a new sources field, you should subclass `MultipleSourcesField` or
1930
    `SingleSourceField`, which set up the field's `alias` and data type / parsing. However, you
1931
    should use `tgt.get(SourcesField)` when you need to operate on all sources types, such as
1932
    with `HydrateSourcesRequest`, so that both subclasses work.
1933

1934
    Subclasses may set the following class properties:
1935

1936
    - `expected_file_extensions` -- A tuple of strings containing the expected file extensions for
1937
        source files. The default is no expected file extensions.
1938
    - `expected_num_files` -- An integer or range stating the expected total number of source
1939
        files. The default is no limit on the number of source files.
1940
    - `uses_source_roots` -- Whether the concept of "source root" pertains to the source files
1941
        referenced by this field.
1942
    - `default` -- A default value for this field.
1943
    - `default_glob_match_error_behavior` -- Advanced option, should very rarely be used. Override
1944
        glob match error behavior when using the default value. If setting this to
1945
        `GlobMatchErrorBehavior.ignore`, make sure you have other validation in place in case the
1946
        default glob doesn't match any files, if required, to alert the user appropriately.
1947
    """
1948

1949
    expected_file_extensions: ClassVar[tuple[str, ...] | None] = None
1✔
1950
    expected_num_files: ClassVar[int | range | None] = None
1✔
1951
    uses_source_roots: ClassVar[bool] = True
1✔
1952

1953
    default: ClassVar[ImmutableValue] = None
1✔
1954
    default_glob_match_error_behavior: ClassVar[GlobMatchErrorBehavior | None] = None
1✔
1955

1956
    @property
1✔
1957
    def globs(self) -> tuple[str, ...]:
1✔
1958
        """The raw globs, relative to the BUILD file."""
1959

1960
        # NB: We give a default implementation because it's common to use
1961
        # `tgt.get(SourcesField)`, and that must not error. But, subclasses need to
1962
        # implement this for the field to be useful (they should subclass `MultipleSourcesField`
1963
        # and `SingleSourceField`).
1964
        return ()
1✔
1965

1966
    def validate_resolved_files(self, files: Sequence[str]) -> None:
1✔
1967
        """Perform any additional validation on the resulting source files, e.g. ensuring that
1968
        certain banned files are not used.
1969

1970
        To enforce that the resulting files end in certain extensions, such as `.py` or `.java`, set
1971
        the class property `expected_file_extensions`.
1972

1973
        To enforce that there are only a certain number of resulting files, such as binary targets
1974
        checking for only 0-1 sources, set the class property `expected_num_files`.
1975
        """
1976
        if self.expected_file_extensions is not None:
1✔
1977
            bad_files = [
1✔
1978
                fp for fp in files if PurePath(fp).suffix not in self.expected_file_extensions
1979
            ]
1980
            if bad_files:
1✔
UNCOV
1981
                expected = (
×
1982
                    f"one of {sorted(self.expected_file_extensions)}"
1983
                    if len(self.expected_file_extensions) > 1
1984
                    else repr(self.expected_file_extensions[0])
1985
                )
UNCOV
1986
                raise InvalidFieldException(
×
1987
                    f"The {repr(self.alias)} field in target {self.address} can only contain "
1988
                    f"files that end in {expected}, but it had these files: {sorted(bad_files)}."
1989
                    "\n\nMaybe create a `resource`/`resources` or `file`/`files` target and "
1990
                    "include it in the `dependencies` field?"
1991
                )
1992
        if self.expected_num_files is not None:
1✔
1993
            num_files = len(files)
1✔
1994
            is_bad_num_files = (
1✔
1995
                num_files not in self.expected_num_files
1996
                if isinstance(self.expected_num_files, range)
1997
                else num_files != self.expected_num_files
1998
            )
1999
            if is_bad_num_files:
1✔
UNCOV
2000
                if isinstance(self.expected_num_files, range):
×
UNCOV
2001
                    if len(self.expected_num_files) == 2:
×
UNCOV
2002
                        expected_str = (
×
2003
                            " or ".join(str(n) for n in self.expected_num_files) + " files"
2004
                        )
2005
                    else:
2006
                        expected_str = f"a number of files in the range `{self.expected_num_files}`"
×
2007
                else:
UNCOV
2008
                    expected_str = pluralize(self.expected_num_files, "file")
×
UNCOV
2009
                raise InvalidFieldException(
×
2010
                    f"The {repr(self.alias)} field in target {self.address} must have "
2011
                    f"{expected_str}, but it had {pluralize(num_files, 'file')}."
2012
                )
2013

2014
    @staticmethod
1✔
2015
    def prefix_glob_with_dirpath(dirpath: str, glob: str) -> str:
1✔
2016
        if glob.startswith("!"):
1✔
2017
            return f"!{os.path.join(dirpath, glob[1:])}"
1✔
2018
        return os.path.join(dirpath, glob)
1✔
2019

2020
    @final
1✔
2021
    def _prefix_glob_with_address(self, glob: str) -> str:
1✔
2022
        return self.prefix_glob_with_dirpath(self.address.spec_path, glob)
1✔
2023

2024
    @final
1✔
2025
    @classmethod
1✔
2026
    def can_generate(
1✔
2027
        cls, output_type: type[SourcesField], union_membership: UnionMembership
2028
    ) -> bool:
2029
        """Can this field be used to generate the output_type?
2030

2031
        Generally, this method does not need to be used. Most call sites can simply use the below,
2032
        and the engine will generate the sources if possible or will return an instance of
2033
        HydratedSources with an empty snapshot if not possible:
2034

2035
            await hydrate_sources(
2036
                HydrateSourcesRequest(
2037
                    sources_field,
2038
                    for_sources_types=[FortranSources],
2039
                    enable_codegen=True,
2040
                ),
2041
                **implicitly(),
2042
            )
2043

2044
        This method is useful when you need to filter targets before hydrating them, such as how
2045
        you may filter targets via `tgt.has_field(MyField)`.
2046
        """
2047
        generate_request_types = union_membership.get(GenerateSourcesRequest)
1✔
2048
        return any(
1✔
2049
            issubclass(cls, generate_request_type.input)
2050
            and issubclass(generate_request_type.output, output_type)
2051
            for generate_request_type in generate_request_types
2052
        )
2053

2054
    @final
1✔
2055
    def path_globs(self, unmatched_build_file_globs: UnmatchedBuildFileGlobs) -> PathGlobs:
1✔
2056
        if not self.globs:
1✔
2057
            return PathGlobs([])
1✔
2058

2059
        # SingleSourceField has str as default type.
2060
        default_globs = (
1✔
2061
            [self.default] if self.default and isinstance(self.default, str) else self.default
2062
        )
2063

2064
        using_default_globs = default_globs and (set(self.globs) == set(default_globs)) or False
1✔
2065

2066
        # Use fields default error behavior if defined, if we use default globs else the provided
2067
        # error behavior.
2068
        error_behavior = (
1✔
2069
            unmatched_build_file_globs.error_behavior
2070
            if not using_default_globs or self.default_glob_match_error_behavior is None
2071
            else self.default_glob_match_error_behavior
2072
        )
2073

2074
        return PathGlobs(
1✔
2075
            (self._prefix_glob_with_address(glob) for glob in self.globs),
2076
            conjunction=GlobExpansionConjunction.any_match,
2077
            glob_match_error_behavior=error_behavior,
2078
            description_of_origin=(
2079
                f"{self.address}'s `{self.alias}` field"
2080
                if error_behavior != GlobMatchErrorBehavior.ignore
2081
                else None
2082
            ),
2083
        )
2084

2085
    @memoized_property
1✔
2086
    def filespec(self) -> Filespec:
1✔
2087
        """The original globs, returned in the Filespec dict format.
2088

2089
        The globs will be relativized to the build root.
2090
        """
2091
        includes = []
1✔
2092
        excludes = []
1✔
2093
        for glob in self.globs:
1✔
2094
            if glob.startswith("!"):
1✔
2095
                excludes.append(os.path.join(self.address.spec_path, glob[1:]))
1✔
2096
            else:
2097
                includes.append(os.path.join(self.address.spec_path, glob))
1✔
2098
        result: Filespec = {"includes": includes}
1✔
2099
        if excludes:
1✔
2100
            result["excludes"] = excludes
1✔
2101
        return result
1✔
2102

2103
    @memoized_property
1✔
2104
    def filespec_matcher(self) -> FilespecMatcher:
1✔
2105
        # Note: memoized because parsing the globs is expensive:
2106
        # https://github.com/pantsbuild/pants/issues/16122
2107
        return FilespecMatcher(self.filespec["includes"], self.filespec.get("excludes", []))
1✔
2108

2109

2110
class MultipleSourcesField(SourcesField, StringSequenceField):
1✔
2111
    """The `sources: list[str]` field.
2112

2113
    See the docstring for `SourcesField` for some class properties you can set, such as
2114
    `expected_file_extensions`.
2115

2116
    When you need to get the sources for all targets, use `tgt.get(SourcesField)` rather than
2117
    `tgt.get(MultipleSourcesField)`.
2118
    """
2119

2120
    alias = "sources"
1✔
2121

2122
    ban_subdirectories: ClassVar[bool] = False
1✔
2123

2124
    @property
1✔
2125
    def globs(self) -> tuple[str, ...]:
1✔
2126
        return self.value or ()
1✔
2127

2128
    @classmethod
1✔
2129
    def compute_value(
1✔
2130
        cls, raw_value: Iterable[str] | None, address: Address
2131
    ) -> tuple[str, ...] | None:
2132
        value = super().compute_value(raw_value, address)
1✔
2133
        invalid_globs = [glob for glob in (value or ()) if glob.startswith("../") or "/../" in glob]
1✔
2134
        if invalid_globs:
1✔
UNCOV
2135
            raise InvalidFieldException(
×
2136
                softwrap(
2137
                    f"""
2138
                    The {repr(cls.alias)} field in target {address} must not have globs with the
2139
                    pattern `../` because targets can only have sources in the current directory
2140
                    or subdirectories. It was set to: {sorted(value or ())}
2141
                    """
2142
                )
2143
            )
2144
        if cls.ban_subdirectories:
1✔
2145
            invalid_globs = [glob for glob in (value or ()) if "**" in glob or os.path.sep in glob]
1✔
2146
            if invalid_globs:
1✔
UNCOV
2147
                raise InvalidFieldException(
×
2148
                    softwrap(
2149
                        f"""
2150
                        The {repr(cls.alias)} field in target {address} must only have globs for
2151
                        the target's directory, i.e. it cannot include values with `**` or
2152
                        `{os.path.sep}`. It was set to: {sorted(value or ())}
2153
                        """
2154
                    )
2155
                )
2156
        return value
1✔
2157

2158

2159
class OptionalSingleSourceField(SourcesField, StringField):
1✔
2160
    """The `source: str` field.
2161

2162
    See the docstring for `SourcesField` for some class properties you can set, such as
2163
    `expected_file_extensions`.
2164

2165
    When you need to get the sources for all targets, use `tgt.get(SourcesField)` rather than
2166
    `tgt.get(OptionalSingleSourceField)`.
2167

2168
    Use `SingleSourceField` if the source must exist.
2169
    """
2170

2171
    alias = "source"
1✔
2172
    help = help_text(
1✔
2173
        """
2174
        A single file that belongs to this target.
2175

2176
        Path is relative to the BUILD file's directory, e.g. `source='example.ext'`.
2177
        """
2178
    )
2179
    required = False
1✔
2180
    default: ClassVar[str | None] = None
1✔
2181
    expected_num_files: ClassVar[int | range] = range(0, 2)
1✔
2182

2183
    @classmethod
1✔
2184
    def compute_value(cls, raw_value: str | None, address: Address) -> str | None:
1✔
2185
        value_or_default = super().compute_value(raw_value, address)
1✔
2186
        if value_or_default is None:
1✔
2187
            return None
1✔
2188
        if value_or_default.startswith("../") or "/../" in value_or_default:
1✔
UNCOV
2189
            raise InvalidFieldException(
×
2190
                softwrap(
2191
                    f"""\
2192
                    The {repr(cls.alias)} field in target {address} should not include `../`
2193
                    patterns because targets can only have sources in the current directory or
2194
                    subdirectories. It was set to {value_or_default}. Instead, use a normalized
2195
                    literal file path (relative to the BUILD file).
2196
                    """
2197
                )
2198
            )
2199
        if "*" in value_or_default:
1✔
UNCOV
2200
            raise InvalidFieldException(
×
2201
                softwrap(
2202
                    f"""\
2203
                    The {repr(cls.alias)} field in target {address} should not include `*` globs,
2204
                    but was set to {value_or_default}. Instead, use a literal file path (relative
2205
                    to the BUILD file).
2206
                    """
2207
                )
2208
            )
2209
        if value_or_default.startswith("!"):
1✔
UNCOV
2210
            raise InvalidFieldException(
×
2211
                softwrap(
2212
                    f"""\
2213
                    The {repr(cls.alias)} field in target {address} should not start with `!`,
2214
                    which is usually used in the `sources` field to exclude certain files. Instead,
2215
                    use a literal file path (relative to the BUILD file).
2216
                    """
2217
                )
2218
            )
2219
        return value_or_default
1✔
2220

2221
    @property
1✔
2222
    def file_path(self) -> str | None:
1✔
2223
        """The path to the file, relative to the build root.
2224

2225
        This works without hydration because we validate that `*` globs and `!` ignores are not
2226
        used. However, consider still hydrating so that you verify the source file actually exists.
2227

2228
        The return type is optional because it's possible to have 0-1 files.
2229
        """
2230
        if self.value is None:
1✔
2231
            return None
1✔
2232
        return os.path.join(self.address.spec_path, self.value)
1✔
2233

2234
    @property
1✔
2235
    def globs(self) -> tuple[str, ...]:
1✔
2236
        if self.value is None:
1✔
UNCOV
2237
            return ()
×
2238
        return (self.value,)
1✔
2239

2240

2241
class SingleSourceField(OptionalSingleSourceField):
1✔
2242
    """The `source: str` field.
2243

2244
    Unlike `OptionalSingleSourceField`, the `.value` must be defined, whether by setting the
2245
    `default` or making the field `required`.
2246

2247
    See the docstring for `SourcesField` for some class properties you can set, such as
2248
    `expected_file_extensions`.
2249

2250
    When you need to get the sources for all targets, use `tgt.get(SourcesField)` rather than
2251
    `tgt.get(SingleSourceField)`.
2252
    """
2253

2254
    required = True
1✔
2255
    expected_num_files = 1
1✔
2256
    value: str
1✔
2257

2258
    @property
1✔
2259
    def file_path(self) -> str:
1✔
2260
        result = super().file_path
1✔
2261
        assert result is not None
1✔
2262
        return result
1✔
2263

2264

2265
@dataclass(frozen=True)
1✔
2266
class HydrateSourcesRequest(EngineAwareParameter):
1✔
2267
    field: SourcesField
1✔
2268
    for_sources_types: tuple[type[SourcesField], ...]
1✔
2269
    enable_codegen: bool
1✔
2270

2271
    def __init__(
1✔
2272
        self,
2273
        field: SourcesField,
2274
        *,
2275
        for_sources_types: Iterable[type[SourcesField]] = (SourcesField,),
2276
        enable_codegen: bool = False,
2277
    ) -> None:
2278
        """Convert raw sources globs into an instance of HydratedSources.
2279

2280
        If you only want to handle certain SourcesFields, such as only PythonSources, set
2281
        `for_sources_types`. Any invalid sources will return a `HydratedSources` instance with an
2282
        empty snapshot and `sources_type = None`.
2283

2284
        If `enable_codegen` is set to `True`, any codegen sources will try to be converted to one
2285
        of the `for_sources_types`.
2286
        """
2287
        object.__setattr__(self, "field", field)
1✔
2288
        object.__setattr__(self, "for_sources_types", tuple(for_sources_types))
1✔
2289
        object.__setattr__(self, "enable_codegen", enable_codegen)
1✔
2290

2291
        self.__post_init__()
1✔
2292

2293
    def __post_init__(self) -> None:
1✔
2294
        if self.enable_codegen and self.for_sources_types == (SourcesField,):
1✔
2295
            raise ValueError(
×
2296
                "When setting `enable_codegen=True` on `HydrateSourcesRequest`, you must also "
2297
                "explicitly set `for_source_types`. Why? `for_source_types` is used to "
2298
                "determine which language(s) to try to generate. For example, "
2299
                "`for_source_types=(PythonSources,)` will hydrate `PythonSources` like normal, "
2300
                "and, if it encounters codegen sources that can be converted into Python, it will "
2301
                "generate Python files."
2302
            )
2303

2304
    def debug_hint(self) -> str:
1✔
2305
        return self.field.address.spec
1✔
2306

2307

2308
@dataclass(frozen=True)
1✔
2309
class HydratedSources:
1✔
2310
    """The result of hydrating a SourcesField.
2311

2312
    The `sources_type` will indicate which of the `HydrateSourcesRequest.for_sources_type` the
2313
    result corresponds to, e.g. if the result comes from `FilesSources` vs. `PythonSources`. If this
2314
    value is None, then the input `SourcesField` was not one of the expected types; or, when codegen
2315
    was enabled in the request, there was no valid code generator to generate the requested language
2316
    from the original input. This property allows for switching on the result, e.g. handling
2317
    hydrated files() sources differently than hydrated Python sources.
2318
    """
2319

2320
    snapshot: Snapshot
1✔
2321
    filespec: Filespec
1✔
2322
    sources_type: type[SourcesField] | None
1✔
2323

2324

2325
@union(in_scope_types=[EnvironmentName])
1✔
2326
@dataclass(frozen=True)
1✔
2327
class GenerateSourcesRequest:
1✔
2328
    """A request to go from protocol sources -> a particular language.
2329

2330
    This should be subclassed for each distinct codegen implementation. The subclasses must define
2331
    the class properties `input` and `output`. The subclass must also be registered via
2332
    `UnionRule(GenerateSourcesRequest, GenerateFortranFromAvroRequest)`, for example.
2333

2334
    The rule to actually implement the codegen should take the subclass as input, and it must
2335
    return `GeneratedSources`.
2336

2337
    The `exportable` attribute disables the use of this codegen by the `export-codegen` goal when
2338
    set to False.
2339

2340
    For example:
2341

2342
        class GenerateFortranFromAvroRequest:
2343
            input = AvroSources
2344
            output = FortranSources
2345

2346
        @rule
2347
        async def generate_fortran_from_avro(request: GenerateFortranFromAvroRequest) -> GeneratedSources:
2348
            ...
2349

2350
        def rules():
2351
            return [
2352
                generate_fortran_from_avro,
2353
                UnionRule(GenerateSourcesRequest, GenerateFortranFromAvroRequest),
2354
            ]
2355
    """
2356

2357
    protocol_sources: Snapshot
1✔
2358
    protocol_target: Target
1✔
2359

2360
    input: ClassVar[type[SourcesField]]
1✔
2361
    output: ClassVar[type[SourcesField]]
1✔
2362

2363
    exportable: ClassVar[bool] = True
1✔
2364

2365

2366
@dataclass(frozen=True)
1✔
2367
class GeneratedSources:
1✔
2368
    snapshot: Snapshot
1✔
2369

2370

2371
@rule(polymorphic=True)
1✔
2372
async def generate_sources(
1✔
2373
    req: GenerateSourcesRequest, env_name: EnvironmentName
2374
) -> GeneratedSources:
2375
    raise NotImplementedError()
×
2376

2377

2378
class SourcesPaths(Paths):
1✔
2379
    """The resolved file names of the `source`/`sources` field.
2380

2381
    This does not consider codegen, and only captures the files from the field.
2382
    """
2383

2384

2385
@dataclass(frozen=True)
1✔
2386
class SourcesPathsRequest(EngineAwareParameter):
1✔
2387
    """A request to resolve the file names of the `source`/`sources` field.
2388

2389
    Use via `await resolve_source_paths(SourcesPathRequest(tgt.get(SourcesField))`.
2390

2391
    This is faster than `await hydrate_sources(HydrateSourcesRequest)` because it does not snapshot
2392
    the files and it only resolves the file names.
2393

2394
    This does not consider codegen, and only captures the files from the field. Use
2395
    `HydrateSourcesRequest` to use codegen.
2396
    """
2397

2398
    field: SourcesField
1✔
2399

2400
    def debug_hint(self) -> str:
1✔
UNCOV
2401
        return self.field.address.spec
×
2402

2403

2404
def targets_with_sources_types(
1✔
2405
    sources_types: Iterable[type[SourcesField]],
2406
    targets: Iterable[Target],
2407
    union_membership: UnionMembership,
2408
) -> tuple[Target, ...]:
2409
    """Return all targets either with the specified sources subclass(es) or which can generate those
2410
    sources."""
UNCOV
2411
    return tuple(
×
2412
        tgt
2413
        for tgt in targets
2414
        if any(
2415
            tgt.has_field(sources_type)
2416
            or tgt.get(SourcesField).can_generate(sources_type, union_membership)
2417
            for sources_type in sources_types
2418
        )
2419
    )
2420

2421

2422
# -----------------------------------------------------------------------------------------------
2423
# `Dependencies` field
2424
# -----------------------------------------------------------------------------------------------
2425

2426

2427
class Dependencies(StringSequenceField, AsyncFieldMixin):
1✔
2428
    """The dependencies field.
2429

2430
    To resolve all dependencies—including the results of dependency inference—use either
2431
    `await resolve_dependencies(DependenciesRequest(tgt[Dependencies])` or
2432
    `await resolve_targets(**implicitly(DependenciesRequest(tgt[Dependencies]))`.
2433
    """
2434

2435
    alias = "dependencies"
1✔
2436
    help = help_text(
1✔
2437
        f"""
2438
        Addresses to other targets that this target depends on, e.g.
2439
        `['helloworld/subdir:lib', 'helloworld/main.py:lib', '3rdparty:reqs#django']`.
2440

2441
        This augments any dependencies inferred by Pants, such as by analyzing your imports. Use
2442
        `{bin_name()} dependencies` or `{bin_name()} peek` on this target to get the final
2443
        result.
2444

2445
        See {doc_url("docs/using-pants/key-concepts/targets-and-build-files")} for more about how addresses are formed, including for generated
2446
        targets. You can also run `{bin_name()} list ::` to find all addresses in your project, or
2447
        `{bin_name()} list dir` to find all addresses defined in that directory.
2448

2449
        If the target is in the same BUILD file, you can leave off the BUILD file path, e.g.
2450
        `:tgt` instead of `helloworld/subdir:tgt`. For generated first-party addresses, use
2451
        `./` for the file path, e.g. `./main.py:tgt`; for all other generated targets,
2452
        use `:tgt#generated_name`.
2453

2454
        You may exclude dependencies by prefixing with `!`, e.g.
2455
        `['!helloworld/subdir:lib', '!./sibling.txt']`. Ignores are intended for false positives
2456
        with dependency inference; otherwise, simply leave off the dependency from the BUILD file.
2457
        """
2458
    )
2459
    supports_transitive_excludes = False
1✔
2460

2461
    @memoized_property
1✔
2462
    def unevaluated_transitive_excludes(self) -> UnparsedAddressInputs:
1✔
2463
        val = (
1✔
2464
            (v[2:] for v in self.value if v.startswith("!!"))
2465
            if self.supports_transitive_excludes and self.value
2466
            else ()
2467
        )
2468
        return UnparsedAddressInputs(
1✔
2469
            val,
2470
            owning_address=self.address,
2471
            description_of_origin=f"the `{self.alias}` field from the target {self.address}",
2472
        )
2473

2474

2475
@dataclass(frozen=True)
1✔
2476
class DependenciesRequest(EngineAwareParameter):
1✔
2477
    field: Dependencies
1✔
2478
    should_traverse_deps_predicate: ShouldTraverseDepsPredicate = TraverseIfDependenciesField()
1✔
2479

2480
    def debug_hint(self) -> str:
1✔
2481
        return self.field.address.spec
1✔
2482

2483

2484
# NB: ExplicitlyProvidedDependenciesRequest does not have a predicate unlike DependenciesRequest.
2485
@dataclass(frozen=True)
1✔
2486
class ExplicitlyProvidedDependenciesRequest(EngineAwareParameter):
1✔
2487
    field: Dependencies
1✔
2488

2489
    def debug_hint(self) -> str:
1✔
2490
        return self.field.address.spec
×
2491

2492

2493
@dataclass(frozen=True)
1✔
2494
class ExplicitlyProvidedDependencies:
1✔
2495
    """The literal addresses from a BUILD file `dependencies` field.
2496

2497
    Almost always, you should use `await resolve_dependencies(DependenciesRequest, **implicitly())`
2498
    instead, which will consider dependency inference and apply ignores. However, this type can be
2499
    useful particularly within inference rules to see if a user already explicitly
2500
    provided a dependency.
2501

2502
    Resolve using
2503
    `await determine_explicitly_provided_dependencies(**implicitly(DependenciesRequest))`.
2504

2505
    Note that the `includes` are not filtered based on the `ignores`: this type preserves exactly
2506
    what was in the BUILD file.
2507
    """
2508

2509
    address: Address
1✔
2510
    includes: FrozenOrderedSet[Address]
1✔
2511
    ignores: FrozenOrderedSet[Address]
1✔
2512

2513
    @memoized_method
1✔
2514
    def any_are_covered_by_includes(self, addresses: Iterable[Address]) -> bool:
1✔
2515
        """Return True if every address is in the explicitly provided includes.
2516

2517
        Note that if the input addresses are generated targets, they will still be marked as covered
2518
        if their original target generator is in the explicitly provided includes.
2519
        """
UNCOV
2520
        return any(
×
2521
            addr in self.includes or addr.maybe_convert_to_target_generator() in self.includes
2522
            for addr in addresses
2523
        )
2524

2525
    @memoized_method
1✔
2526
    def remaining_after_disambiguation(
1✔
2527
        self, addresses: Iterable[Address], owners_must_be_ancestors: bool
2528
    ) -> frozenset[Address]:
2529
        """All addresses that remain after ineligible candidates are discarded.
2530

2531
        Candidates are removed if they appear as ignores (`!` and `!!)` in the `dependencies`
2532
        field. Note that if the input addresses are generated targets, they will still be marked as
2533
        covered if their original target generator is in the explicitly provided ignores.
2534

2535
        Candidates are also removed if `owners_must_be_ancestors` is True and the targets are not
2536
        ancestors, e.g. `root2:tgt` is not a valid candidate for something defined in `root1`.
2537
        """
UNCOV
2538
        original_addr_path = PurePath(self.address.spec_path)
×
2539

UNCOV
2540
        def is_valid(addr: Address) -> bool:
×
UNCOV
2541
            is_ignored = (
×
2542
                addr in self.ignores or addr.maybe_convert_to_target_generator() in self.ignores
2543
            )
UNCOV
2544
            if owners_must_be_ancestors is False:
×
UNCOV
2545
                return not is_ignored
×
2546
            # NB: `PurePath.is_relative_to()` was not added until Python 3.9. This emulates it.
UNCOV
2547
            try:
×
UNCOV
2548
                original_addr_path.relative_to(addr.spec_path)
×
UNCOV
2549
                return not is_ignored
×
UNCOV
2550
            except ValueError:
×
UNCOV
2551
                return False
×
2552

UNCOV
2553
        return frozenset(filter(is_valid, addresses))
×
2554

2555
    def maybe_warn_of_ambiguous_dependency_inference(
1✔
2556
        self,
2557
        ambiguous_addresses: Iterable[Address],
2558
        original_address: Address,
2559
        *,
2560
        context: str,
2561
        import_reference: str,
2562
        owners_must_be_ancestors: bool = False,
2563
    ) -> None:
2564
        """If the module is ambiguous and the user did not disambiguate, warn that dependency
2565
        inference will not be used.
2566

2567
        Disambiguation usually happens by using ignores in the `dependencies` field with `!` and
2568
        `!!`. If `owners_must_be_ancestors` is True, any addresses which are not ancestors of the
2569
        target in question will also be ignored.
2570
        """
2571
        if not ambiguous_addresses or self.any_are_covered_by_includes(ambiguous_addresses):
1✔
2572
            return
1✔
UNCOV
2573
        remaining = self.remaining_after_disambiguation(
×
2574
            ambiguous_addresses, owners_must_be_ancestors=owners_must_be_ancestors
2575
        )
UNCOV
2576
        if len(remaining) <= 1:
×
UNCOV
2577
            return
×
UNCOV
2578
        logger.warning(
×
2579
            f"{context}, but Pants cannot safely infer a dependency because more than one target "
2580
            f"owns this {import_reference}, so it is ambiguous which to use: "
2581
            f"{sorted(addr.spec for addr in remaining)}."
2582
            f"\n\nPlease explicitly include the dependency you want in the `dependencies` "
2583
            f"field of {original_address}, or ignore the ones you do not want by prefixing "
2584
            f"with `!` or `!!` so that one or no targets are left."
2585
            f"\n\nAlternatively, you can remove the ambiguity by deleting/changing some of the "
2586
            f"targets so that only 1 target owns this {import_reference}. Refer to "
2587
            f"{doc_url('docs/using-pants/troubleshooting-common-issues#import-errors-and-missing-dependencies')}."
2588
        )
2589

2590
    def disambiguated(
1✔
2591
        self, ambiguous_addresses: Iterable[Address], owners_must_be_ancestors: bool = False
2592
    ) -> Address | None:
2593
        """If exactly one of the input addresses remains after disambiguation, return it.
2594

2595
        Disambiguation usually happens by using ignores in the `dependencies` field with `!` and
2596
        `!!`. If `owners_must_be_ancestors` is True, any addresses which are not ancestors of the
2597
        target in question will also be ignored.
2598
        """
2599
        if not ambiguous_addresses or self.any_are_covered_by_includes(ambiguous_addresses):
1✔
2600
            return None
1✔
UNCOV
2601
        remaining_after_ignores = self.remaining_after_disambiguation(
×
2602
            ambiguous_addresses, owners_must_be_ancestors=owners_must_be_ancestors
2603
        )
UNCOV
2604
        return list(remaining_after_ignores)[0] if len(remaining_after_ignores) == 1 else None
×
2605

2606

2607
FS = TypeVar("FS", bound="FieldSet")
1✔
2608

2609

2610
@union(in_scope_types=[EnvironmentName])
1✔
2611
@dataclass(frozen=True)
1✔
2612
class InferDependenciesRequest(Generic[FS], EngineAwareParameter):
1✔
2613
    """A request to infer dependencies by analyzing source files.
2614

2615
    To set up a new inference implementation, subclass this class. Set the class property
2616
    `infer_from` to the FieldSet subclass you are able to infer from. This will cause the FieldSet
2617
    class, and any subclass, to use your inference implementation.
2618

2619
    Note that there cannot be more than one implementation for a particular `FieldSet` class.
2620

2621
    Register this subclass with `UnionRule(InferDependenciesRequest, InferFortranDependencies)`, for example.
2622

2623
    Then, create a rule that takes the subclass as a parameter and returns `InferredDependencies`.
2624

2625
    For example:
2626

2627
        class InferFortranDependencies(InferDependenciesRequest):
2628
            infer_from = FortranDependenciesInferenceFieldSet
2629

2630
        @rule
2631
        async def infer_fortran_dependencies(request: InferFortranDependencies) -> InferredDependencies:
2632
            hydrated_sources = await hydrate_sources(HydrateSources(request.field_set.sources))
2633
            ...
2634
            return InferredDependencies(...)
2635

2636
        def rules():
2637
            return [
2638
                infer_fortran_dependencies,
2639
                UnionRule(InferDependenciesRequest, InferFortranDependencies),
2640
            ]
2641
    """
2642

2643
    infer_from: ClassVar[type[FS]]
1✔
2644

2645
    field_set: FS
1✔
2646

2647

2648
@dataclass(frozen=True)
1✔
2649
class InferredDependencies:
1✔
2650
    include: FrozenOrderedSet[Address]
1✔
2651
    exclude: FrozenOrderedSet[Address]
1✔
2652

2653
    def __init__(
1✔
2654
        self,
2655
        include: Iterable[Address],
2656
        *,
2657
        exclude: Iterable[Address] = (),
2658
    ) -> None:
2659
        """The result of inferring dependencies."""
2660
        object.__setattr__(self, "include", FrozenOrderedSet(sorted(include)))
1✔
2661
        object.__setattr__(self, "exclude", FrozenOrderedSet(sorted(exclude)))
1✔
2662

2663

2664
@union(in_scope_types=[EnvironmentName])
1✔
2665
@dataclass(frozen=True)
1✔
2666
class TransitivelyExcludeDependenciesRequest(Generic[FS], EngineAwareParameter):
1✔
2667
    """A request to transitvely exclude dependencies of a "root" node.
2668

2669
    This is similar to `InferDependenciesRequest`, except the request is only made for "root" nodes
2670
    in the dependency graph.
2671

2672
    This mirrors the public facing "transitive exclude" dependency feature (i.e. `!!<address>`).
2673
    """
2674

2675
    infer_from: ClassVar[type[FS]]
1✔
2676

2677
    field_set: FS
1✔
2678

2679

2680
class TransitivelyExcludeDependencies(FrozenOrderedSet[Address]):
1✔
2681
    pass
1✔
2682

2683

2684
@union(in_scope_types=[EnvironmentName])
1✔
2685
@dataclass(frozen=True)
1✔
2686
class ValidateDependenciesRequest(Generic[FS], ABC):
1✔
2687
    """A request to validate dependencies after they have been computed.
2688

2689
    An implementing rule should raise an exception if dependencies are invalid.
2690
    """
2691

2692
    field_set_type: ClassVar[type[FS]]
1✔
2693

2694
    field_set: FS
1✔
2695
    dependencies: Addresses
1✔
2696

2697

2698
@dataclass(frozen=True)
1✔
2699
class ValidatedDependencies:
1✔
2700
    pass
1✔
2701

2702

2703
@dataclass(frozen=True)
1✔
2704
class DependenciesRuleApplicationRequest:
1✔
2705
    """A request to return the applicable dependency rule action for each dependency of a target."""
2706

2707
    address: Address
1✔
2708
    dependencies: Addresses
1✔
2709
    description_of_origin: str = dataclasses.field(hash=False, compare=False)
1✔
2710

2711

2712
@dataclass(frozen=True)
1✔
2713
class DependenciesRuleApplication:
1✔
2714
    """Maps all dependencies to their respective dependency rule application of an origin target
2715
    address.
2716

2717
    The `applications` will be empty and the `address` `None` if there is no dependency rule
2718
    implementation.
2719
    """
2720

2721
    address: Address | None = None
1✔
2722
    dependencies_rule: FrozenDict[Address, DependencyRuleApplication] = FrozenDict()
1✔
2723

2724
    def __post_init__(self):
1✔
UNCOV
2725
        if self.dependencies_rule and self.address is None:
×
2726
            raise ValueError(
×
2727
                "The `address` field must not be None when there are `dependencies_rule`s."
2728
            )
2729

2730
    @classmethod
1✔
2731
    @memoized_method
1✔
2732
    def allow_all(cls) -> DependenciesRuleApplication:
1✔
2733
        return cls()
×
2734

2735
    def execute_actions(self) -> None:
1✔
UNCOV
2736
        errors = [
×
2737
            action_error.replace("\n", "\n    ")
2738
            for action_error in (rule.execute() for rule in self.dependencies_rule.values())
2739
            if action_error is not None
2740
        ]
UNCOV
2741
        if errors:
×
UNCOV
2742
            err_count = len(errors)
×
UNCOV
2743
            raise DependencyRuleActionDeniedError(
×
2744
                softwrap(
2745
                    f"""
2746
                    {self.address} has {pluralize(err_count, "dependency violation")}:
2747

2748
                    {bullet_list(errors)}
2749
                    """
2750
                )
2751
            )
2752

2753

2754
class SpecialCasedDependencies(StringSequenceField, AsyncFieldMixin):
1✔
2755
    """Subclass this for fields that act similarly to the `dependencies` field, but are handled
2756
    differently than normal dependencies.
2757

2758
    For example, you might have a field for package/binary dependencies, which you will call
2759
    the equivalent of `./pants package` on. While you could put these in the normal
2760
    `dependencies` field, it is often clearer to the user to call out this magic through a
2761
    dedicated field.
2762

2763
    This type will ensure that the dependencies show up in project introspection,
2764
    like `dependencies` and `dependents`, but not show up when you
2765
    `await transitive_targets(TransitiveTargetsRequest(...), **implicitly())` and
2766
    `await resolve_dependencies(DependenciesRequest(...), **implicitly())`.
2767

2768
    To hydrate this field's dependencies, use
2769
    `await resolve_unparsed_address_inputs(tgt.get(MyField).to_unparsed_address_inputs(), **implicitly())`.
2770
    """
2771

2772
    def to_unparsed_address_inputs(self) -> UnparsedAddressInputs:
1✔
2773
        return UnparsedAddressInputs(
1✔
2774
            self.value or (),
2775
            owning_address=self.address,
2776
            description_of_origin=f"the `{self.alias}` from the target {self.address}",
2777
        )
2778

2779

2780
# -----------------------------------------------------------------------------------------------
2781
# Other common Fields used across most targets
2782
# -----------------------------------------------------------------------------------------------
2783

2784

2785
class Tags(StringSequenceField):
1✔
2786
    alias = "tags"
1✔
2787
    help = help_text(
1✔
2788
        f"""
2789
        Arbitrary strings to describe a target.
2790

2791
        For example, you may tag some test targets with 'integration_test' so that you could run
2792
        `{bin_name()} --tag='integration_test' test ::`  to only run on targets with that tag.
2793
        """
2794
    )
2795

2796

2797
class DescriptionField(StringField):
1✔
2798
    alias = "description"
1✔
2799
    help = help_text(
1✔
2800
        f"""
2801
        A human-readable description of the target.
2802

2803
        Use `{bin_name()} list --documented ::` to see all targets with descriptions.
2804
        """
2805
    )
2806

2807

2808
COMMON_TARGET_FIELDS = (Tags, DescriptionField)
1✔
2809

2810

2811
class OverridesField(AsyncFieldMixin, Field):
1✔
2812
    """A mapping of keys (e.g. target names, source globs) to field names with their overridden
2813
    values.
2814

2815
    This is meant for target generators to reduce boilerplate. It's up to the corresponding target
2816
    generator rule to determine how to implement the field, such as how users specify the key. For
2817
    example, `{"f.ext": {"tags": ['my_tag']}}`.
2818
    """
2819

2820
    alias = "overrides"
1✔
2821
    value: dict[tuple[str, ...], dict[str, Any]] | None
1✔
2822
    default: ClassVar[None] = None  # A default does not make sense for this field.
1✔
2823

2824
    @classmethod
1✔
2825
    def compute_value(
1✔
2826
        cls,
2827
        raw_value: dict[str | tuple[str, ...], dict[str, Any]] | None,
2828
        address: Address,
2829
    ) -> FrozenDict[tuple[str, ...], FrozenDict[str, ImmutableValue]] | None:
2830
        value_or_default = super().compute_value(raw_value, address)
1✔
2831
        if value_or_default is None:
1✔
2832
            return None
1✔
2833

2834
        def invalid_type_exception() -> InvalidFieldException:
1✔
UNCOV
2835
            return InvalidFieldTypeException(
×
2836
                address,
2837
                cls.alias,
2838
                raw_value,
2839
                expected_type="dict[str | tuple[str, ...], dict[str, Any]]",
2840
            )
2841

2842
        if not isinstance(value_or_default, collections.abc.Mapping):
1✔
UNCOV
2843
            raise invalid_type_exception()
×
2844

2845
        result: dict[tuple[str, ...], FrozenDict[str, ImmutableValue]] = {}
1✔
2846
        for outer_key, nested_value in value_or_default.items():
1✔
2847
            if isinstance(outer_key, str):
1✔
2848
                outer_key = (outer_key,)
1✔
2849
            if not isinstance(outer_key, collections.abc.Sequence) or not all(
1✔
2850
                isinstance(elem, str) for elem in outer_key
2851
            ):
UNCOV
2852
                raise invalid_type_exception()
×
2853
            if not isinstance(nested_value, collections.abc.Mapping):
1✔
UNCOV
2854
                raise invalid_type_exception()
×
2855
            if not all(isinstance(inner_key, str) for inner_key in nested_value):
1✔
UNCOV
2856
                raise invalid_type_exception()
×
2857
            result[tuple(outer_key)] = FrozenDict.deep_freeze(cast(Mapping[str, Any], nested_value))
1✔
2858

2859
        return FrozenDict(result)
1✔
2860

2861
    @classmethod
1✔
2862
    def to_path_globs(
1✔
2863
        cls,
2864
        address: Address,
2865
        overrides_keys: Iterable[str],
2866
        unmatched_build_file_globs: UnmatchedBuildFileGlobs,
2867
    ) -> tuple[PathGlobs, ...]:
2868
        """Create a `PathGlobs` for each key.
2869

2870
        This should only be used if the keys are file globs.
2871
        """
2872

2873
        def relativize_glob(glob: str) -> str:
1✔
UNCOV
2874
            return (
×
2875
                f"!{os.path.join(address.spec_path, glob[1:])}"
2876
                if glob.startswith("!")
2877
                else os.path.join(address.spec_path, glob)
2878
            )
2879

2880
        return tuple(
1✔
2881
            PathGlobs(
2882
                [relativize_glob(glob)],
2883
                glob_match_error_behavior=unmatched_build_file_globs.error_behavior,
2884
                description_of_origin=f"the `overrides` field for {address}",
2885
            )
2886
            for glob in overrides_keys
2887
        )
2888

2889
    def flatten(self) -> dict[str, dict[str, Any]]:
1✔
2890
        """Combine all overrides for every key into a single dictionary."""
2891
        result: dict[str, dict[str, Any]] = {}
1✔
2892
        for keys, override in (self.value or {}).items():
1✔
2893
            for key in keys:
1✔
2894
                for field, value in override.items():
1✔
2895
                    if key not in result:
1✔
2896
                        result[key] = {field: value}
1✔
2897
                        continue
1✔
2898
                    if field not in result[key]:
1✔
2899
                        result[key][field] = value
1✔
2900
                        continue
1✔
2901
                    raise InvalidFieldException(
×
2902
                        f"Conflicting overrides in the `{self.alias}` field of "
2903
                        f"`{self.address}` for the key `{key}` for "
2904
                        f"the field `{field}`. You cannot specify the same field name "
2905
                        "multiple times for the same key.\n\n"
2906
                        f"(One override sets the field to `{repr(result[key][field])}` "
2907
                        f"but another sets to `{repr(value)}`.)"
2908
                    )
2909
        return result
1✔
2910

2911
    @classmethod
1✔
2912
    def flatten_paths(
1✔
2913
        cls,
2914
        address: Address,
2915
        paths_and_overrides: Iterable[tuple[Paths, PathGlobs, dict[str, Any]]],
2916
    ) -> dict[str, dict[str, Any]]:
2917
        """Combine all overrides for each file into a single dictionary."""
2918
        result: dict[str, dict[str, Any]] = {}
1✔
2919
        for paths, globs, override in paths_and_overrides:
1✔
2920
            # NB: If some globs did not result in any Paths, we preserve them to ensure that
2921
            # unconsumed overrides trigger errors during generation.
UNCOV
2922
            for path in paths.files or globs.globs:
×
UNCOV
2923
                for field, value in override.items():
×
UNCOV
2924
                    if path not in result:
×
UNCOV
2925
                        result[path] = {field: value}
×
UNCOV
2926
                        continue
×
UNCOV
2927
                    if field not in result[path]:
×
UNCOV
2928
                        result[path][field] = value
×
UNCOV
2929
                        continue
×
UNCOV
2930
                    relpath = fast_relpath(path, address.spec_path)
×
UNCOV
2931
                    raise InvalidFieldException(
×
2932
                        f"Conflicting overrides for `{address}` for the relative path "
2933
                        f"`{relpath}` for the field `{field}`. You cannot specify the same field "
2934
                        f"name multiple times for the same path.\n\n"
2935
                        f"(One override sets the field to `{repr(result[path][field])}` "
2936
                        f"but another sets to `{repr(value)}`.)"
2937
                    )
2938
        return result
1✔
2939

2940

2941
def generate_multiple_sources_field_help_message(files_example: str) -> str:
1✔
2942
    return softwrap(
1✔
2943
        """
2944
        A list of files and globs that belong to this target.
2945

2946
        Paths are relative to the BUILD file's directory. You can ignore files/globs by
2947
        prefixing them with `!`.
2948

2949
        """
2950
        + files_example
2951
    )
2952

2953

2954
def generate_file_based_overrides_field_help_message(
1✔
2955
    generated_target_name: str, example: str
2956
) -> str:
2957
    example = textwrap.dedent(example.lstrip("\n"))  # noqa: PNT20
1✔
2958
    example = textwrap.indent(example, " " * 4)
1✔
2959
    return "\n".join(
1✔
2960
        [
2961
            softwrap(
2962
                f"""
2963
                Override the field values for generated `{generated_target_name}` targets.
2964

2965
                Expects a dictionary of relative file paths and globs to a dictionary for the
2966
                overrides. You may either use a string for a single path / glob,
2967
                or a string tuple for multiple paths / globs. Each override is a dictionary of
2968
                field names to the overridden value.
2969

2970
                For example:
2971

2972
                {example}
2973
                """
2974
            ),
2975
            "",
2976
            softwrap(
2977
                f"""
2978
                File paths and globs are relative to the BUILD file's directory. Every overridden file is
2979
                validated to belong to this target's `sources` field.
2980

2981
                If you'd like to override a field's value for every `{generated_target_name}` target
2982
                generated by this target, change the field directly on this target rather than using the
2983
                `overrides` field.
2984

2985
                You can specify the same file name in multiple keys, so long as you don't override the
2986
                same field more than one time for the file.
2987
                """
2988
            ),
2989
        ],
2990
    )
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