-
-
Notifications
You must be signed in to change notification settings - Fork 606
/
dependees.py
185 lines (155 loc) · 6.3 KB
/
dependees.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
# 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 dataclasses import dataclass
from enum import Enum
from typing import Iterable, Set, cast
from pants.base.specs import AddressSpecs, DescendantAddresses
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.rules import goal_rule, register_rules, rule
from pants.engine.selectors import Get, MultiGet
from pants.engine.target import Dependencies, DependenciesRequest, Targets
from pants.util.frozendict import FrozenDict
from pants.util.meta import frozen_after_init
from pants.util.ordered_set import FrozenOrderedSet
@dataclass(frozen=True)
class AddressToDependees:
mapping: FrozenDict[Address, FrozenOrderedSet[Address]]
@rule
async def map_addresses_to_dependees() -> AddressToDependees:
# Get every target in the project so that we can iterate over them to find their dependencies.
all_explicit_targets = await Get(Targets, AddressSpecs([DescendantAddresses("")]))
dependencies_per_target = await MultiGet(
Get(Addresses, DependenciesRequest(tgt.get(Dependencies))) for tgt in all_explicit_targets
)
address_to_dependees = defaultdict(set)
for tgt, dependencies in zip(all_explicit_targets, dependencies_per_target):
for dependency in dependencies:
# TODO(#10354): teach dependees how to work with generated subtargets.
dependency = dependency.maybe_convert_to_base_target()
address_to_dependees[dependency].add(tgt.address)
return AddressToDependees(
FrozenDict(
{addr: FrozenOrderedSet(dependees) for addr, dependees in address_to_dependees.items()}
)
)
@frozen_after_init
@dataclass(unsafe_hash=True)
class DependeesRequest:
addresses: FrozenOrderedSet[Address]
transitive: bool
include_roots: bool
def __init__(
self, addresses: Iterable[Address], *, transitive: bool, include_roots: bool
) -> None:
self.addresses = FrozenOrderedSet(addr.maybe_convert_to_base_target() for addr in addresses)
self.transitive = transitive
self.include_roots = include_roots
class Dependees(DeduplicatedCollection[Address]):
sort_input = True
@rule
def find_dependees(
request: DependeesRequest, address_to_dependees: AddressToDependees
) -> Dependees:
check = set(request.addresses)
known_dependents: Set[Address] = set()
while True:
dependents = set(known_dependents)
for target in check:
target_dependees = address_to_dependees.mapping.get(target, FrozenOrderedSet())
dependents.update(target_dependees)
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 Dependees(result)
known_dependents = dependents
class DependeesOutputFormat(Enum):
text = "text"
json = "json"
class DependeesSubsystem(LineOriented, GoalSubsystem):
"""List all targets that depend on any of the input targets."""
name = "dependees"
@classmethod
def register_options(cls, register):
super().register_options(register)
register(
"--transitive",
default=False,
type=bool,
help=(
"List all targets which transitively depend on the specified target, rather than "
"only targets that directly depend on the specified target."
),
)
register(
"--closed",
type=bool,
default=False,
help="Include the input targets in the output, along with the dependees.",
)
register(
"--output-format",
type=DependeesOutputFormat,
default=DependeesOutputFormat.text,
help=(
"Use `text` for a flattened list of target addresses; use `json` for each key to be "
"the address of one of the specified targets, with its value being "
"a list of that target's dependees, e.g. `{':example': [':dep1', ':dep2']}`."
),
)
@property
def transitive(self) -> bool:
return cast(bool, self.options.transitive)
@property
def closed(self) -> bool:
return cast(bool, self.options.closed)
@property
def output_format(self) -> DependeesOutputFormat:
return cast(DependeesOutputFormat, self.options.output_format)
class DependeesGoal(Goal):
subsystem_cls = DependeesSubsystem
@goal_rule
async def dependees_goal(
specified_addresses: Addresses, dependees_subsystem: DependeesSubsystem, console: Console
) -> DependeesGoal:
if dependees_subsystem.output_format == DependeesOutputFormat.json:
dependees_per_target = await MultiGet(
Get(
Dependees,
DependeesRequest(
[specified_address],
transitive=dependees_subsystem.transitive,
include_roots=dependees_subsystem.closed,
),
)
for specified_address in specified_addresses
)
json_result = {
specified_address.spec: [dependee.spec for dependee in dependees]
for specified_address, dependees in zip(specified_addresses, dependees_per_target)
}
with dependees_subsystem.line_oriented(console) as print_stdout:
print_stdout(json.dumps(json_result, indent=4, separators=(",", ": "), sort_keys=True))
return DependeesGoal(exit_code=0)
dependees = await Get(
Dependees,
DependeesRequest(
specified_addresses,
transitive=dependees_subsystem.transitive,
include_roots=dependees_subsystem.closed,
),
)
with dependees_subsystem.line_oriented(console) as print_stdout:
for address in dependees:
print_stdout(address.spec)
return DependeesGoal(exit_code=0)
def rules():
return register_rules()