25259185675

Committed 02 May 2026 06:47PM UTC coverage: 92.141% (-0.8%) from 92.955%

Build # 25259185675

Build Type

push

github

Committed by

web-flow

Commit Message

Fix the dynamic UI. (#23306)

In #23114 we upgraded to indicatif 0.18.4,
which included a fix to respect TERM, and 
display nothing if it's unset.

Since we did not pass TERM through pantsd, the
dynamic ui is now not shown. 

This change fixes that, and also pass NO_COLOR
through, since indicatif inspects it too.

Coverage Stats

88773 of 96345 relevant lines covered (92.14%)

3.83 hits per line

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

79.52

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

pantsbuild / pants / 25259185675

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