pantsbuild / pants / 19015773527

# Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).

# Licensed under the Apache License, Version 2.0 (see LICENSE).

    _get_target_family_and_adaptor_for_dep_rules,

    get_dependencies_rule_application,

)

    AlwaysTraverseDeps,

    Dependencies,

    DependenciesRequest,

    DependenciesRuleApplicationRequest,

    Field,

    FieldSet,

    HydrateSourcesRequest,

    ImmutableValue,

    NoApplicableTargetsBehavior,

    SourcesField,

    Target,

    TargetRootsToFieldSetsRequest,

    UnexpandedTargets,

)

    """Make possible to avoid adding concrete types to serialize objects."""

    def asdict(self) -> Mapping[str, Any]: ...

    """Display detailed target information in JSON form."""

        default=False,

        help="Whether to leave off values that match the target-defined default values.",

    )

        default=False,

        help=softwrap(

            """

            Whether to include `_dependencies_rules`, `_dependents_rules` and `_applicable_dep_rules`

            that apply to the target and its dependencies.

            """

        ),

    )

        default=False, help="Whether to include additional information generated by plugins."

    )

    """Union type to attach data to a target that will appear under the "additional_info" field in

    the output of `pants peek` if the `--peek-include-additional-info` option is enabled."""

    field_set: HasAdditionalTargetDataFieldSet, env_name: EnvironmentName

) -> AdditionalTargetData:

    # Sources may not be registered on the target, so we'll have nothing to expand.

            (

                f"{k.alias}_raw" if issubclass(k, (SourcesField, Dependencies)) else k.alias

            ): _normalize_value(v.value)

            for k, v in self.target.field_values.items()

            if not (exclude_defaults and getattr(k, "default", nothing) == v.value)

        }

                additional_target_data.label: additional_target_data.data

                for additional_target_data in sorted(

                    self.additional_info, key=lambda atd: atd.label

                )

            }

            "address": self.target.address.spec,

            "target_type": self.target.alias,

            **dict(sorted(fields.items())),

        }

    tds: Iterable[TargetData], exclude_defaults: bool = False, include_dep_rules: bool = False

) -> str:

    """Allow us to serialize some commonly found types in BUILD files."""

        """Return a serializable object for o."""

            # NB: `dataclasses.asdict` creates a deep copy by default, which is unnecessary for

            # this case.

            isinstance(o, collections.abc.Sequence)

            or isinstance(o, set)

            or isinstance(o, collections.abc.Set)

        ):

    """Returns a mapping from a target alias to the goals that can operate on that target.

    For instance, `pex_binary` would map to `("run", "package")`.

    :return: A mapping from a target alias to the goals that can operate on that target.

    """

    # This is manually curated for now - we'll have to use it a bit to determine if we want to show all goals or not

        DeployFieldSet,

        PackageFieldSet,

        PublishFieldSet,

        RunFieldSet,

        TestFieldSet,

    ]

    # Goal holds a GoalSubsystem which has the name we care about, and it's exposed via a classmethod on Goal

    # There is no tightly coupled relationship between a Goal/GoalSubsystem and the associated FieldSet

    # This gets murkier with Fix/Fmt/Lint/etc... So, we'll just manually map them for now

        DeployFieldSet: Deploy.name,

        PackageFieldSet: Package.name,

        PublishFieldSet: Publish.name,

        RunFieldSet: Run.name,

        TestFieldSet: Test.name,

    }

        "Must have a goal string for each field set"

    )

        find_valid_field_sets_for_target_roots(

            TargetRootsToFieldSetsRequest(

                field_set_superclass=fs,

                goal_description="",

                no_applicable_targets_behavior=NoApplicableTargetsBehavior.ignore,

            ),

            **implicitly(),

        )

        for fs in peekable_field_sets

    ]

    # Create a collection of target aliases per target roots: e.g. [frozenset(), frozenset({'pyoxidizer_binary', 'pex_binary'}), ...]

        frozenset(tgt.alias for tgt in tgt_root.targets) for tgt_root in target_roots_to_field_sets

    ]

    # Create a mapping from the goal name to a collection of target aliases: e.g. {'run': frozenset({'pyoxidizer_binary', 'pex_binary'}), 'test': frozenset(), ...}

    # Inverse the goal_to_aliases_map to create a mapping from a target alias to a collection of goal names: e.g. {'pyoxidizer_binary': frozenset({'package', 'run'}), 'pex_binary': frozenset({'package', 'run'}), ...}

    # Convert the goal sets to tuples for JSON serialization

    targets: UnexpandedTargets,

    subsys: PeekSubsystem,

    union_membership: UnionMembership,

) -> TargetDatas:

    """Given a set of unexpanded targets, return a mapping of target addresses to their data.

    The data includes the target's fields, expanded sources, expanded dependencies, and any

    dependency rules that apply to the target.

    NB: We must preserve target generators, not replace with their generated targets.

    :param targets: The unexpanded targets to get data for.

    :param subsys: The `PeekSubsystem` instance.

    :return: A mapping of target addresses to their data.

    """

    # We "hydrate" sources fields with the engine, but not every target has them registered.

    # When determining dependencies, we replace target generators with their generated targets.

        resolve_targets(

            **implicitly(

                DependenciesRequest(

                    tgt.get(Dependencies), should_traverse_deps_predicate=AlwaysTraverseDeps()

                )

            )

        )

        for tgt in sorted_targets

    )

        hydrate_sources(HydrateSourcesRequest(tgt[SourcesField]), **implicitly())

        for tgt in targets_with_sources

    )

            field_set_type.create(tgt)

            for tgt in sorted_targets

            for field_set_type in union_membership[HasAdditionalTargetDataFieldSet]

            if field_set_type.is_applicable(tgt)

        ]

            get_additional_target_data(**implicitly({field_set: HasAdditionalTargetDataFieldSet}))

            for field_set in additional_info_field_sets

        )

            address: tuple(additional_info)

            for address, additional_info in group_additional_infos_by_address_builder.items()

        }

    else:

    # TODO: This feels like something that could be merged with the above code somewhere

        tgt.address: hs.snapshot

        for tgt, hs in zip(targets_with_sources, hydrated_sources_per_target)

    }

        tuple(dep.address.spec for dep in deps)

        for _, deps in zip(sorted_targets, dependencies_per_target)

    ]

        # TODO: Why are we pulling a private definition from `build_files`?

            *(tgt.address for tgt in sorted_targets),

            description_of_origin="`peek` goal",

        )

            tgt.address: describe_ruleset(

                family.dependencies_rules.get_ruleset(tgt.address, adaptor)

            )

            for tgt, (family, adaptor) in zip(sorted_targets, family_adaptors)

            if family.dependencies_rules is not None

        }

            tgt.address: describe_ruleset(family.dependents_rules.get_ruleset(tgt.address, adaptor))

            for tgt, (family, adaptor) in zip(sorted_targets, family_adaptors)

            if family.dependents_rules is not None

        }

            get_dependencies_rule_application(

                DependenciesRuleApplicationRequest(

                    tgt.address,

                    Addresses(dep.address for dep in deps),

                    description_of_origin="`peek` goal",

                ),

                **implicitly(),

            )

            for tgt, deps in zip(sorted_targets, dependencies_per_target)

        )

            application.address: tuple(application.dependencies_rule.values())

            for application in all_applicable_dep_rules

        }

    # TODO: This currently exists in the `goal_rule` section of code below, but I'd prefer to have it here

    # target_alias_to_goals_map = await _create_target_alias_to_goals_map() if subsys.include_goals else {}

        TargetData(

            tgt,

            expanded_dependencies=expanded_deps,

            expanded_sources=expanded_sources_map.get(tgt.address),

            dependencies_rules=dependencies_rules_map.get(tgt.address),

            dependents_rules=dependents_rules_map.get(tgt.address),

            applicable_dep_rules=applicable_dep_rules_map.get(tgt.address),

            goals=None,

            additional_info=group_additional_infos_by_address.get(

                tgt.address, () if subsys.include_additional_info else None

            ),

        )

        for tgt, expanded_deps in zip(sorted_targets, expanded_dependencies)

    )

    console: Console,

    subsys: PeekSubsystem,

    targets: UnexpandedTargets,

) -> Peek:

    """Display detailed target information in JSON form.

    :param console: The console to write the JSON to.

    :param subsys: The `PeekSubsystem` instance.

    :param targets: The targets to get data for.

    :return: The `Peek` goal.

    """

    # This method needs to be called in a @goal_rule, otherwise it fails out with Rule errors (when called in an @rule)

        # Attach the goals to the target data, in the hopes that we can pull `_create_target_alias_to_goals_map` back into `get_target_data`

        # TargetData is frozen so we need to create a new collection

            [

                replace(

                    td,

                    goals=target_alias_to_goals_map.get(td.target.alias),

                )

                for td in tds

            ]

        )