25441711719

Committed 06 May 2026 02:31PM UTC coverage: 92.915%. Remained the same

Build # 25441711719

Build Type

push

github

Committed by

web-flow

Commit Message

use sha pin (with comment) format for generated actions (#23312)

Per the GitHub Action best practices we recently enabled at #23249, we
should pin each action to a SHA so that the reference is actually
immutable.

This will -- I hope -- knock out a large chunk of the 421 alerts we
currently get from zizmor. The next followup would then be upgrades and
harmonizing the generated and none-generated pins.

Notice: This idea was suggested by Claude while going over pinact output
and I was surprised to see that post processing the yaml wasn't too
gross.

Coverage Stats

92206 of 99237 relevant lines covered (92.91%)

4.04 hits per line

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

100.0

/src/python/pants/backend/project_info/dependents.py

# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).
import json
from collections import defaultdict
from collections.abc import Iterable
from dataclasses import dataclass
from enum import Enum

from pants.engine.addresses import Address, Addresses
from pants.engine.collection import DeduplicatedCollection
from pants.engine.console import Console
from pants.engine.goal import Goal, GoalSubsystem, LineOriented
from pants.engine.internals.graph import resolve_dependencies
from pants.engine.rules import collect_rules, concurrently, goal_rule, implicitly, rule
from pants.engine.target import (
    AllUnexpandedTargets,
    AlwaysTraverseDeps,
    Dependencies,
    DependenciesRequest,
)
from pants.option.option_types import BoolOption, EnumOption
from pants.util.frozendict import FrozenDict
from pants.util.logging import LogLevel
from pants.util.ordered_set import FrozenOrderedSet


@dataclass(frozen=True)
class AddressToDependents:
    mapping: FrozenDict[Address, FrozenOrderedSet[Address]]


class DependentsOutputFormat(Enum):
    """Output format for listing dependents.

    text: List all dependents as a single list of targets in plain text.
    json: List all dependents as a mapping `{target: [dependents]}`.
    """

    text = "text"
    json = "json"


@rule(desc="Map all targets to their dependents", level=LogLevel.DEBUG)
async def map_addresses_to_dependents(all_targets: AllUnexpandedTargets) -> AddressToDependents:
    dependencies_per_target = await concurrently(
        resolve_dependencies(
            DependenciesRequest(
                tgt.get(Dependencies), should_traverse_deps_predicate=AlwaysTraverseDeps()
            ),
            **implicitly(),
        )
        for tgt in all_targets
    )

    address_to_dependents = defaultdict(set)
    for tgt, dependencies in zip(all_targets, dependencies_per_target):
        for dependency in dependencies:
            address_to_dependents[dependency].add(tgt.address)
    return AddressToDependents(
        FrozenDict(
            {
                addr: FrozenOrderedSet(dependents)
                for addr, dependents in address_to_dependents.items()
            }
        )
    )


@dataclass(frozen=True)
class DependentsRequest:
    addresses: FrozenOrderedSet[Address]
    transitive: bool
    include_roots: bool

    def __init__(
        self, addresses: Iterable[Address], *, transitive: bool, include_roots: bool
    ) -> None:
        object.__setattr__(self, "addresses", FrozenOrderedSet(addresses))
        object.__setattr__(self, "transitive", transitive)
        object.__setattr__(self, "include_roots", include_roots)


class Dependents(DeduplicatedCollection[Address]):
    sort_input = True


@rule(level=LogLevel.DEBUG)
async def find_dependents(
    request: DependentsRequest, address_to_dependents: AddressToDependents
) -> Dependents:
    check = set(request.addresses)
    known_dependents: set[Address] = set()
    while True:
        dependents = set(known_dependents)
        for target in check:
            target_dependents = address_to_dependents.mapping.get(target, FrozenOrderedSet())
            dependents.update(target_dependents)
        check = dependents - known_dependents
        if not check or not request.transitive:
            result = (
                dependents | set(request.addresses)
                if request.include_roots
                else dependents - set(request.addresses)
            )
            return Dependents(result)
        known_dependents = dependents


class DependentsSubsystem(LineOriented, GoalSubsystem):
    name = "dependents"
    help = "List all targets that depend on any of the input files/targets."

    transitive = BoolOption(
        default=False,
        help="List all transitive dependents. If unspecified, list direct dependents only.",
    )
    closed = BoolOption(
        default=False,
        help="Include the input targets in the output, along with the dependents.",
    )
    format = EnumOption(
        default=DependentsOutputFormat.text,
        help="Output format for listing dependents.",
    )


class DependentsGoal(Goal):
    subsystem_cls = DependentsSubsystem
    environment_behavior = Goal.EnvironmentBehavior.LOCAL_ONLY


async def list_dependents_as_plain_text(
    addresses: Addresses, dependents_subsystem: DependentsSubsystem, console: Console
) -> None:
    """Get dependents for given addresses and list them in the console as a single list."""
    dependents = await find_dependents(
        DependentsRequest(
            addresses,
            transitive=dependents_subsystem.transitive,
            include_roots=dependents_subsystem.closed,
        ),
        **implicitly(),
    )
    with dependents_subsystem.line_oriented(console) as print_stdout:
        for address in dependents:
            print_stdout(address.spec)


async def list_dependents_as_json(
    addresses: Addresses, dependents_subsystem: DependentsSubsystem, console: Console
) -> None:
    """Get dependents for given addresses and list them in the console in JSON."""
    dependents_group = await concurrently(
        find_dependents(
            DependentsRequest(
                (address,),
                transitive=dependents_subsystem.transitive,
                include_roots=dependents_subsystem.closed,
            ),
            **implicitly(),
        )
        for address in addresses
    )
    iterated_addresses = []
    for dependents in dependents_group:
        iterated_addresses.append(sorted([str(address) for address in dependents]))
    mapping = dict(zip([str(address) for address in addresses], iterated_addresses))
    output = json.dumps(mapping, indent=4)
    with dependents_subsystem.line_oriented(console) as print_stdout:
        print_stdout(output)


@goal_rule
async def dependents_goal(
    specified_addresses: Addresses, dependents_subsystem: DependentsSubsystem, console: Console
) -> DependentsGoal:
    if DependentsOutputFormat.text == dependents_subsystem.format:
        await list_dependents_as_plain_text(
            addresses=specified_addresses,
            dependents_subsystem=dependents_subsystem,
            console=console,
        )
    elif DependentsOutputFormat.json == dependents_subsystem.format:
        await list_dependents_as_json(
            addresses=specified_addresses,
            dependents_subsystem=dependents_subsystem,
            console=console,
        )
    return DependentsGoal(exit_code=0)


def rules():
    return collect_rules()

1	# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3	import json	12✔
4	from collections import defaultdict	12✔
5	from collections.abc import Iterable	12✔
6	from dataclasses import dataclass	12✔
7	from enum import Enum	12✔
8
9	from pants.engine.addresses import Address, Addresses	12✔
10	from pants.engine.collection import DeduplicatedCollection	12✔
11	from pants.engine.console import Console	12✔
12	from pants.engine.goal import Goal, GoalSubsystem, LineOriented	12✔
13	from pants.engine.internals.graph import resolve_dependencies	12✔
14	from pants.engine.rules import collect_rules, concurrently, goal_rule, implicitly, rule	12✔
15	from pants.engine.target import (	12✔
16	AllUnexpandedTargets,
17	AlwaysTraverseDeps,
18	Dependencies,
19	DependenciesRequest,
20	)
21	from pants.option.option_types import BoolOption, EnumOption	12✔
22	from pants.util.frozendict import FrozenDict	12✔
23	from pants.util.logging import LogLevel	12✔
24	from pants.util.ordered_set import FrozenOrderedSet	12✔
25
26
27	@dataclass(frozen=True)	12✔
28	class AddressToDependents:	12✔
29	mapping: FrozenDict[Address, FrozenOrderedSet[Address]]
30
31
32	class DependentsOutputFormat(Enum):	12✔
33	"""Output format for listing dependents.
34
35	text: List all dependents as a single list of targets in plain text.
36	json: List all dependents as a mapping `{target: [dependents]}`.
37	"""
38
39	text = "text"	12✔
40	json = "json"	12✔
41
42
43	@rule(desc="Map all targets to their dependents", level=LogLevel.DEBUG)	12✔
44	async def map_addresses_to_dependents(all_targets: AllUnexpandedTargets) -> AddressToDependents:	12✔
45	dependencies_per_target = await concurrently(	2✔
46	resolve_dependencies(
47	DependenciesRequest(
48	tgt.get(Dependencies), should_traverse_deps_predicate=AlwaysTraverseDeps()
49	),
50	**implicitly(),
51	)
52	for tgt in all_targets
53	)
54
55	address_to_dependents = defaultdict(set)	2✔
56	for tgt, dependencies in zip(all_targets, dependencies_per_target):	2✔
57	for dependency in dependencies:	2✔
58	address_to_dependents[dependency].add(tgt.address)	2✔
59	return AddressToDependents(	2✔
60	FrozenDict(
61	{
62	addr: FrozenOrderedSet(dependents)
63	for addr, dependents in address_to_dependents.items()
64	}
65	)
66	)
67
68
69	@dataclass(frozen=True)	12✔
70	class DependentsRequest:	12✔
71	addresses: FrozenOrderedSet[Address]
72	transitive: bool
73	include_roots: bool
74
75	def __init__(	12✔
76	self, addresses: Iterable[Address], *, transitive: bool, include_roots: bool
77	) -> None:
78	object.__setattr__(self, "addresses", FrozenOrderedSet(addresses))	2✔
79	object.__setattr__(self, "transitive", transitive)	2✔
80	object.__setattr__(self, "include_roots", include_roots)	2✔
81
82
83	class Dependents(DeduplicatedCollection[Address]):	12✔
84	sort_input = True	12✔
85
86
87	@rule(level=LogLevel.DEBUG)	12✔
88	async def find_dependents(	12✔
89	request: DependentsRequest, address_to_dependents: AddressToDependents
90	) -> Dependents:
91	check = set(request.addresses)	2✔
92	known_dependents: set[Address] = set()	2✔
93	while True:	2✔
94	dependents = set(known_dependents)	2✔
95	for target in check:	2✔
96	target_dependents = address_to_dependents.mapping.get(target, FrozenOrderedSet())	2✔
97	dependents.update(target_dependents)	2✔
98	check = dependents - known_dependents	2✔
99	if not check or not request.transitive:	2✔
100	result = (	2✔
101	dependents \| set(request.addresses)
102	if request.include_roots
103	else dependents - set(request.addresses)
104	)
105	return Dependents(result)	2✔
106	known_dependents = dependents	2✔
107
108
109	class DependentsSubsystem(LineOriented, GoalSubsystem):	12✔
110	name = "dependents"	12✔
111	help = "List all targets that depend on any of the input files/targets."	12✔
112
113	transitive = BoolOption(	12✔
114	default=False,
115	help="List all transitive dependents. If unspecified, list direct dependents only.",
116	)
117	closed = BoolOption(	12✔
118	default=False,
119	help="Include the input targets in the output, along with the dependents.",
120	)
121	format = EnumOption(	12✔
122	default=DependentsOutputFormat.text,
123	help="Output format for listing dependents.",
124	)
125
126
127	class DependentsGoal(Goal):	12✔
128	subsystem_cls = DependentsSubsystem	12✔
129	environment_behavior = Goal.EnvironmentBehavior.LOCAL_ONLY	12✔
130
131
132	async def list_dependents_as_plain_text(	12✔
133	addresses: Addresses, dependents_subsystem: DependentsSubsystem, console: Console
134	) -> None:
135	"""Get dependents for given addresses and list them in the console as a single list."""
136	dependents = await find_dependents(	1✔
137	DependentsRequest(
138	addresses,
139	transitive=dependents_subsystem.transitive,
140	include_roots=dependents_subsystem.closed,
141	),
142	**implicitly(),
143	)
144	with dependents_subsystem.line_oriented(console) as print_stdout:	1✔
145	for address in dependents:	1✔
146	print_stdout(address.spec)	1✔
147
148
149	async def list_dependents_as_json(	12✔
150	addresses: Addresses, dependents_subsystem: DependentsSubsystem, console: Console
151	) -> None:
152	"""Get dependents for given addresses and list them in the console in JSON."""
153	dependents_group = await concurrently(	1✔
154	find_dependents(
155	DependentsRequest(
156	(address,),
157	transitive=dependents_subsystem.transitive,
158	include_roots=dependents_subsystem.closed,
159	),
160	**implicitly(),
161	)
162	for address in addresses
163	)
164	iterated_addresses = []	1✔
165	for dependents in dependents_group:	1✔
166	iterated_addresses.append(sorted([str(address) for address in dependents]))	1✔
167	mapping = dict(zip([str(address) for address in addresses], iterated_addresses))	1✔
168	output = json.dumps(mapping, indent=4)	1✔
169	with dependents_subsystem.line_oriented(console) as print_stdout:	1✔
170	print_stdout(output)	1✔
171
172
173	@goal_rule	12✔
174	async def dependents_goal(	12✔
175	specified_addresses: Addresses, dependents_subsystem: DependentsSubsystem, console: Console
176	) -> DependentsGoal:
177	if DependentsOutputFormat.text == dependents_subsystem.format:	1✔
178	await list_dependents_as_plain_text(	1✔
179	addresses=specified_addresses,
180	dependents_subsystem=dependents_subsystem,
181	console=console,
182	)
183	elif DependentsOutputFormat.json == dependents_subsystem.format:	1✔
184	await list_dependents_as_json(	1✔
185	addresses=specified_addresses,
186	dependents_subsystem=dependents_subsystem,
187	console=console,
188	)
189	return DependentsGoal(exit_code=0)	1✔
190
191
192	def rules():	12✔
193	return collect_rules()	12✔

pantsbuild / pants / 25441711719

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

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