-
-
Notifications
You must be signed in to change notification settings - Fork 606
/
dependencies.py
107 lines (87 loc) · 3.52 KB
/
dependencies.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
# Copyright 2019 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).
import itertools
from enum import Enum
from typing import Set, cast
from pants.backend.python.target_types import PythonRequirementsField
from pants.engine.addresses import Addresses
from pants.engine.console import Console
from pants.engine.goal import Goal, GoalSubsystem, LineOriented
from pants.engine.rules import goal_rule, register_rules
from pants.engine.selectors import Get, MultiGet
from pants.engine.target import Dependencies as DependenciesField
from pants.engine.target import DependenciesRequest, Targets, TransitiveTargets
class DependencyType(Enum):
SOURCE = "source"
THIRD_PARTY = "3rdparty"
SOURCE_AND_THIRD_PARTY = "source-and-3rdparty"
class DependenciesSubsystem(LineOriented, GoalSubsystem):
"""List the dependencies of the input targets."""
name = "dependencies"
@classmethod
def register_options(cls, register):
super().register_options(register)
register(
"--transitive",
default=False,
type=bool,
help=(
"List all transitive dependencies. If unspecified, list direct dependencies only."
),
)
register(
"--type",
type=DependencyType,
default=DependencyType.SOURCE,
help=(
"Which types of dependencies to list, where `source` means source code "
"dependencies and `3rdparty` means third-party requirements."
),
)
@property
def transitive(self) -> bool:
return cast(bool, self.options.transitive)
@property
def type(self) -> DependencyType:
return cast(DependencyType, self.options.type)
class Dependencies(Goal):
subsystem_cls = DependenciesSubsystem
@goal_rule
async def dependencies(
console: Console, addresses: Addresses, dependencies_subsystem: DependenciesSubsystem,
) -> Dependencies:
if dependencies_subsystem.transitive:
transitive_targets = await Get(TransitiveTargets, Addresses, addresses)
targets = Targets(transitive_targets.dependencies)
else:
target_roots = await Get(Targets, Addresses, addresses)
dependencies_per_target_root = await MultiGet(
Get(Targets, DependenciesRequest(tgt.get(DependenciesField))) for tgt in target_roots
)
targets = Targets(itertools.chain.from_iterable(dependencies_per_target_root))
include_source = dependencies_subsystem.type in [
DependencyType.SOURCE,
DependencyType.SOURCE_AND_THIRD_PARTY,
]
include_3rdparty = dependencies_subsystem.type in [
DependencyType.THIRD_PARTY,
DependencyType.SOURCE_AND_THIRD_PARTY,
]
address_strings = set()
third_party_requirements: Set[str] = set()
for tgt in targets:
if include_source:
address_strings.add(tgt.address.spec)
if include_3rdparty:
if tgt.has_field(PythonRequirementsField):
third_party_requirements.update(
str(python_req.requirement) for python_req in tgt[PythonRequirementsField].value
)
with dependencies_subsystem.line_oriented(console) as print_stdout:
for address in sorted(address_strings):
print_stdout(address)
for requirement_string in sorted(third_party_requirements):
print_stdout(requirement_string)
return Dependencies(exit_code=0)
def rules():
return register_rules()