forked from pantsbuild/pants
-
Notifications
You must be signed in to change notification settings - Fork 1
/
rules.py
167 lines (141 loc) · 5.77 KB
/
rules.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
# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).
from pathlib import PurePath
from pants.backend.python.dependency_inference import module_mapper
from pants.backend.python.dependency_inference.import_parser import find_python_imports
from pants.backend.python.dependency_inference.module_mapper import PythonModule, PythonModuleOwner
from pants.backend.python.dependency_inference.python_stdlib.combined import combined_stdlib
from pants.backend.python.rules import inject_ancestor_files
from pants.backend.python.rules.inject_ancestor_files import AncestorFiles, AncestorFilesRequest
from pants.backend.python.target_types import PythonSources, PythonTestsSources
from pants.core.util_rules.strip_source_roots import (
SourceRootStrippedSources,
StripSourcesFieldRequest,
)
from pants.engine.fs import Digest, DigestContents
from pants.engine.internals.graph import Owners, OwnersRequest
from pants.engine.rules import collect_rules, rule
from pants.engine.selectors import Get, MultiGet
from pants.engine.target import (
HydratedSources,
HydrateSourcesRequest,
InferDependenciesRequest,
InferredDependencies,
)
from pants.engine.unions import UnionRule
from pants.option.global_options import OwnersNotFoundBehavior
from pants.subsystem.subsystem import Subsystem
class PythonInference(Subsystem):
"""Options controlling which dependencies will be inferred for Python targets."""
options_scope = "python-infer"
@classmethod
def register_options(cls, register):
super().register_options(register)
register(
"--imports",
default=False,
type=bool,
help=(
"Infer a target's imported dependencies by parsing import statements from sources."
),
)
register(
"--inits",
default=True,
type=bool,
help=(
"Infer a target's dependencies on any __init__.py files existing for the packages "
"it is located in (recursively upward in the directory structure)."
),
)
register(
"--conftests",
default=True,
type=bool,
help=(
"Infer a test target's dependencies on any conftest.py files in parent directories."
),
)
class InferPythonDependencies(InferDependenciesRequest):
infer_from = PythonSources
@rule(desc="Inferring Python dependencies.")
async def infer_python_dependencies(
request: InferPythonDependencies, python_inference: PythonInference
) -> InferredDependencies:
if not python_inference.get_options().imports:
return InferredDependencies()
stripped_sources = await Get(
SourceRootStrippedSources, StripSourcesFieldRequest(request.sources_field)
)
modules = tuple(
PythonModule.create_from_stripped_path(PurePath(fp))
for fp in stripped_sources.snapshot.files
)
digest_contents = await Get(DigestContents, Digest, stripped_sources.snapshot.digest)
imports_per_file = tuple(
find_python_imports(file_content.content.decode(), module_name=module.module)
for file_content, module in zip(digest_contents, modules)
)
owner_per_import = await MultiGet(
Get(PythonModuleOwner, PythonModule(imported_module))
for file_imports in imports_per_file
for imported_module in file_imports.explicit_imports
if imported_module not in combined_stdlib
)
return InferredDependencies(
owner.address
for owner in owner_per_import
if (
owner.address
and owner.address.maybe_convert_to_base_target() != request.sources_field.address
)
)
class InferInitDependencies(InferDependenciesRequest):
infer_from = PythonSources
@rule
async def infer_python_init_dependencies(
request: InferInitDependencies, python_inference: PythonInference
) -> InferredDependencies:
if not python_inference.get_options().inits:
return InferredDependencies()
# Locate __init__.py files not already in the Snapshot.
hydrated_sources = await Get(HydratedSources, HydrateSourcesRequest(request.sources_field))
extra_init_files = await Get(
AncestorFiles,
AncestorFilesRequest("__init__.py", hydrated_sources.snapshot, sources_stripped=False),
)
# And add dependencies on their owners.
return InferredDependencies(
await Get(
Owners, OwnersRequest(extra_init_files.snapshot.files, OwnersNotFoundBehavior.error)
)
)
class InferConftestDependencies(InferDependenciesRequest):
infer_from = PythonTestsSources
@rule
async def infer_python_conftest_dependencies(
request: InferConftestDependencies, python_inference: PythonInference,
) -> InferredDependencies:
if not python_inference.get_options().conftests:
return InferredDependencies()
# Locate conftest.py files not already in the Snapshot.
hydrated_sources = await Get(HydratedSources, HydrateSourcesRequest(request.sources_field))
extra_conftest_files = await Get(
AncestorFiles,
AncestorFilesRequest("conftest.py", hydrated_sources.snapshot, sources_stripped=False),
)
# And add dependencies on their owners.
return InferredDependencies(
await Get(
Owners, OwnersRequest(extra_conftest_files.snapshot.files, OwnersNotFoundBehavior.error)
)
)
def rules():
return [
*collect_rules(),
*inject_ancestor_files.rules(),
*module_mapper.rules(),
UnionRule(InferDependenciesRequest, InferPythonDependencies),
UnionRule(InferDependenciesRequest, InferInitDependencies),
UnionRule(InferDependenciesRequest, InferConftestDependencies),
]