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

from __future__ import annotations

import collections.abc
import dataclasses
import enum
import glob as glob_stdlib
import itertools
import logging
import os.path
import textwrap
import zlib
from abc import ABC, ABCMeta, abstractmethod
from collections import deque
from collections.abc import Callable, Iterable, Iterator, KeysView, Mapping, Sequence
from dataclasses import dataclass
from enum import Enum
from operator import attrgetter
from pathlib import PurePath
from typing import (
    AbstractSet,
    Any,
    ClassVar,
    Generic,
    Protocol,
    Self,
    TypeVar,
    cast,
    final,
    get_type_hints,
)

from pants.base.deprecated import warn_or_error
from pants.engine.addresses import Address, Addresses, UnparsedAddressInputs, assert_single_address
from pants.engine.collection import Collection
from pants.engine.engine_aware import EngineAwareParameter
from pants.engine.environment import EnvironmentName
from pants.engine.fs import (
    GlobExpansionConjunction,
    GlobMatchErrorBehavior,
    PathGlobs,
    Paths,
    Snapshot,
)
from pants.engine.internals.dep_rules import (
    DependencyRuleActionDeniedError,
    DependencyRuleApplication,
)
from pants.engine.internals.native_engine import NO_VALUE as NO_VALUE  # noqa: F401
from pants.engine.internals.native_engine import Field as Field
from pants.engine.internals.target_adaptor import SourceBlock, SourceBlocks  # noqa: F401
from pants.engine.rules import rule
from pants.engine.unions import UnionMembership, UnionRule, distinct_union_type_per_subclass, union
from pants.option.bootstrap_options import UnmatchedBuildFileGlobs
from pants.source.filespec import Filespec, FilespecMatcher
from pants.util.collections import ensure_list, ensure_str_list
from pants.util.dirutil import fast_relpath
from pants.util.docutil import bin_name, doc_url
from pants.util.frozendict import FrozenDict
from pants.util.memo import memoized_classproperty, memoized_method, memoized_property
from pants.util.ordered_set import FrozenOrderedSet
from pants.util.strutil import bullet_list, help_text, pluralize, softwrap

logger = logging.getLogger(__name__)

# -----------------------------------------------------------------------------------------------
# Core Field abstractions
# -----------------------------------------------------------------------------------------------

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


# NB: By subclassing `Field`, MyPy understands our type hints, and it means it doesn't matter which
# order you use for inheriting the field template vs. the mixin.
class AsyncFieldMixin(Field):
    """A mixin to store the field's original `Address` for use during hydration by the engine.

    Typically, you should also create a dataclass representing the hydrated value and another for
    the request, then a rule to go from the request to the hydrated value. The request class should
    store the async field as a property.

    (Why use the request class as the rule input, rather than the field itself? It's a wrapper so
    that subclasses of the async field work properly, given that the engine uses exact type IDs.
    This is like WrappedTarget.)

    For example:

        class Sources(StringSequenceField, AsyncFieldMixin):
            alias = "sources"

            # Often, async fields will want to define entry points like this to allow subclasses to
            # change behavior.
            def validate_resolved_files(self, files: Sequence[str]) -> None:
                pass


        @dataclass(frozen=True)
        class HydrateSourcesRequest:
            field: Sources


        @dataclass(frozen=True)
        class HydratedSources:
            snapshot: Snapshot


        @rule
        async def hydrate_sources(request: HydrateSourcesRequest) -> HydratedSources:
            digest = await path_globs_to_digest(PathGlobs(request.field.value))
            result = await digest_to_snapshot(digest)
            request.field.validate_resolved_files(result.files)
            ...
            return HydratedSources(result)

    Then, call sites can `await` if they need to hydrate the field, even if they subclassed
    the original async field to have custom behavior:

        sources1 = hydrate_sources(HydrateSourcesRequest(my_tgt.get(Sources)))
        sources2 = hydrate_sources(HydrateSourcesRequest(custom_tgt.get(CustomSources)))
    """

    address: Address

    @final
    def __new__(cls, raw_value: Any | None, address: Address) -> Self:
        obj = super().__new__(cls, raw_value, address)  # type: ignore[call-arg]
        # N.B.: We store the address here and not in the Field base class, because the memory usage
        # of storing this value in every field was shown to be excessive / lead to performance
        # issues.
        object.__setattr__(obj, "address", address)
        return obj

    def __repr__(self) -> str:
        params = [
            f"alias={self.alias!r}",
            f"address={self.address}",
            f"value={self.value!r}",
        ]
        if hasattr(self, "default"):
            params.append(f"default={self.default!r}")
        return f"{self.__class__}({', '.join(params)})"

    def __hash__(self) -> int:
        return hash((self.__class__, self.value, self.address))

    def __eq__(self, other: Any) -> bool:
        if not isinstance(other, AsyncFieldMixin):
            return False
        return (
            self.__class__ == other.__class__
            and self.value == other.value
            and self.address == other.address
        )

    def __ne__(self, other: Any) -> bool:
        return not (self == other)


@union
@dataclass(frozen=True)
class FieldDefaultFactoryRequest:
    """Registers a dynamic default for a Field.

    See `FieldDefaults`.
    """

    field_type: ClassVar[type[Field]]


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


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

    default_factory: FieldDefaultFactory


@dataclass(frozen=True)
class FieldDefaults:
    """Generic Field default values. To install a default, see `FieldDefaultFactoryRequest`.

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

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

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

    _factories: FrozenDict[type[Field], FieldDefaultFactory]

    @memoized_method
    def factory(self, field_type: type[Field]) -> FieldDefaultFactory:
        """Looks up a Field default factory in a subclass-aware way."""
        factory = self._factories.get(field_type, None)
        if factory is not None:
            return factory

        for ft, factory in self._factories.items():
            if issubclass(field_type, ft):
                return factory

        return lambda f: f.value

    def value_or_default(self, field: Field) -> Any:
        return (self.factory(type(field)))(field)


# -----------------------------------------------------------------------------------------------
# Core Target abstractions
# -----------------------------------------------------------------------------------------------


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


@dataclass(frozen=True)
class Target:
    """A Target represents an addressable set of metadata.

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

    # Subclasses must define these
    alias: ClassVar[str]
    core_fields: ClassVar[tuple[type[Field], ...]]
    help: ClassVar[str | Callable[[], str]]

    removal_version: ClassVar[str | None] = None
    removal_hint: ClassVar[str | None] = None

    deprecated_alias: ClassVar[str | None] = None
    deprecated_alias_removal_version: ClassVar[str | None] = None

    # These get calculated in the constructor
    address: Address
    field_values: FrozenDict[type[Field], Field]
    residence_dir: str
    name_explicitly_set: bool
    description_of_origin: str
    origin_sources_blocks: FrozenDict[str, SourceBlocks]

    @final
    def __init__(
        self,
        unhydrated_values: Mapping[str, Any],
        address: Address,
        # NB: `union_membership` is only optional to facilitate tests. In production, we should
        # always provide this parameter. This should be safe to do because production code should
        # rarely directly instantiate Targets and should instead use the engine to request them.
        union_membership: UnionMembership | None = None,
        *,
        name_explicitly_set: bool = True,
        residence_dir: str | None = None,
        ignore_unrecognized_fields: bool = False,
        description_of_origin: str | None = None,
        origin_sources_blocks: FrozenDict[str, SourceBlocks] = FrozenDict(),
    ) -> None:
        """Create a target.

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

        if origin_sources_blocks:
            _validate_origin_sources_blocks(origin_sources_blocks)

        object.__setattr__(
            self, "residence_dir", residence_dir if residence_dir is not None else address.spec_path
        )
        object.__setattr__(self, "address", address)
        object.__setattr__(
            self, "description_of_origin", description_of_origin or self.residence_dir
        )
        object.__setattr__(self, "origin_sources_blocks", origin_sources_blocks)
        object.__setattr__(self, "name_explicitly_set", name_explicitly_set)
        try:
            object.__setattr__(
                self,
                "field_values",
                self._calculate_field_values(
                    unhydrated_values,
                    address,
                    union_membership,
                    ignore_unrecognized_fields=ignore_unrecognized_fields,
                ),
            )

            self.validate()
        except Exception as e:
            raise InvalidTargetException(
                str(e), description_of_origin=self.description_of_origin
            ) from e

    @final
    def _calculate_field_values(
        self,
        unhydrated_values: Mapping[str, Any],
        address: Address,
        # See `__init__`.
        union_membership: UnionMembership | None,
        *,
        ignore_unrecognized_fields: bool,
    ) -> FrozenDict[type[Field], Field]:
        all_field_types = self.class_field_types(union_membership)
        field_values = {}
        aliases_to_field_types = self._get_field_aliases_to_field_types(all_field_types)

        for alias, value in unhydrated_values.items():
            if alias not in aliases_to_field_types:
                if ignore_unrecognized_fields:
                    continue
                valid_aliases = set(aliases_to_field_types.keys())
                if isinstance(self, TargetGenerator):
                    # Even though moved_fields don't live on the target generator, they are valid
                    # for users to specify. It's intentional that these are only used for
                    # `InvalidFieldException` and are not stored as normal fields with
                    # `aliases_to_field_types`.
                    for field_type in self.moved_fields:
                        valid_aliases.add(field_type.alias)
                        if field_type.deprecated_alias is not None:
                            valid_aliases.add(field_type.deprecated_alias)
                raise InvalidFieldException(
                    f"Unrecognized field `{alias}={value}` in target {address}. Valid fields for "
                    f"the target type `{self.alias}`: {sorted(valid_aliases)}.",
                )
            field_type = aliases_to_field_types[alias]
            field_values[field_type] = field_type(value, address)

        # For undefined fields, mark the raw value as missing.
        for field_type in all_field_types:
            if field_type in field_values:
                continue
            field_values[field_type] = field_type(NO_VALUE, address)
        return FrozenDict(
            sorted(
                field_values.items(),
                key=lambda field_type_to_val_pair: field_type_to_val_pair[0].alias,
            )
        )

    @final
    @classmethod
    def _get_field_aliases_to_field_types(
        cls, field_types: Iterable[type[Field]]
    ) -> dict[str, type[Field]]:
        aliases_to_field_types = {}
        for field_type in field_types:
            aliases_to_field_types[field_type.alias] = field_type
            if field_type.deprecated_alias is not None:
                aliases_to_field_types[field_type.deprecated_alias] = field_type
        return aliases_to_field_types

    @final
    @property
    def field_types(self) -> KeysView[type[Field]]:
        return self.field_values.keys()

    @distinct_union_type_per_subclass
    class PluginField:
        pass

    def __repr__(self) -> str:
        fields = ", ".join(str(field) for field in self.field_values.values())
        return (
            f"{self.__class__}("
            f"address={self.address}, "
            f"alias={self.alias!r}, "
            f"residence_dir={self.residence_dir!r}, "
            f"origin={self.description_of_origin}, "
            f"{fields})"
        )

    def __str__(self) -> str:
        fields = ", ".join(str(field) for field in self.field_values.values())
        address = f'address="{self.address}"{", " if fields else ""}'
        return f"{self.alias}({address}{fields})"

    def __hash__(self) -> int:
        return hash((self.__class__, self.address, self.residence_dir, self.field_values))

    def __eq__(self, other: Target | Any) -> bool:
        if not isinstance(other, Target):
            return NotImplemented
        return (self.__class__, self.address, self.residence_dir, self.field_values) == (
            other.__class__,
            other.address,
            other.residence_dir,
            other.field_values,
        )

    def __lt__(self, other: Any) -> bool:
        if not isinstance(other, Target):
            return NotImplemented
        return self.address < other.address

    def __gt__(self, other: Any) -> bool:
        if not isinstance(other, Target):
            return NotImplemented
        return self.address > other.address

    @classmethod
    @memoized_method
    def _find_plugin_fields(cls, union_membership: UnionMembership) -> tuple[type[Field], ...]:
        result: set[type[Field]] = set()
        classes = [cls]
        while classes:
            cls = classes.pop()
            classes.extend(cls.__bases__)
            if issubclass(cls, Target):
                result.update(cast("set[type[Field]]", union_membership.get(cls.PluginField)))

        return tuple(sorted(result, key=attrgetter("alias")))

    @final
    @classmethod
    def _find_registered_field_subclass(
        cls, requested_field: type[_F], *, registered_fields: Iterable[type[Field]]
    ) -> type[_F] | None:
        """Check if the Target has registered a subclass of the requested Field.

        This is necessary to allow targets to override the functionality of common fields. For
        example, you could subclass `Tags` to define `CustomTags` with a different default. At the
        same time, we still want to be able to call `tgt.get(Tags)`, in addition to
        `tgt.get(CustomTags)`.
        """
        subclass = next(
            (
                registered_field
                for registered_field in registered_fields
                if issubclass(registered_field, requested_field)
            ),
            None,
        )
        return subclass

    @final
    def _maybe_get(self, field: type[_F]) -> _F | None:
        result = self.field_values.get(field, None)
        if result is not None:
            return cast(_F, result)
        field_subclass = self._find_registered_field_subclass(
            field, registered_fields=self.field_values.keys()
        )
        if field_subclass is not None:
            return cast(_F, self.field_values[field_subclass])
        return None

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

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

        See the docstring for `Target.get()` for how this method handles subclasses of the
        requested Field and for tips on how to use the returned value.
        """
        result = self._maybe_get(field)
        if result is not None:
            return result
        raise KeyError(
            f"The target `{self}` does not have a field `{field.__name__}`. Before calling "
            f"`my_tgt[{field.__name__}]`, call `my_tgt.has_field({field.__name__})` to "
            f"filter out any irrelevant Targets or call `my_tgt.get({field.__name__})` to use the "
            f"default Field value."
        )

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

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

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

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

    @final
    @classmethod
    def _has_fields(
        cls, fields: Iterable[type[Field]], *, registered_fields: AbstractSet[type[Field]]
    ) -> bool:
        unrecognized_fields = [field for field in fields if field not in registered_fields]
        if not unrecognized_fields:
            return True
        for unrecognized_field in unrecognized_fields:
            maybe_subclass = cls._find_registered_field_subclass(
                unrecognized_field, registered_fields=registered_fields
            )
            if maybe_subclass is None:
                return False
        return True

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

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

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

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

    @final
    @classmethod
    @memoized_method
    def class_field_types(
        cls, union_membership: UnionMembership | None
    ) -> FrozenOrderedSet[type[Field]]:
        """Return all registered Fields belonging to this target type.

        You can also use the instance property `tgt.field_types` to avoid having to pass the
        parameter UnionMembership.
        """
        if union_membership is None:
            return FrozenOrderedSet(cls.core_fields)
        else:
            return FrozenOrderedSet((*cls.core_fields, *cls._find_plugin_fields(union_membership)))

    @final
    @classmethod
    def class_has_field(cls, field: type[Field], union_membership: UnionMembership) -> bool:
        """Behaves like `Target.has_field()`, but works as a classmethod rather than an instance
        method."""
        return cls.class_has_fields([field], union_membership)

    @final
    @classmethod
    def class_has_fields(
        cls, fields: Iterable[type[Field]], union_membership: UnionMembership
    ) -> bool:
        """Behaves like `Target.has_fields()`, but works as a classmethod rather than an instance
        method."""
        return cls._has_fields(fields, registered_fields=cls.class_field_types(union_membership))

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

        This will error if the field is not registered, so you should call Target.class_has_field()
        first.
        """
        class_fields = cls.class_field_types(union_membership)
        result = next(
            (
                registered_field
                for registered_field in class_fields
                if issubclass(registered_field, field)
            ),
            None,
        )
        if result is None:
            raise KeyError(
                f"The target type `{cls.alias}` does not have a field `{field.__name__}`. Before "
                f"calling `TargetType.class_get_field({field.__name__})`, call "
                f"`TargetType.class_has_field({field.__name__})`."
            )
        return result

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

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

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

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


def _validate_origin_sources_blocks(origin_sources_blocks: FrozenDict[str, SourceBlocks]) -> None:
    if not isinstance(origin_sources_blocks, FrozenDict):
        raise ValueError(
            f"Expected `origin_sources_blocks` to be of type `FrozenDict`, got {type(origin_sources_blocks)=} {origin_sources_blocks=}"
        )
    for blocks in origin_sources_blocks.values():
        if not isinstance(blocks, SourceBlocks):
            raise ValueError(
                f"Expected `origin_sources_blocks` to be a `FrozenDict` with values of type `SourceBlocks`, got values of {type(blocks)=} {blocks=}"
            )
        for block in blocks:
            if not isinstance(block, SourceBlock):
                raise ValueError(
                    f"Expected `origin_sources_blocks` to be a `FrozenDict` with values of type `SourceBlocks`, got values of {type(blocks)=} {blocks=}"
                )


@dataclass(frozen=True)
class WrappedTargetRequest:
    """Used with `WrappedTarget` to get the Target corresponding to an address.

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

    address: Address
    description_of_origin: str = dataclasses.field(hash=False, compare=False)


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

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

    target: Target


class Targets(Collection[Target]):
    """A heterogeneous collection of instances of Target subclasses.

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

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

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

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

    def expect_single(self) -> Target:
        assert_single_address([tgt.address for tgt in self])
        return self[0]


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

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

    def expect_single(self) -> Target:
        assert_single_address([tgt.address for tgt in self])
        return self[0]


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

    def expect_single(self) -> Target:
        assert_single_address([tgt.address for tgt in self])
        return self[0]


class DepsTraversalBehavior(Enum):
    """The return value for ShouldTraverseDepsPredicate.

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

    INCLUDE = "include"
    EXCLUDE = "exclude"


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

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

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

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

    def __post_init__(self):
        object.__setattr__(self, "_callable", type(self).__call__)

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


class TraverseIfDependenciesField(ShouldTraverseDepsPredicate):
    """This is the default ShouldTraverseDepsPredicate implementation.

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

    def __call__(
        self, target: Target, field: Dependencies | SpecialCasedDependencies
    ) -> DepsTraversalBehavior:
        if isinstance(field, Dependencies):
            return DepsTraversalBehavior.INCLUDE
