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

pantsbuild / pants / 24055979590

06 Apr 2026 11:17PM UTC coverage: 52.37% (-40.5%) from 92.908%
24055979590

Pull #23225

github

web-flow
Merge 67474653c into 542ca048d
Pull Request #23225: Add --test-show-all-batch-targets to expose all targets in batched pytest

6 of 17 new or added lines in 2 files covered. (35.29%)

23030 existing lines in 605 files now uncovered.

31643 of 60422 relevant lines covered (52.37%)

1.05 hits per line

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

79.53
/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
2✔
5

6
import collections.abc
2✔
7
import dataclasses
2✔
8
import enum
2✔
9
import glob as glob_stdlib
2✔
10
import itertools
2✔
11
import logging
2✔
12
import os.path
2✔
13
import textwrap
2✔
14
import zlib
2✔
15
from abc import ABC, ABCMeta, abstractmethod
2✔
16
from collections import deque
2✔
17
from collections.abc import Callable, Iterable, Iterator, KeysView, Mapping, Sequence
2✔
18
from dataclasses import dataclass
2✔
19
from enum import Enum
2✔
20
from operator import attrgetter
2✔
21
from pathlib import PurePath
2✔
22
from typing import (
2✔
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
2✔
35
from pants.engine.addresses import Address, Addresses, UnparsedAddressInputs, assert_single_address
2✔
36
from pants.engine.collection import Collection
2✔
37
from pants.engine.engine_aware import EngineAwareParameter
2✔
38
from pants.engine.environment import EnvironmentName
2✔
39
from pants.engine.fs import (
2✔
40
    GlobExpansionConjunction,
41
    GlobMatchErrorBehavior,
42
    PathGlobs,
43
    Paths,
44
    Snapshot,
45
)
46
from pants.engine.internals.dep_rules import (
2✔
47
    DependencyRuleActionDeniedError,
48
    DependencyRuleApplication,
49
)
50
from pants.engine.internals.native_engine import NO_VALUE as NO_VALUE  # noqa: F401
2✔
51
from pants.engine.internals.native_engine import AsyncFieldMixin as AsyncFieldMixin
2✔
52
from pants.engine.internals.native_engine import BoolField as BoolField  # noqa: F401
2✔
53
from pants.engine.internals.native_engine import Field as Field
2✔
54
from pants.engine.internals.native_engine import ScalarField as ScalarField
2✔
55
from pants.engine.internals.native_engine import SequenceField as SequenceField  # noqa: F401
2✔
56
from pants.engine.internals.native_engine import StringField as StringField
2✔
57
from pants.engine.internals.native_engine import StringSequenceField as StringSequenceField
2✔
58
from pants.engine.internals.native_engine import TriBoolField as TriBoolField  # noqa: F401
2✔
59
from pants.engine.internals.target_adaptor import SourceBlock, SourceBlocks  # noqa: F401
2✔
60
from pants.engine.rules import rule
2✔
61
from pants.engine.unions import UnionMembership, UnionRule, distinct_union_type_per_subclass, union
2✔
62
from pants.option.bootstrap_options import UnmatchedBuildFileGlobs
2✔
63
from pants.source.filespec import Filespec, FilespecMatcher
2✔
64
from pants.util.collections import ensure_str_list
2✔
65
from pants.util.dirutil import fast_relpath
2✔
66
from pants.util.docutil import bin_name, doc_url
2✔
67
from pants.util.frozendict import FrozenDict
2✔
68
from pants.util.memo import memoized_classproperty, memoized_method, memoized_property
2✔
69
from pants.util.ordered_set import FrozenOrderedSet
2✔
70
from pants.util.strutil import bullet_list, help_text, pluralize, softwrap
2✔
71

72
logger = logging.getLogger(__name__)
2✔
73

74
# -----------------------------------------------------------------------------------------------
75
# Core Field abstractions
76
# -----------------------------------------------------------------------------------------------
77

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

82

83
@union
2✔
84
@dataclass(frozen=True)
2✔
85
class FieldDefaultFactoryRequest:
2✔
86
    """Registers a dynamic default for a Field.
87

88
    See `FieldDefaults`.
89
    """
90

91
    field_type: ClassVar[type[Field]]
2✔
92

93

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

100

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

105
    default_factory: FieldDefaultFactory
2✔
106

107

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

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

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

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

129
    _factories: FrozenDict[type[Field], FieldDefaultFactory]
2✔
130

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

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

UNCOV
142
        return lambda f: f.value
×
143

144
    def value_or_default(self, field: Field) -> Any:
2✔
UNCOV
145
        return (self.factory(type(field)))(field)
×
146

147

148
# -----------------------------------------------------------------------------------------------
149
# Core Target abstractions
150
# -----------------------------------------------------------------------------------------------
151

152

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

157

158
@dataclass(frozen=True)
2✔
159
class Target:
2✔
160
    """A Target represents an addressable set of metadata.
161

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

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

172
    removal_version: ClassVar[str | None] = None
2✔
173
    removal_hint: ClassVar[str | None] = None
2✔
174

175
    deprecated_alias: ClassVar[str | None] = None
2✔
176
    deprecated_alias_removal_version: ClassVar[str | None] = None
2✔
177

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

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

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

233
        if origin_sources_blocks:
2✔
UNCOV
234
            _validate_origin_sources_blocks(origin_sources_blocks)
×
235

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

257
            self.validate()
2✔
258
        except Exception as e:
2✔
259
            raise InvalidTargetException(
2✔
260
                str(e), description_of_origin=self.description_of_origin
261
            ) from e
262

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

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

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

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

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

327
    @distinct_union_type_per_subclass
2✔
328
    class PluginField:
2✔
329
        pass
2✔
330

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

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

347
    def __hash__(self) -> int:
2✔
348
        return hash((self.__class__, self.address, self.residence_dir, self.field_values))
2✔
349

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

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

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

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

381
        return tuple(sorted(result, key=attrgetter("alias")))
2✔
382

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

573

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

590

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

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

599
    address: Address
2✔
600
    description_of_origin: str = dataclasses.field(hash=False, compare=False)
2✔
601

602

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

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

611
    target: Target
2✔
612

613

614
class Targets(Collection[Target]):
2✔
615
    """A heterogeneous collection of instances of Target subclasses.
616

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

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

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

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

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

634

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

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

649
    def expect_single(self) -> Target:
2✔
650
        assert_single_address([tgt.address for tgt in self])
×
651
        return self[0]
×
652

653

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

658
    def expect_single(self) -> Target:
2✔
659
        assert_single_address([tgt.address for tgt in self])
×
660
        return self[0]
×
661

662

663
class DepsTraversalBehavior(Enum):
2✔
664
    """The return value for ShouldTraverseDepsPredicate.
665

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

672
    INCLUDE = "include"
2✔
673
    EXCLUDE = "exclude"
2✔
674

675

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

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

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

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

698
    def __post_init__(self):
2✔
699
        object.__setattr__(self, "_callable", type(self).__call__)
2✔
700

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

707

708
class TraverseIfDependenciesField(ShouldTraverseDepsPredicate):
2✔
709
    """This is the default ShouldTraverseDepsPredicate implementation.
710

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

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

722

723
class AlwaysTraverseDeps(ShouldTraverseDepsPredicate):
2✔
724
    """A predicate to use when a request needs all deps.
725

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

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

734

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

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

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

750
    def debug_hint(self) -> str:
2✔
751
        return str(self)
×
752

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

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

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

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

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

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

784
    def __hash__(self) -> int:
2✔
785
        return self._hashcode
2✔
786

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

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

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

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

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

818
    def __repr__(self) -> str:
2✔
819
        return f"{self.__class__.__name__}({str(self)})"
×
820

821

822
class CoarsenedTargets(Collection[CoarsenedTarget]):
2✔
823
    """The CoarsenedTarget roots of a transitive graph walk for some addresses.
824

825
    To collect all reachable CoarsenedTarget members, use `def closure`.
826
    """
827

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

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

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

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

850
    def __hash__(self):
2✔
851
        return super().__hash__()
2✔
852

853

854
@dataclass(frozen=True)
2✔
855
class CoarsenedTargetsRequest:
2✔
856
    """A request to get CoarsenedTargets for input roots."""
857

858
    roots: tuple[Address, ...]
2✔
859
    expanded_targets: bool
2✔
860
    should_traverse_deps_predicate: ShouldTraverseDepsPredicate
2✔
861

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

873

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

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

882
    roots: tuple[Target, ...]
2✔
883
    dependencies: FrozenOrderedSet[Target]
2✔
884

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

890

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

895
    roots: tuple[Address, ...]
2✔
896
    should_traverse_deps_predicate: ShouldTraverseDepsPredicate
2✔
897

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

907

908
@dataclass(frozen=True)
2✔
909
class RegisteredTargetTypes:
2✔
910
    aliases_to_types: FrozenDict[str, type[Target]]
2✔
911

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

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

924
    @property
2✔
925
    def aliases(self) -> FrozenOrderedSet[str]:
2✔
926
        return FrozenOrderedSet(self.aliases_to_types.keys())
2✔
927

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

932

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

937

938
class AllUnexpandedTargets(Collection[Target]):
2✔
939
    """All targets in the project, including generated targets.
940

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

946

947
# -----------------------------------------------------------------------------------------------
948
# Target generation
949
# -----------------------------------------------------------------------------------------------
950

951

952
class TargetGenerator(Target):
2✔
953
    """A Target type which generates other Targets via installed `@rule` logic.
954

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

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

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

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

990
    @distinct_union_type_per_subclass
2✔
991
    class MovedPluginField:
2✔
992
        """A plugin field that should be moved into the generated targets."""
993

994
    def validate(self) -> None:
2✔
995
        super().validate()
2✔
996

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

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

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

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

1038
        return tuple(result)
2✔
1039

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

1048

1049
class TargetFilesGenerator(TargetGenerator):
2✔
1050
    """A TargetGenerator which generates a Target per file matched by the generator.
1051

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

1057
    settings_request_cls: ClassVar[type[TargetFilesGeneratorSettingsRequest] | None] = None
2✔
1058

1059
    def validate(self) -> None:
2✔
1060
        super().validate()
2✔
1061

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

1071

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

1076
    See `TargetFilesGenerator`.
1077
    """
1078

1079

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

1094

1095
_TargetGenerator = TypeVar("_TargetGenerator", bound=TargetGenerator)
2✔
1096

1097

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

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

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

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

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

1133

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

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

1167

1168
@rule(polymorphic=True)
2✔
1169
async def generate_targets(req: GenerateTargetsRequest) -> GeneratedTargets:
2✔
1170
    raise NotImplementedError()
×
1171

1172

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

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

1186

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

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

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

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

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

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

1221
    normalized_overrides = dict(overrides or {})
2✔
1222

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

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

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

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

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

1277
    result = tuple(
2✔
1278
        gen_tgt(address, full_fp, fields) for address, full_fp, fields in all_generated_items
1279
    )
1280

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

1296
    return GeneratedTargets(generator, result)
2✔
1297

1298

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

1312

1313
_FS = TypeVar("_FS", bound="FieldSet")
2✔
1314

1315

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

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

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

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

1330
    For example:
1331

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

1336
            sources: FortranSources
1337
            fortran_version: FortranVersion
1338

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

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

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

1351
    FieldSets are consumed like any normal dataclass:
1352

1353
        print(field_set.address)
1354
        print(field_set.sources)
1355
    """
1356

1357
    required_fields: ClassVar[tuple[type[Field], ...]]
2✔
1358

1359
    address: Address
2✔
1360

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

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

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

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

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

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

1406
    def debug_hint(self) -> str:
2✔
1407
        return self.address.spec
2✔
1408

1409
    def metadata(self) -> dict[str, Any]:
2✔
1410
        return {"address": self.address.spec}
2✔
1411

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

1417

1418
@dataclass(frozen=True)
2✔
1419
class TargetRootsToFieldSets(Generic[_FS]):
2✔
1420
    mapping: FrozenDict[Target, tuple[_FS, ...]]
2✔
1421

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

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

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

1441

1442
class NoApplicableTargetsBehavior(Enum):
2✔
1443
    ignore = "ignore"
2✔
1444
    warn = "warn"
2✔
1445
    error = "error"
2✔
1446

1447

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

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

1467

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

1474

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

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

1498
    def is_in_shard(self, key: str) -> bool:
2✔
1499
        return get_shard(key, self.num_shards) == self.shard
×
1500

1501

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

1507
    def __init__(self, collection: Iterable[Iterable[_FS]]):
2✔
1508
        object.__setattr__(self, "collection", tuple(tuple(iterable) for iterable in collection))
2✔
1509

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

1514

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

1520
    def __init__(self, field_set_superclass: type[_FS], targets: Iterable[Target]):
2✔
1521
        object.__setattr__(self, "field_set_superclass", field_set_superclass)
2✔
1522
        object.__setattr__(self, "targets", tuple(targets))
2✔
1523

1524

1525
# -----------------------------------------------------------------------------------------------
1526
# Exception messages
1527
# -----------------------------------------------------------------------------------------------
1528

1529

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

1533
    Suggested template:
1534

1535
         f"The `{alias!r}` target {address} ..."
1536
    """
1537

1538
    def __init__(self, message: Any, *, description_of_origin: str | None = None) -> None:
2✔
1539
        self.description_of_origin = description_of_origin
2✔
1540
        super().__init__(message)
2✔
1541

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

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

1552

1553
class InvalidGeneratedTargetException(InvalidTargetException):
2✔
1554
    pass
2✔
1555

1556

1557
class InvalidFieldException(Exception):
2✔
1558
    """Use when there's an issue with a particular field.
1559

1560
    Suggested template:
1561

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

1565
    def __init__(self, message: Any, *, description_of_origin: str | None = None) -> None:
2✔
1566
        self.description_of_origin = description_of_origin
2✔
1567
        super().__init__(message)
2✔
1568

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

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

1579

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

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

1600

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

1626

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

1636

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

1653

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

1668
                All valid target types: {sorted(registered_target_types.aliases)}
1669

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

1673
                """
1674
            ),
1675
            description_of_origin=description_of_origin,
1676
        )
1677

1678

1679
# -----------------------------------------------------------------------------------------------
1680
# Field templates
1681
# -----------------------------------------------------------------------------------------------
1682

1683
T = TypeVar("T")
2✔
1684

1685

1686
class ValidNumbers(Enum):
2✔
1687
    """What range of numbers are allowed for IntField and FloatField."""
1688

1689
    positive_only = enum.auto()
2✔
1690
    positive_and_zero = enum.auto()
2✔
1691
    all = enum.auto()
2✔
1692

1693
    def validate(self, num: float | int | None, alias: str, address: Address) -> None:
2✔
1694
        if num is None or self == self.all:
2✔
1695
            return
2✔
1696
        if self == self.positive_and_zero:
2✔
1697
            if num < 0:
2✔
UNCOV
1698
                raise InvalidFieldException(
×
1699
                    f"The {repr(alias)} field in target {address} must be greater than or equal to "
1700
                    f"zero, but was set to `{num}`."
1701
                )
1702
            return
2✔
1703
        if num <= 0:
2✔
UNCOV
1704
            raise InvalidFieldException(
×
1705
                f"The {repr(alias)} field in target {address} must be greater than zero, but was "
1706
                f"set to `{num}`."
1707
            )
1708

1709

1710
class IntField(ScalarField[int]):
2✔
1711
    expected_type = int
2✔
1712
    expected_type_description = "an integer"
2✔
1713
    valid_numbers: ClassVar[ValidNumbers] = ValidNumbers.all
2✔
1714

1715
    @classmethod
2✔
1716
    def compute_value(cls, raw_value: int | None, address: Address) -> int | None:
2✔
1717
        value_or_default = super().compute_value(raw_value, address)
2✔
1718
        cls.valid_numbers.validate(value_or_default, cls.alias, address)
2✔
1719
        return value_or_default
2✔
1720

1721

1722
class FloatField(ScalarField[float]):
2✔
1723
    expected_type = float
2✔
1724
    expected_type_description = "a float"
2✔
1725
    valid_numbers: ClassVar[ValidNumbers] = ValidNumbers.all
2✔
1726

1727
    @classmethod
2✔
1728
    def compute_value(cls, raw_value: float | None, address: Address) -> float | None:
2✔
UNCOV
1729
        value_or_default = super().compute_value(raw_value, address)
×
UNCOV
1730
        cls.valid_numbers.validate(value_or_default, cls.alias, address)
×
UNCOV
1731
        return value_or_default
×
1732

1733

1734
class TupleSequenceField(Generic[T], Field):
2✔
1735
    # this cannot be a SequenceField as compute_value's use of ensure_list
1736
    # does not work with expected_element_type=tuple when the value itself
1737
    # is already a tuple.
1738
    expected_element_type: ClassVar[type]
2✔
1739
    expected_element_count: ClassVar[int]  # -1 for unlimited
2✔
1740
    expected_type_description: ClassVar[str]
2✔
1741
    expected_element_type_description: ClassVar[str]
2✔
1742

1743
    value: tuple[tuple[T, ...], ...] | None
2✔
1744
    default: ClassVar[tuple[tuple[T, ...], ...] | None] = None
2✔
1745

1746
    @classmethod
2✔
1747
    def compute_value(
2✔
1748
        cls, raw_value: Iterable[Iterable[T]] | None, address: Address
1749
    ) -> tuple[tuple[T, ...], ...] | None:
1750
        value_or_default = super().compute_value(raw_value, address)
2✔
1751
        if value_or_default is None:
2✔
UNCOV
1752
            return value_or_default
×
1753
        if isinstance(value_or_default, str) or not isinstance(
2✔
1754
            value_or_default, collections.abc.Iterable
1755
        ):
UNCOV
1756
            raise InvalidFieldTypeException(
×
1757
                address,
1758
                cls.alias,
1759
                raw_value,
1760
                expected_type=cls.expected_type_description,
1761
            )
1762

1763
        def invalid_member_exception(
2✔
1764
            at_index: int, wrong_element: Any
1765
        ) -> InvalidFieldMemberTypeException:
UNCOV
1766
            return InvalidFieldMemberTypeException(
×
1767
                address,
1768
                cls.alias,
1769
                raw_value,
1770
                expected_type=cls.expected_element_type_description,
1771
                wrong_element=wrong_element,
1772
                at_index=at_index,
1773
            )
1774

1775
        validated: list[tuple[T, ...]] = []
2✔
1776
        for i, x in enumerate(value_or_default):
2✔
1777
            if isinstance(x, str) or not isinstance(x, collections.abc.Iterable):
2✔
UNCOV
1778
                raise invalid_member_exception(i, x)
×
1779
            element = tuple(x)
2✔
1780
            if cls.expected_element_count >= 0 and cls.expected_element_count != len(element):
2✔
1781
                raise invalid_member_exception(i, x)
×
1782
            for s in element:
2✔
1783
                if not isinstance(s, cls.expected_element_type):
2✔
UNCOV
1784
                    raise invalid_member_exception(i, x)
×
1785
            validated.append(cast(tuple[T, ...], element))
2✔
1786

1787
        return tuple(validated)
2✔
1788

1789

1790
class DictStringToStringField(Field):
2✔
1791
    value: FrozenDict[str, str] | None
2✔
1792
    default: ClassVar[FrozenDict[str, str] | None] = None
2✔
1793

1794
    @classmethod
2✔
1795
    def compute_value(
2✔
1796
        cls, raw_value: dict[str, str] | None, address: Address
1797
    ) -> FrozenDict[str, str] | None:
1798
        value_or_default = super().compute_value(raw_value, address)
2✔
1799
        if value_or_default is None:
2✔
1800
            return None
2✔
1801
        invalid_type_exception = InvalidFieldTypeException(
2✔
1802
            address, cls.alias, raw_value, expected_type="a dictionary of string -> string"
1803
        )
1804
        if not isinstance(value_or_default, collections.abc.Mapping):
2✔
UNCOV
1805
            raise invalid_type_exception
×
1806
        if not all(isinstance(k, str) and isinstance(v, str) for k, v in value_or_default.items()):
2✔
UNCOV
1807
            raise invalid_type_exception
×
1808
        return FrozenDict(value_or_default)
2✔
1809

1810

1811
class ListOfDictStringToStringField(Field):
2✔
1812
    value: tuple[FrozenDict[str, str]] | None
2✔
1813
    default: ClassVar[list[FrozenDict[str, str]] | None] = None
2✔
1814

1815
    @classmethod
2✔
1816
    def compute_value(
2✔
1817
        cls, raw_value: list[dict[str, str]] | None, address: Address
1818
    ) -> tuple[FrozenDict[str, str], ...] | None:
1819
        value_or_default = super().compute_value(raw_value, address)
2✔
1820
        if value_or_default is None:
2✔
1821
            return None
2✔
UNCOV
1822
        invalid_type_exception = InvalidFieldTypeException(
×
1823
            address,
1824
            cls.alias,
1825
            raw_value,
1826
            expected_type="a list of dictionaries (or a single dictionary) of string -> string",
1827
        )
1828

1829
        # Also support passing in a single dictionary by wrapping it
UNCOV
1830
        if not isinstance(value_or_default, (list, tuple)):
×
UNCOV
1831
            value_or_default = [value_or_default]
×
1832

UNCOV
1833
        result_lst: list[FrozenDict[str, str]] = []
×
UNCOV
1834
        for item in value_or_default:
×
UNCOV
1835
            if not isinstance(item, collections.abc.Mapping):
×
UNCOV
1836
                raise invalid_type_exception
×
UNCOV
1837
            if not all(isinstance(k, str) and isinstance(v, str) for k, v in item.items()):
×
UNCOV
1838
                raise invalid_type_exception
×
UNCOV
1839
            result_lst.append(FrozenDict(item))
×
1840

UNCOV
1841
        return tuple(result_lst)
×
1842

1843

1844
class NestedDictStringToStringField(Field):
2✔
1845
    value: FrozenDict[str, FrozenDict[str, str]] | None
2✔
1846
    default: ClassVar[FrozenDict[str, FrozenDict[str, str]] | None] = None
2✔
1847

1848
    @classmethod
2✔
1849
    def compute_value(
2✔
1850
        cls, raw_value: dict[str, dict[str, str]] | None, address: Address
1851
    ) -> FrozenDict[str, FrozenDict[str, str]] | None:
UNCOV
1852
        value_or_default = super().compute_value(raw_value, address)
×
UNCOV
1853
        if value_or_default is None:
×
UNCOV
1854
            return None
×
UNCOV
1855
        invalid_type_exception = InvalidFieldTypeException(
×
1856
            address,
1857
            cls.alias,
1858
            raw_value,
1859
            expected_type="dict[str, dict[str, str]]",
1860
        )
UNCOV
1861
        if not isinstance(value_or_default, collections.abc.Mapping):
×
UNCOV
1862
            raise invalid_type_exception
×
UNCOV
1863
        for key, nested_value in value_or_default.items():
×
UNCOV
1864
            if not isinstance(key, str) or not isinstance(nested_value, collections.abc.Mapping):
×
UNCOV
1865
                raise invalid_type_exception
×
UNCOV
1866
            if not all(isinstance(k, str) and isinstance(v, str) for k, v in nested_value.items()):
×
1867
                raise invalid_type_exception
×
UNCOV
1868
        return FrozenDict(
×
1869
            {key: FrozenDict(nested_value) for key, nested_value in value_or_default.items()}
1870
        )
1871

1872

1873
class DictStringToStringSequenceField(Field):
2✔
1874
    value: FrozenDict[str, tuple[str, ...]] | None
2✔
1875
    default: ClassVar[FrozenDict[str, tuple[str, ...]] | None] = None
2✔
1876

1877
    @classmethod
2✔
1878
    def compute_value(
2✔
1879
        cls, raw_value: dict[str, Iterable[str]] | None, address: Address
1880
    ) -> FrozenDict[str, tuple[str, ...]] | None:
1881
        value_or_default = super().compute_value(raw_value, address)
2✔
1882
        if value_or_default is None:
2✔
1883
            return None
2✔
1884
        invalid_type_exception = InvalidFieldTypeException(
2✔
1885
            address,
1886
            cls.alias,
1887
            raw_value,
1888
            expected_type="a dictionary of string -> an iterable of strings",
1889
        )
1890
        if not isinstance(value_or_default, collections.abc.Mapping):
2✔
UNCOV
1891
            raise invalid_type_exception
×
1892
        result = {}
2✔
1893
        for k, v in value_or_default.items():
2✔
1894
            if not isinstance(k, str):
2✔
UNCOV
1895
                raise invalid_type_exception
×
1896
            try:
2✔
1897
                result[k] = tuple(ensure_str_list(v))
2✔
UNCOV
1898
            except ValueError:
×
UNCOV
1899
                raise invalid_type_exception
×
1900
        return FrozenDict(result)
2✔
1901

1902

1903
# -----------------------------------------------------------------------------------------------
1904
# Sources and codegen
1905
# -----------------------------------------------------------------------------------------------
1906

1907

1908
class SourcesField(AsyncFieldMixin, Field):
2✔
1909
    """A field for the sources that a target owns.
1910

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

1916
    Subclasses may set the following class properties:
1917

1918
    - `expected_file_extensions` -- A tuple of strings containing the expected file extensions for
1919
        source files. The default is no expected file extensions.
1920
    - `expected_num_files` -- An integer or range stating the expected total number of source
1921
        files. The default is no limit on the number of source files.
1922
    - `uses_source_roots` -- Whether the concept of "source root" pertains to the source files
1923
        referenced by this field.
1924
    - `default` -- A default value for this field.
1925
    - `default_glob_match_error_behavior` -- Advanced option, should very rarely be used. Override
1926
        glob match error behavior when using the default value. If setting this to
1927
        `GlobMatchErrorBehavior.ignore`, make sure you have other validation in place in case the
1928
        default glob doesn't match any files, if required, to alert the user appropriately.
1929
    """
1930

1931
    expected_file_extensions: ClassVar[tuple[str, ...] | None] = None
2✔
1932
    expected_num_files: ClassVar[int | range | None] = None
2✔
1933
    uses_source_roots: ClassVar[bool] = True
2✔
1934

1935
    default: ClassVar[ImmutableValue] = None
2✔
1936
    default_glob_match_error_behavior: ClassVar[GlobMatchErrorBehavior | None] = None
2✔
1937

1938
    @property
2✔
1939
    def globs(self) -> tuple[str, ...]:
2✔
1940
        """The raw globs, relative to the BUILD file."""
1941

1942
        # NB: We give a default implementation because it's common to use
1943
        # `tgt.get(SourcesField)`, and that must not error. But, subclasses need to
1944
        # implement this for the field to be useful (they should subclass `MultipleSourcesField`
1945
        # and `SingleSourceField`).
1946
        return ()
2✔
1947

1948
    def validate_resolved_files(self, files: Sequence[str]) -> None:
2✔
1949
        """Perform any additional validation on the resulting source files, e.g. ensuring that
1950
        certain banned files are not used.
1951

1952
        To enforce that the resulting files end in certain extensions, such as `.py` or `.java`, set
1953
        the class property `expected_file_extensions`.
1954

1955
        To enforce that there are only a certain number of resulting files, such as binary targets
1956
        checking for only 0-1 sources, set the class property `expected_num_files`.
1957
        """
1958
        if self.expected_file_extensions is not None:
2✔
1959
            bad_files = [
2✔
1960
                fp for fp in files if PurePath(fp).suffix not in self.expected_file_extensions
1961
            ]
1962
            if bad_files:
2✔
UNCOV
1963
                expected = (
×
1964
                    f"one of {sorted(self.expected_file_extensions)}"
1965
                    if len(self.expected_file_extensions) > 1
1966
                    else repr(self.expected_file_extensions[0])
1967
                )
UNCOV
1968
                raise InvalidFieldException(
×
1969
                    f"The {repr(self.alias)} field in target {self.address} can only contain "
1970
                    f"files that end in {expected}, but it had these files: {sorted(bad_files)}."
1971
                    "\n\nMaybe create a `resource`/`resources` or `file`/`files` target and "
1972
                    "include it in the `dependencies` field?"
1973
                )
1974
        if self.expected_num_files is not None:
2✔
1975
            num_files = len(files)
2✔
1976
            is_bad_num_files = (
2✔
1977
                num_files not in self.expected_num_files
1978
                if isinstance(self.expected_num_files, range)
1979
                else num_files != self.expected_num_files
1980
            )
1981
            if is_bad_num_files:
2✔
UNCOV
1982
                if isinstance(self.expected_num_files, range):
×
UNCOV
1983
                    if len(self.expected_num_files) == 2:
×
UNCOV
1984
                        expected_str = (
×
1985
                            " or ".join(str(n) for n in self.expected_num_files) + " files"
1986
                        )
1987
                    else:
1988
                        expected_str = f"a number of files in the range `{self.expected_num_files}`"
×
1989
                else:
UNCOV
1990
                    expected_str = pluralize(self.expected_num_files, "file")
×
UNCOV
1991
                raise InvalidFieldException(
×
1992
                    f"The {repr(self.alias)} field in target {self.address} must have "
1993
                    f"{expected_str}, but it had {pluralize(num_files, 'file')}."
1994
                )
1995

1996
    @staticmethod
2✔
1997
    def prefix_glob_with_dirpath(dirpath: str, glob: str) -> str:
2✔
1998
        if glob.startswith("!"):
2✔
1999
            return f"!{os.path.join(dirpath, glob[1:])}"
2✔
2000
        return os.path.join(dirpath, glob)
2✔
2001

2002
    @final
2✔
2003
    def _prefix_glob_with_address(self, glob: str) -> str:
2✔
2004
        return self.prefix_glob_with_dirpath(self.address.spec_path, glob)
2✔
2005

2006
    @final
2✔
2007
    @classmethod
2✔
2008
    def can_generate(
2✔
2009
        cls, output_type: type[SourcesField], union_membership: UnionMembership
2010
    ) -> bool:
2011
        """Can this field be used to generate the output_type?
2012

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

2017
            await hydrate_sources(
2018
                HydrateSourcesRequest(
2019
                    sources_field,
2020
                    for_sources_types=[FortranSources],
2021
                    enable_codegen=True,
2022
                ),
2023
                **implicitly(),
2024
            )
2025

2026
        This method is useful when you need to filter targets before hydrating them, such as how
2027
        you may filter targets via `tgt.has_field(MyField)`.
2028
        """
2029
        generate_request_types = union_membership.get(GenerateSourcesRequest)
2✔
2030
        return any(
2✔
2031
            issubclass(cls, generate_request_type.input)
2032
            and issubclass(generate_request_type.output, output_type)
2033
            for generate_request_type in generate_request_types
2034
        )
2035

2036
    @final
2✔
2037
    def path_globs(self, unmatched_build_file_globs: UnmatchedBuildFileGlobs) -> PathGlobs:
2✔
2038
        if not self.globs:
2✔
2039
            return PathGlobs([])
2✔
2040

2041
        # SingleSourceField has str as default type.
2042
        default_globs = (
2✔
2043
            [self.default] if self.default and isinstance(self.default, str) else self.default
2044
        )
2045

2046
        using_default_globs = default_globs and (set(self.globs) == set(default_globs)) or False
2✔
2047

2048
        # Use fields default error behavior if defined, if we use default globs else the provided
2049
        # error behavior.
2050
        error_behavior = (
2✔
2051
            unmatched_build_file_globs.error_behavior
2052
            if not using_default_globs or self.default_glob_match_error_behavior is None
2053
            else self.default_glob_match_error_behavior
2054
        )
2055

2056
        return PathGlobs(
2✔
2057
            (self._prefix_glob_with_address(glob) for glob in self.globs),
2058
            conjunction=GlobExpansionConjunction.any_match,
2059
            glob_match_error_behavior=error_behavior,
2060
            description_of_origin=(
2061
                f"{self.address}'s `{self.alias}` field"
2062
                if error_behavior != GlobMatchErrorBehavior.ignore
2063
                else None
2064
            ),
2065
        )
2066

2067
    @memoized_property
2✔
2068
    def filespec(self) -> Filespec:
2✔
2069
        """The original globs, returned in the Filespec dict format.
2070

2071
        The globs will be relativized to the build root.
2072
        """
2073
        includes = []
2✔
2074
        excludes = []
2✔
2075
        for glob in self.globs:
2✔
2076
            if glob.startswith("!"):
2✔
2077
                excludes.append(os.path.join(self.address.spec_path, glob[1:]))
2✔
2078
            else:
2079
                includes.append(os.path.join(self.address.spec_path, glob))
2✔
2080
        result: Filespec = {"includes": includes}
2✔
2081
        if excludes:
2✔
2082
            result["excludes"] = excludes
2✔
2083
        return result
2✔
2084

2085
    @memoized_property
2✔
2086
    def filespec_matcher(self) -> FilespecMatcher:
2✔
2087
        # Note: memoized because parsing the globs is expensive:
2088
        # https://github.com/pantsbuild/pants/issues/16122
2089
        return FilespecMatcher(self.filespec["includes"], self.filespec.get("excludes", []))
2✔
2090

2091

2092
class MultipleSourcesField(SourcesField, StringSequenceField):
2✔
2093
    """The `sources: list[str]` field.
2094

2095
    See the docstring for `SourcesField` for some class properties you can set, such as
2096
    `expected_file_extensions`.
2097

2098
    When you need to get the sources for all targets, use `tgt.get(SourcesField)` rather than
2099
    `tgt.get(MultipleSourcesField)`.
2100
    """
2101

2102
    alias = "sources"
2✔
2103

2104
    ban_subdirectories: ClassVar[bool] = False
2✔
2105

2106
    @property
2✔
2107
    def globs(self) -> tuple[str, ...]:
2✔
2108
        return self.value or ()
2✔
2109

2110
    @classmethod
2✔
2111
    def compute_value(
2✔
2112
        cls, raw_value: Iterable[str] | None, address: Address
2113
    ) -> tuple[str, ...] | None:
2114
        value = super().compute_value(raw_value, address)
2✔
2115
        invalid_globs = [glob for glob in (value or ()) if glob.startswith("../") or "/../" in glob]
2✔
2116
        if invalid_globs:
2✔
UNCOV
2117
            raise InvalidFieldException(
×
2118
                softwrap(
2119
                    f"""
2120
                    The {repr(cls.alias)} field in target {address} must not have globs with the
2121
                    pattern `../` because targets can only have sources in the current directory
2122
                    or subdirectories. It was set to: {sorted(value or ())}
2123
                    """
2124
                )
2125
            )
2126
        if cls.ban_subdirectories:
2✔
2127
            invalid_globs = [glob for glob in (value or ()) if "**" in glob or os.path.sep in glob]
2✔
2128
            if invalid_globs:
2✔
UNCOV
2129
                raise InvalidFieldException(
×
2130
                    softwrap(
2131
                        f"""
2132
                        The {repr(cls.alias)} field in target {address} must only have globs for
2133
                        the target's directory, i.e. it cannot include values with `**` or
2134
                        `{os.path.sep}`. It was set to: {sorted(value or ())}
2135
                        """
2136
                    )
2137
                )
2138
        return value
2✔
2139

2140

2141
class OptionalSingleSourceField(SourcesField, StringField):
2✔
2142
    """The `source: str` field.
2143

2144
    See the docstring for `SourcesField` for some class properties you can set, such as
2145
    `expected_file_extensions`.
2146

2147
    When you need to get the sources for all targets, use `tgt.get(SourcesField)` rather than
2148
    `tgt.get(OptionalSingleSourceField)`.
2149

2150
    Use `SingleSourceField` if the source must exist.
2151
    """
2152

2153
    alias = "source"
2✔
2154
    help = help_text(
2✔
2155
        """
2156
        A single file that belongs to this target.
2157

2158
        Path is relative to the BUILD file's directory, e.g. `source='example.ext'`.
2159
        """
2160
    )
2161
    required = False
2✔
2162
    default: ClassVar[str | None] = None
2✔
2163
    expected_num_files: ClassVar[int | range] = range(0, 2)
2✔
2164

2165
    @classmethod
2✔
2166
    def compute_value(cls, raw_value: str | None, address: Address) -> str | None:
2✔
2167
        value_or_default = super().compute_value(raw_value, address)
2✔
2168
        if value_or_default is None:
2✔
2169
            return None
2✔
2170
        if value_or_default.startswith("../") or "/../" in value_or_default:
2✔
UNCOV
2171
            raise InvalidFieldException(
×
2172
                softwrap(
2173
                    f"""\
2174
                    The {repr(cls.alias)} field in target {address} should not include `../`
2175
                    patterns because targets can only have sources in the current directory or
2176
                    subdirectories. It was set to {value_or_default}. Instead, use a normalized
2177
                    literal file path (relative to the BUILD file).
2178
                    """
2179
                )
2180
            )
2181
        if "*" in value_or_default:
2✔
UNCOV
2182
            raise InvalidFieldException(
×
2183
                softwrap(
2184
                    f"""\
2185
                    The {repr(cls.alias)} field in target {address} should not include `*` globs,
2186
                    but was set to {value_or_default}. Instead, use a literal file path (relative
2187
                    to the BUILD file).
2188
                    """
2189
                )
2190
            )
2191
        if value_or_default.startswith("!"):
2✔
UNCOV
2192
            raise InvalidFieldException(
×
2193
                softwrap(
2194
                    f"""\
2195
                    The {repr(cls.alias)} field in target {address} should not start with `!`,
2196
                    which is usually used in the `sources` field to exclude certain files. Instead,
2197
                    use a literal file path (relative to the BUILD file).
2198
                    """
2199
                )
2200
            )
2201
        return value_or_default
2✔
2202

2203
    @property
2✔
2204
    def file_path(self) -> str | None:
2✔
2205
        """The path to the file, relative to the build root.
2206

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

2210
        The return type is optional because it's possible to have 0-1 files.
2211
        """
2212
        if self.value is None:
2✔
2213
            return None
2✔
2214
        return os.path.join(self.address.spec_path, self.value)
2✔
2215

2216
    @property
2✔
2217
    def globs(self) -> tuple[str, ...]:
2✔
2218
        if self.value is None:
2✔
UNCOV
2219
            return ()
×
2220
        return (self.value,)
2✔
2221

2222

2223
class SingleSourceField(OptionalSingleSourceField):
2✔
2224
    """The `source: str` field.
2225

2226
    Unlike `OptionalSingleSourceField`, the `.value` must be defined, whether by setting the
2227
    `default` or making the field `required`.
2228

2229
    See the docstring for `SourcesField` for some class properties you can set, such as
2230
    `expected_file_extensions`.
2231

2232
    When you need to get the sources for all targets, use `tgt.get(SourcesField)` rather than
2233
    `tgt.get(SingleSourceField)`.
2234
    """
2235

2236
    required = True
2✔
2237
    expected_num_files = 1
2✔
2238
    value: str
2✔
2239

2240
    @property
2✔
2241
    def file_path(self) -> str:
2✔
2242
        result = super().file_path
2✔
2243
        assert result is not None
2✔
2244
        return result
2✔
2245

2246

2247
@dataclass(frozen=True)
2✔
2248
class HydrateSourcesRequest(EngineAwareParameter):
2✔
2249
    field: SourcesField
2✔
2250
    for_sources_types: tuple[type[SourcesField], ...]
2✔
2251
    enable_codegen: bool
2✔
2252

2253
    def __init__(
2✔
2254
        self,
2255
        field: SourcesField,
2256
        *,
2257
        for_sources_types: Iterable[type[SourcesField]] = (SourcesField,),
2258
        enable_codegen: bool = False,
2259
    ) -> None:
2260
        """Convert raw sources globs into an instance of HydratedSources.
2261

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

2266
        If `enable_codegen` is set to `True`, any codegen sources will try to be converted to one
2267
        of the `for_sources_types`.
2268
        """
2269
        object.__setattr__(self, "field", field)
2✔
2270
        object.__setattr__(self, "for_sources_types", tuple(for_sources_types))
2✔
2271
        object.__setattr__(self, "enable_codegen", enable_codegen)
2✔
2272

2273
        self.__post_init__()
2✔
2274

2275
    def __post_init__(self) -> None:
2✔
2276
        if self.enable_codegen and self.for_sources_types == (SourcesField,):
2✔
2277
            raise ValueError(
×
2278
                "When setting `enable_codegen=True` on `HydrateSourcesRequest`, you must also "
2279
                "explicitly set `for_source_types`. Why? `for_source_types` is used to "
2280
                "determine which language(s) to try to generate. For example, "
2281
                "`for_source_types=(PythonSources,)` will hydrate `PythonSources` like normal, "
2282
                "and, if it encounters codegen sources that can be converted into Python, it will "
2283
                "generate Python files."
2284
            )
2285

2286
    def debug_hint(self) -> str:
2✔
2287
        return self.field.address.spec
2✔
2288

2289

2290
@dataclass(frozen=True)
2✔
2291
class HydratedSources:
2✔
2292
    """The result of hydrating a SourcesField.
2293

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

2302
    snapshot: Snapshot
2✔
2303
    filespec: Filespec
2✔
2304
    sources_type: type[SourcesField] | None
2✔
2305

2306

2307
@union(in_scope_types=[EnvironmentName])
2✔
2308
@dataclass(frozen=True)
2✔
2309
class GenerateSourcesRequest:
2✔
2310
    """A request to go from protocol sources -> a particular language.
2311

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

2316
    The rule to actually implement the codegen should take the subclass as input, and it must
2317
    return `GeneratedSources`.
2318

2319
    The `exportable` attribute disables the use of this codegen by the `export-codegen` goal when
2320
    set to False.
2321

2322
    For example:
2323

2324
        class GenerateFortranFromAvroRequest:
2325
            input = AvroSources
2326
            output = FortranSources
2327

2328
        @rule
2329
        async def generate_fortran_from_avro(request: GenerateFortranFromAvroRequest) -> GeneratedSources:
2330
            ...
2331

2332
        def rules():
2333
            return [
2334
                generate_fortran_from_avro,
2335
                UnionRule(GenerateSourcesRequest, GenerateFortranFromAvroRequest),
2336
            ]
2337
    """
2338

2339
    protocol_sources: Snapshot
2✔
2340
    protocol_target: Target
2✔
2341

2342
    input: ClassVar[type[SourcesField]]
2✔
2343
    output: ClassVar[type[SourcesField]]
2✔
2344

2345
    exportable: ClassVar[bool] = True
2✔
2346

2347

2348
@dataclass(frozen=True)
2✔
2349
class GeneratedSources:
2✔
2350
    snapshot: Snapshot
2✔
2351

2352

2353
@rule(polymorphic=True)
2✔
2354
async def generate_sources(
2✔
2355
    req: GenerateSourcesRequest, env_name: EnvironmentName
2356
) -> GeneratedSources:
2357
    raise NotImplementedError()
×
2358

2359

2360
class SourcesPaths(Paths):
2✔
2361
    """The resolved file names of the `source`/`sources` field.
2362

2363
    This does not consider codegen, and only captures the files from the field.
2364
    """
2365

2366

2367
@dataclass(frozen=True)
2✔
2368
class SourcesPathsRequest(EngineAwareParameter):
2✔
2369
    """A request to resolve the file names of the `source`/`sources` field.
2370

2371
    Use via `await resolve_source_paths(SourcesPathRequest(tgt.get(SourcesField))`.
2372

2373
    This is faster than `await hydrate_sources(HydrateSourcesRequest)` because it does not snapshot
2374
    the files and it only resolves the file names.
2375

2376
    This does not consider codegen, and only captures the files from the field. Use
2377
    `HydrateSourcesRequest` to use codegen.
2378
    """
2379

2380
    field: SourcesField
2✔
2381

2382
    def debug_hint(self) -> str:
2✔
UNCOV
2383
        return self.field.address.spec
×
2384

2385

2386
def targets_with_sources_types(
2✔
2387
    sources_types: Iterable[type[SourcesField]],
2388
    targets: Iterable[Target],
2389
    union_membership: UnionMembership,
2390
) -> tuple[Target, ...]:
2391
    """Return all targets either with the specified sources subclass(es) or which can generate those
2392
    sources."""
UNCOV
2393
    return tuple(
×
2394
        tgt
2395
        for tgt in targets
2396
        if any(
2397
            tgt.has_field(sources_type)
2398
            or tgt.get(SourcesField).can_generate(sources_type, union_membership)
2399
            for sources_type in sources_types
2400
        )
2401
    )
2402

2403

2404
# -----------------------------------------------------------------------------------------------
2405
# `Dependencies` field
2406
# -----------------------------------------------------------------------------------------------
2407

2408

2409
class Dependencies(StringSequenceField, AsyncFieldMixin):
2✔
2410
    """The dependencies field.
2411

2412
    To resolve all dependencies—including the results of dependency inference—use either
2413
    `await resolve_dependencies(DependenciesRequest(tgt[Dependencies])` or
2414
    `await resolve_targets(**implicitly(DependenciesRequest(tgt[Dependencies]))`.
2415
    """
2416

2417
    alias = "dependencies"
2✔
2418
    help = help_text(
2✔
2419
        f"""
2420
        Addresses to other targets that this target depends on, e.g.
2421
        `['helloworld/subdir:lib', 'helloworld/main.py:lib', '3rdparty:reqs#django']`.
2422

2423
        This augments any dependencies inferred by Pants, such as by analyzing your imports. Use
2424
        `{bin_name()} dependencies` or `{bin_name()} peek` on this target to get the final
2425
        result.
2426

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

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

2436
        You may exclude dependencies by prefixing with `!`, e.g.
2437
        `['!helloworld/subdir:lib', '!./sibling.txt']`. Ignores are intended for false positives
2438
        with dependency inference; otherwise, simply leave off the dependency from the BUILD file.
2439
        """
2440
    )
2441
    supports_transitive_excludes = False
2✔
2442

2443
    @memoized_property
2✔
2444
    def unevaluated_transitive_excludes(self) -> UnparsedAddressInputs:
2✔
2445
        val = (
2✔
2446
            (v[2:] for v in self.value if v.startswith("!!"))
2447
            if self.supports_transitive_excludes and self.value
2448
            else ()
2449
        )
2450
        return UnparsedAddressInputs(
2✔
2451
            val,
2452
            owning_address=self.address,
2453
            description_of_origin=f"the `{self.alias}` field from the target {self.address}",
2454
        )
2455

2456

2457
@dataclass(frozen=True)
2✔
2458
class DependenciesRequest(EngineAwareParameter):
2✔
2459
    field: Dependencies
2✔
2460
    should_traverse_deps_predicate: ShouldTraverseDepsPredicate = TraverseIfDependenciesField()
2✔
2461

2462
    def debug_hint(self) -> str:
2✔
2463
        return self.field.address.spec
2✔
2464

2465

2466
# NB: ExplicitlyProvidedDependenciesRequest does not have a predicate unlike DependenciesRequest.
2467
@dataclass(frozen=True)
2✔
2468
class ExplicitlyProvidedDependenciesRequest(EngineAwareParameter):
2✔
2469
    field: Dependencies
2✔
2470

2471
    def debug_hint(self) -> str:
2✔
2472
        return self.field.address.spec
×
2473

2474

2475
@dataclass(frozen=True)
2✔
2476
class ExplicitlyProvidedDependencies:
2✔
2477
    """The literal addresses from a BUILD file `dependencies` field.
2478

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

2484
    Resolve using
2485
    `await determine_explicitly_provided_dependencies(**implicitly(DependenciesRequest))`.
2486

2487
    Note that the `includes` are not filtered based on the `ignores`: this type preserves exactly
2488
    what was in the BUILD file.
2489
    """
2490

2491
    address: Address
2✔
2492
    includes: FrozenOrderedSet[Address]
2✔
2493
    ignores: FrozenOrderedSet[Address]
2✔
2494

2495
    @memoized_method
2✔
2496
    def any_are_covered_by_includes(self, addresses: Iterable[Address]) -> bool:
2✔
2497
        """Return True if every address is in the explicitly provided includes.
2498

2499
        Note that if the input addresses are generated targets, they will still be marked as covered
2500
        if their original target generator is in the explicitly provided includes.
2501
        """
UNCOV
2502
        return any(
×
2503
            addr in self.includes or addr.maybe_convert_to_target_generator() in self.includes
2504
            for addr in addresses
2505
        )
2506

2507
    @memoized_method
2✔
2508
    def remaining_after_disambiguation(
2✔
2509
        self, addresses: Iterable[Address], owners_must_be_ancestors: bool
2510
    ) -> frozenset[Address]:
2511
        """All addresses that remain after ineligible candidates are discarded.
2512

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

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

UNCOV
2522
        def is_valid(addr: Address) -> bool:
×
UNCOV
2523
            is_ignored = (
×
2524
                addr in self.ignores or addr.maybe_convert_to_target_generator() in self.ignores
2525
            )
UNCOV
2526
            if owners_must_be_ancestors is False:
×
UNCOV
2527
                return not is_ignored
×
2528
            # NB: `PurePath.is_relative_to()` was not added until Python 3.9. This emulates it.
UNCOV
2529
            try:
×
UNCOV
2530
                original_addr_path.relative_to(addr.spec_path)
×
UNCOV
2531
                return not is_ignored
×
UNCOV
2532
            except ValueError:
×
UNCOV
2533
                return False
×
2534

UNCOV
2535
        return frozenset(filter(is_valid, addresses))
×
2536

2537
    def maybe_warn_of_ambiguous_dependency_inference(
2✔
2538
        self,
2539
        ambiguous_addresses: Iterable[Address],
2540
        original_address: Address,
2541
        *,
2542
        context: str,
2543
        import_reference: str,
2544
        owners_must_be_ancestors: bool = False,
2545
    ) -> None:
2546
        """If the module is ambiguous and the user did not disambiguate, warn that dependency
2547
        inference will not be used.
2548

2549
        Disambiguation usually happens by using ignores in the `dependencies` field with `!` and
2550
        `!!`. If `owners_must_be_ancestors` is True, any addresses which are not ancestors of the
2551
        target in question will also be ignored.
2552
        """
2553
        if not ambiguous_addresses or self.any_are_covered_by_includes(ambiguous_addresses):
2✔
2554
            return
2✔
UNCOV
2555
        remaining = self.remaining_after_disambiguation(
×
2556
            ambiguous_addresses, owners_must_be_ancestors=owners_must_be_ancestors
2557
        )
UNCOV
2558
        if len(remaining) <= 1:
×
UNCOV
2559
            return
×
UNCOV
2560
        logger.warning(
×
2561
            f"{context}, but Pants cannot safely infer a dependency because more than one target "
2562
            f"owns this {import_reference}, so it is ambiguous which to use: "
2563
            f"{sorted(addr.spec for addr in remaining)}."
2564
            f"\n\nPlease explicitly include the dependency you want in the `dependencies` "
2565
            f"field of {original_address}, or ignore the ones you do not want by prefixing "
2566
            f"with `!` or `!!` so that one or no targets are left."
2567
            f"\n\nAlternatively, you can remove the ambiguity by deleting/changing some of the "
2568
            f"targets so that only 1 target owns this {import_reference}. Refer to "
2569
            f"{doc_url('docs/using-pants/troubleshooting-common-issues#import-errors-and-missing-dependencies')}."
2570
        )
2571

2572
    def disambiguated(
2✔
2573
        self, ambiguous_addresses: Iterable[Address], owners_must_be_ancestors: bool = False
2574
    ) -> Address | None:
2575
        """If exactly one of the input addresses remains after disambiguation, return it.
2576

2577
        Disambiguation usually happens by using ignores in the `dependencies` field with `!` and
2578
        `!!`. If `owners_must_be_ancestors` is True, any addresses which are not ancestors of the
2579
        target in question will also be ignored.
2580
        """
2581
        if not ambiguous_addresses or self.any_are_covered_by_includes(ambiguous_addresses):
2✔
2582
            return None
2✔
UNCOV
2583
        remaining_after_ignores = self.remaining_after_disambiguation(
×
2584
            ambiguous_addresses, owners_must_be_ancestors=owners_must_be_ancestors
2585
        )
UNCOV
2586
        return list(remaining_after_ignores)[0] if len(remaining_after_ignores) == 1 else None
×
2587

2588

2589
FS = TypeVar("FS", bound="FieldSet")
2✔
2590

2591

2592
@union(in_scope_types=[EnvironmentName])
2✔
2593
@dataclass(frozen=True)
2✔
2594
class InferDependenciesRequest(Generic[FS], EngineAwareParameter):
2✔
2595
    """A request to infer dependencies by analyzing source files.
2596

2597
    To set up a new inference implementation, subclass this class. Set the class property
2598
    `infer_from` to the FieldSet subclass you are able to infer from. This will cause the FieldSet
2599
    class, and any subclass, to use your inference implementation.
2600

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

2603
    Register this subclass with `UnionRule(InferDependenciesRequest, InferFortranDependencies)`, for example.
2604

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

2607
    For example:
2608

2609
        class InferFortranDependencies(InferDependenciesRequest):
2610
            infer_from = FortranDependenciesInferenceFieldSet
2611

2612
        @rule
2613
        async def infer_fortran_dependencies(request: InferFortranDependencies) -> InferredDependencies:
2614
            hydrated_sources = await hydrate_sources(HydrateSources(request.field_set.sources))
2615
            ...
2616
            return InferredDependencies(...)
2617

2618
        def rules():
2619
            return [
2620
                infer_fortran_dependencies,
2621
                UnionRule(InferDependenciesRequest, InferFortranDependencies),
2622
            ]
2623
    """
2624

2625
    infer_from: ClassVar[type[FS]]
2✔
2626

2627
    field_set: FS
2✔
2628

2629

2630
@dataclass(frozen=True)
2✔
2631
class InferredDependencies:
2✔
2632
    include: FrozenOrderedSet[Address]
2✔
2633
    exclude: FrozenOrderedSet[Address]
2✔
2634

2635
    def __init__(
2✔
2636
        self,
2637
        include: Iterable[Address],
2638
        *,
2639
        exclude: Iterable[Address] = (),
2640
    ) -> None:
2641
        """The result of inferring dependencies."""
2642
        object.__setattr__(self, "include", FrozenOrderedSet(sorted(include)))
2✔
2643
        object.__setattr__(self, "exclude", FrozenOrderedSet(sorted(exclude)))
2✔
2644

2645

2646
@union(in_scope_types=[EnvironmentName])
2✔
2647
@dataclass(frozen=True)
2✔
2648
class TransitivelyExcludeDependenciesRequest(Generic[FS], EngineAwareParameter):
2✔
2649
    """A request to transitvely exclude dependencies of a "root" node.
2650

2651
    This is similar to `InferDependenciesRequest`, except the request is only made for "root" nodes
2652
    in the dependency graph.
2653

2654
    This mirrors the public facing "transitive exclude" dependency feature (i.e. `!!<address>`).
2655
    """
2656

2657
    infer_from: ClassVar[type[FS]]
2✔
2658

2659
    field_set: FS
2✔
2660

2661

2662
class TransitivelyExcludeDependencies(FrozenOrderedSet[Address]):
2✔
2663
    pass
2✔
2664

2665

2666
@union(in_scope_types=[EnvironmentName])
2✔
2667
@dataclass(frozen=True)
2✔
2668
class ValidateDependenciesRequest(Generic[FS], ABC):
2✔
2669
    """A request to validate dependencies after they have been computed.
2670

2671
    An implementing rule should raise an exception if dependencies are invalid.
2672
    """
2673

2674
    field_set_type: ClassVar[type[FS]]
2✔
2675

2676
    field_set: FS
2✔
2677
    dependencies: Addresses
2✔
2678

2679

2680
@dataclass(frozen=True)
2✔
2681
class ValidatedDependencies:
2✔
2682
    pass
2✔
2683

2684

2685
@dataclass(frozen=True)
2✔
2686
class DependenciesRuleApplicationRequest:
2✔
2687
    """A request to return the applicable dependency rule action for each dependency of a target."""
2688

2689
    address: Address
2✔
2690
    dependencies: Addresses
2✔
2691
    description_of_origin: str = dataclasses.field(hash=False, compare=False)
2✔
2692

2693

2694
@dataclass(frozen=True)
2✔
2695
class DependenciesRuleApplication:
2✔
2696
    """Maps all dependencies to their respective dependency rule application of an origin target
2697
    address.
2698

2699
    The `applications` will be empty and the `address` `None` if there is no dependency rule
2700
    implementation.
2701
    """
2702

2703
    address: Address | None = None
2✔
2704
    dependencies_rule: FrozenDict[Address, DependencyRuleApplication] = FrozenDict()
2✔
2705

2706
    def __post_init__(self):
2✔
UNCOV
2707
        if self.dependencies_rule and self.address is None:
×
2708
            raise ValueError(
×
2709
                "The `address` field must not be None when there are `dependencies_rule`s."
2710
            )
2711

2712
    @classmethod
2✔
2713
    @memoized_method
2✔
2714
    def allow_all(cls) -> DependenciesRuleApplication:
2✔
2715
        return cls()
×
2716

2717
    def execute_actions(self) -> None:
2✔
UNCOV
2718
        errors = [
×
2719
            action_error.replace("\n", "\n    ")
2720
            for action_error in (rule.execute() for rule in self.dependencies_rule.values())
2721
            if action_error is not None
2722
        ]
UNCOV
2723
        if errors:
×
UNCOV
2724
            err_count = len(errors)
×
UNCOV
2725
            raise DependencyRuleActionDeniedError(
×
2726
                softwrap(
2727
                    f"""
2728
                    {self.address} has {pluralize(err_count, "dependency violation")}:
2729

2730
                    {bullet_list(errors)}
2731
                    """
2732
                )
2733
            )
2734

2735

2736
class SpecialCasedDependencies(StringSequenceField, AsyncFieldMixin):
2✔
2737
    """Subclass this for fields that act similarly to the `dependencies` field, but are handled
2738
    differently than normal dependencies.
2739

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

2745
    This type will ensure that the dependencies show up in project introspection,
2746
    like `dependencies` and `dependents`, but not show up when you
2747
    `await transitive_targets(TransitiveTargetsRequest(...), **implicitly())` and
2748
    `await resolve_dependencies(DependenciesRequest(...), **implicitly())`.
2749

2750
    To hydrate this field's dependencies, use
2751
    `await resolve_unparsed_address_inputs(tgt.get(MyField).to_unparsed_address_inputs(), **implicitly())`.
2752
    """
2753

2754
    def to_unparsed_address_inputs(self) -> UnparsedAddressInputs:
2✔
2755
        return UnparsedAddressInputs(
2✔
2756
            self.value or (),
2757
            owning_address=self.address,
2758
            description_of_origin=f"the `{self.alias}` from the target {self.address}",
2759
        )
2760

2761

2762
# -----------------------------------------------------------------------------------------------
2763
# Other common Fields used across most targets
2764
# -----------------------------------------------------------------------------------------------
2765

2766

2767
class Tags(StringSequenceField):
2✔
2768
    alias = "tags"
2✔
2769
    help = help_text(
2✔
2770
        f"""
2771
        Arbitrary strings to describe a target.
2772

2773
        For example, you may tag some test targets with 'integration_test' so that you could run
2774
        `{bin_name()} --tag='integration_test' test ::`  to only run on targets with that tag.
2775
        """
2776
    )
2777

2778

2779
class DescriptionField(StringField):
2✔
2780
    alias = "description"
2✔
2781
    help = help_text(
2✔
2782
        f"""
2783
        A human-readable description of the target.
2784

2785
        Use `{bin_name()} list --documented ::` to see all targets with descriptions.
2786
        """
2787
    )
2788

2789

2790
COMMON_TARGET_FIELDS = (Tags, DescriptionField)
2✔
2791

2792

2793
class OverridesField(AsyncFieldMixin, Field):
2✔
2794
    """A mapping of keys (e.g. target names, source globs) to field names with their overridden
2795
    values.
2796

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

2802
    alias = "overrides"
2✔
2803
    value: dict[tuple[str, ...], dict[str, Any]] | None
2✔
2804
    default: ClassVar[None] = None  # A default does not make sense for this field.
2✔
2805

2806
    @classmethod
2✔
2807
    def compute_value(
2✔
2808
        cls,
2809
        raw_value: dict[str | tuple[str, ...], dict[str, Any]] | None,
2810
        address: Address,
2811
    ) -> FrozenDict[tuple[str, ...], FrozenDict[str, ImmutableValue]] | None:
2812
        value_or_default = super().compute_value(raw_value, address)
2✔
2813
        if value_or_default is None:
2✔
2814
            return None
2✔
2815

2816
        def invalid_type_exception() -> InvalidFieldException:
2✔
UNCOV
2817
            return InvalidFieldTypeException(
×
2818
                address,
2819
                cls.alias,
2820
                raw_value,
2821
                expected_type="dict[str | tuple[str, ...], dict[str, Any]]",
2822
            )
2823

2824
        if not isinstance(value_or_default, collections.abc.Mapping):
2✔
UNCOV
2825
            raise invalid_type_exception()
×
2826

2827
        result: dict[tuple[str, ...], FrozenDict[str, ImmutableValue]] = {}
2✔
2828
        for outer_key, nested_value in value_or_default.items():
2✔
2829
            if isinstance(outer_key, str):
2✔
2830
                outer_key = (outer_key,)
2✔
2831
            if not isinstance(outer_key, collections.abc.Sequence) or not all(
2✔
2832
                isinstance(elem, str) for elem in outer_key
2833
            ):
UNCOV
2834
                raise invalid_type_exception()
×
2835
            if not isinstance(nested_value, collections.abc.Mapping):
2✔
UNCOV
2836
                raise invalid_type_exception()
×
2837
            if not all(isinstance(inner_key, str) for inner_key in nested_value):
2✔
UNCOV
2838
                raise invalid_type_exception()
×
2839
            result[tuple(outer_key)] = FrozenDict.deep_freeze(cast(Mapping[str, Any], nested_value))
2✔
2840

2841
        return FrozenDict(result)
2✔
2842

2843
    @classmethod
2✔
2844
    def to_path_globs(
2✔
2845
        cls,
2846
        address: Address,
2847
        overrides_keys: Iterable[str],
2848
        unmatched_build_file_globs: UnmatchedBuildFileGlobs,
2849
    ) -> tuple[PathGlobs, ...]:
2850
        """Create a `PathGlobs` for each key.
2851

2852
        This should only be used if the keys are file globs.
2853
        """
2854

2855
        def relativize_glob(glob: str) -> str:
2✔
UNCOV
2856
            return (
×
2857
                f"!{os.path.join(address.spec_path, glob[1:])}"
2858
                if glob.startswith("!")
2859
                else os.path.join(address.spec_path, glob)
2860
            )
2861

2862
        return tuple(
2✔
2863
            PathGlobs(
2864
                [relativize_glob(glob)],
2865
                glob_match_error_behavior=unmatched_build_file_globs.error_behavior,
2866
                description_of_origin=f"the `overrides` field for {address}",
2867
            )
2868
            for glob in overrides_keys
2869
        )
2870

2871
    def flatten(self) -> dict[str, dict[str, Any]]:
2✔
2872
        """Combine all overrides for every key into a single dictionary."""
2873
        result: dict[str, dict[str, Any]] = {}
2✔
2874
        for keys, override in (self.value or {}).items():
2✔
2875
            for key in keys:
2✔
2876
                for field, value in override.items():
2✔
2877
                    if key not in result:
2✔
2878
                        result[key] = {field: value}
2✔
2879
                        continue
2✔
2880
                    if field not in result[key]:
2✔
2881
                        result[key][field] = value
2✔
2882
                        continue
2✔
2883
                    raise InvalidFieldException(
×
2884
                        f"Conflicting overrides in the `{self.alias}` field of "
2885
                        f"`{self.address}` for the key `{key}` for "
2886
                        f"the field `{field}`. You cannot specify the same field name "
2887
                        "multiple times for the same key.\n\n"
2888
                        f"(One override sets the field to `{repr(result[key][field])}` "
2889
                        f"but another sets to `{repr(value)}`.)"
2890
                    )
2891
        return result
2✔
2892

2893
    @classmethod
2✔
2894
    def flatten_paths(
2✔
2895
        cls,
2896
        address: Address,
2897
        paths_and_overrides: Iterable[tuple[Paths, PathGlobs, dict[str, Any]]],
2898
    ) -> dict[str, dict[str, Any]]:
2899
        """Combine all overrides for each file into a single dictionary."""
2900
        result: dict[str, dict[str, Any]] = {}
2✔
2901
        for paths, globs, override in paths_and_overrides:
2✔
2902
            # NB: If some globs did not result in any Paths, we preserve them to ensure that
2903
            # unconsumed overrides trigger errors during generation.
UNCOV
2904
            for path in paths.files or globs.globs:
×
UNCOV
2905
                for field, value in override.items():
×
UNCOV
2906
                    if path not in result:
×
UNCOV
2907
                        result[path] = {field: value}
×
UNCOV
2908
                        continue
×
UNCOV
2909
                    if field not in result[path]:
×
UNCOV
2910
                        result[path][field] = value
×
UNCOV
2911
                        continue
×
UNCOV
2912
                    relpath = fast_relpath(path, address.spec_path)
×
UNCOV
2913
                    raise InvalidFieldException(
×
2914
                        f"Conflicting overrides for `{address}` for the relative path "
2915
                        f"`{relpath}` for the field `{field}`. You cannot specify the same field "
2916
                        f"name multiple times for the same path.\n\n"
2917
                        f"(One override sets the field to `{repr(result[path][field])}` "
2918
                        f"but another sets to `{repr(value)}`.)"
2919
                    )
2920
        return result
2✔
2921

2922

2923
def generate_multiple_sources_field_help_message(files_example: str) -> str:
2✔
2924
    return softwrap(
2✔
2925
        """
2926
        A list of files and globs that belong to this target.
2927

2928
        Paths are relative to the BUILD file's directory. You can ignore files/globs by
2929
        prefixing them with `!`.
2930

2931
        """
2932
        + files_example
2933
    )
2934

2935

2936
def generate_file_based_overrides_field_help_message(
2✔
2937
    generated_target_name: str, example: str
2938
) -> str:
2939
    example = textwrap.dedent(example.lstrip("\n"))  # noqa: PNT20
2✔
2940
    example = textwrap.indent(example, " " * 4)
2✔
2941
    return "\n".join(
2✔
2942
        [
2943
            softwrap(
2944
                f"""
2945
                Override the field values for generated `{generated_target_name}` targets.
2946

2947
                Expects a dictionary of relative file paths and globs to a dictionary for the
2948
                overrides. You may either use a string for a single path / glob,
2949
                or a string tuple for multiple paths / globs. Each override is a dictionary of
2950
                field names to the overridden value.
2951

2952
                For example:
2953

2954
                {example}
2955
                """
2956
            ),
2957
            "",
2958
            softwrap(
2959
                f"""
2960
                File paths and globs are relative to the BUILD file's directory. Every overridden file is
2961
                validated to belong to this target's `sources` field.
2962

2963
                If you'd like to override a field's value for every `{generated_target_name}` target
2964
                generated by this target, change the field directly on this target rather than using the
2965
                `overrides` field.
2966

2967
                You can specify the same file name in multiple keys, so long as you don't override the
2968
                same field more than one time for the file.
2969
                """
2970
            ),
2971
        ],
2972
    )
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