19015773527

Committed 02 Nov 2025 05:33PM UTC coverage: 17.872% (-62.4%) from 80.3%

Build # 19015773527

Build Type

Pull #22816

github

Committed by

web-flow

Commit Message

Merge a12d75757 into 6c024e162

Pull Request Pull Request #22816: Update Pants internal Python to 3.14

Run Details

4 of 5 new or added lines in 3 files covered. (80.0%)

28452 existing lines in 683 files now uncovered.

9831 of 55007 relevant lines covered (17.87%)

0.18 hits per line

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

53.01

/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	1✔
4	from collections import defaultdict	1✔
5	from collections.abc import Iterable	1✔
6	from dataclasses import dataclass	1✔
7	from enum import Enum	1✔
8
9	from pants.engine.addresses import Address, Addresses	1✔
10	from pants.engine.collection import DeduplicatedCollection	1✔
11	from pants.engine.console import Console	1✔
12	from pants.engine.goal import Goal, GoalSubsystem, LineOriented	1✔
13	from pants.engine.internals.graph import resolve_dependencies	1✔
14	from pants.engine.rules import collect_rules, concurrently, goal_rule, implicitly, rule	1✔
15	from pants.engine.target import (	1✔
16	AllUnexpandedTargets,
17	AlwaysTraverseDeps,
18	Dependencies,
19	DependenciesRequest,
20	)
21	from pants.option.option_types import BoolOption, EnumOption	1✔
22	from pants.util.frozendict import FrozenDict	1✔
23	from pants.util.logging import LogLevel	1✔
24	from pants.util.ordered_set import FrozenOrderedSet	1✔
25
26
27	@dataclass(frozen=True)	1✔
28	class AddressToDependents:	1✔
29	mapping: FrozenDict[Address, FrozenOrderedSet[Address]]
30
31
32	class DependentsOutputFormat(Enum):	1✔
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"	1✔
40	json = "json"	1✔
41
42
43	@rule(desc="Map all targets to their dependents", level=LogLevel.DEBUG)	1✔
44	async def map_addresses_to_dependents(all_targets: AllUnexpandedTargets) -> AddressToDependents:	1✔
45	dependencies_per_target = await concurrently(	×
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)	×
56	for tgt, dependencies in zip(all_targets, dependencies_per_target):	×
57	for dependency in dependencies:	×
58	address_to_dependents[dependency].add(tgt.address)	×
59	return AddressToDependents(	×
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)	1✔
70	class DependentsRequest:	1✔
71	addresses: FrozenOrderedSet[Address]
72	transitive: bool
73	include_roots: bool
74
75	def __init__(	1✔
76	self, addresses: Iterable[Address], *, transitive: bool, include_roots: bool
77	) -> None:
78	object.__setattr__(self, "addresses", FrozenOrderedSet(addresses))	×
79	object.__setattr__(self, "transitive", transitive)	×
80	object.__setattr__(self, "include_roots", include_roots)	×
81
82
83	class Dependents(DeduplicatedCollection[Address]):	1✔
84	sort_input = True	1✔
85
86
87	@rule(level=LogLevel.DEBUG)	1✔
88	async def find_dependents(	1✔
89	request: DependentsRequest, address_to_dependents: AddressToDependents
90	) -> Dependents:
91	check = set(request.addresses)	×
92	known_dependents: set[Address] = set()	×
93	while True:	×
94	dependents = set(known_dependents)	×
95	for target in check:	×
96	target_dependents = address_to_dependents.mapping.get(target, FrozenOrderedSet())	×
97	dependents.update(target_dependents)	×
98	check = dependents - known_dependents	×
99	if not check or not request.transitive:	×
100	result = (	×
101	dependents \| set(request.addresses)
102	if request.include_roots
103	else dependents - set(request.addresses)
104	)
105	return Dependents(result)	×
106	known_dependents = dependents	×
107
108
109	class DependentsSubsystem(LineOriented, GoalSubsystem):	1✔
110	name = "dependents"	1✔
111	help = "List all targets that depend on any of the input files/targets."	1✔
112
113	transitive = BoolOption(	1✔
114	default=False,
115	help="List all transitive dependents. If unspecified, list direct dependents only.",
116	)
117	closed = BoolOption(	1✔
118	default=False,
119	help="Include the input targets in the output, along with the dependents.",
120	)
121	format = EnumOption(	1✔
122	default=DependentsOutputFormat.text,
123	help="Output format for listing dependents.",
124	)
125
126
127	class DependentsGoal(Goal):	1✔
128	subsystem_cls = DependentsSubsystem	1✔
129	environment_behavior = Goal.EnvironmentBehavior.LOCAL_ONLY	1✔
130
131
132	async def list_dependents_as_plain_text(	1✔
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(	×
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:	×
145	for address in dependents:	×
146	print_stdout(address.spec)	×
147
148
149	async def list_dependents_as_json(	1✔
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(	×
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 = []	×
165	for dependents in dependents_group:	×
166	iterated_addresses.append(sorted([str(address) for address in dependents]))	×
167	mapping = dict(zip([str(address) for address in addresses], iterated_addresses))	×
168	output = json.dumps(mapping, indent=4)	×
169	with dependents_subsystem.line_oriented(console) as print_stdout:	×
170	print_stdout(output)	×
171
172
173	@goal_rule	1✔
174	async def dependents_goal(	1✔
175	specified_addresses: Addresses, dependents_subsystem: DependentsSubsystem, console: Console
176	) -> DependentsGoal:
177	if DependentsOutputFormat.text == dependents_subsystem.format:	×
178	await list_dependents_as_plain_text(	×
179	addresses=specified_addresses,
180	dependents_subsystem=dependents_subsystem,
181	console=console,
182	)
183	elif DependentsOutputFormat.json == dependents_subsystem.format:	×
184	await list_dependents_as_json(	×
185	addresses=specified_addresses,
186	dependents_subsystem=dependents_subsystem,
187	console=console,
188	)
189	return DependentsGoal(exit_code=0)	×
190
191
192	def rules():	1✔
UNCOV 193	return collect_rules()	×

pantsbuild / pants / 19015773527

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