-
-
Notifications
You must be signed in to change notification settings - Fork 607
/
python_sources.py
142 lines (114 loc) · 4.82 KB
/
python_sources.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
# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).
from dataclasses import dataclass
from typing import Iterable, List, Tuple, Type
from pants.backend.python.target_types import PythonSources
from pants.core.target_types import FilesSources, ResourcesSources
from pants.core.util_rules import determine_source_files
from pants.core.util_rules.determine_source_files import AllSourceFilesRequest, SourceFiles
from pants.core.util_rules.strip_source_roots import representative_path_from_address
from pants.engine.fs import Snapshot
from pants.engine.rules import RootRule, rule
from pants.engine.selectors import Get, MultiGet
from pants.engine.target import Sources, Target
from pants.source.source_root import SourceRoot, SourceRootRequest
from pants.util.meta import frozen_after_init
@frozen_after_init
@dataclass(unsafe_hash=True)
class _BasePythonSourcesRequest:
targets: Tuple[Target, ...]
include_resources: bool
include_files: bool
def __init__(
self,
targets: Iterable[Target],
*,
include_resources: bool = True,
include_files: bool = False
) -> None:
self.targets = tuple(targets)
self.include_resources = include_resources
self.include_files = include_files
@property
def valid_sources_types(self) -> Tuple[Type[Sources], ...]:
types: List[Type[Sources]] = [PythonSources]
if self.include_resources:
types.append(ResourcesSources)
if self.include_files:
types.append(FilesSources)
return tuple(types)
@dataclass(frozen=True)
class StrippedPythonSources:
"""Sources that can be imported and used by Python, relative to the root.
Specifically, this will filter out to only have Python, and, optionally, resources() and files()
targets; strip source roots, e.g. `src/python/f.py` -> `f.py`; and will add any missing
`__init__.py` files to ensure that modules are recognized correctly.
Use-cases that execute the Python source code (e.g., the `run`, `binary` and `repl` goals) can
request this type to get a single tree of relevant sources that can be run without sys.path
manipulation.
"""
snapshot: Snapshot
class StrippedPythonSourcesRequest(_BasePythonSourcesRequest):
pass
@dataclass(frozen=True)
class UnstrippedPythonSources:
"""Sources that can be introspected by Python, relative to a set of source roots.
Specifically, this will filter out to only have Python, and, optionally, resources() and
files() targets; and will add any missing `__init__.py` files to ensure that modules are
recognized correctly.
Use-cases that introspect Python source code (e.g., the `test, `lint`, `fmt` goals) can
request this type to get relevant sources that are still relative to their source roots.
That way the paths they report are the unstripped ones the user is familiar with.
The sources can also be imported and used by Python (e.g., for the `test` goal), but only
if sys.path is modified to include the source roots.
"""
snapshot: Snapshot
source_roots: Tuple[str, ...]
class UnstrippedPythonSourcesRequest(_BasePythonSourcesRequest):
pass
@rule
async def prepare_stripped_python_sources(
request: StrippedPythonSourcesRequest,
) -> StrippedPythonSources:
stripped_sources = await Get(
SourceFiles,
AllSourceFilesRequest(
(tgt.get(Sources) for tgt in request.targets),
for_sources_types=request.valid_sources_types,
enable_codegen=True,
strip_source_roots=True,
),
)
return StrippedPythonSources(stripped_sources.snapshot)
@rule
async def prepare_unstripped_python_sources(
request: UnstrippedPythonSourcesRequest,
) -> UnstrippedPythonSources:
sources = await Get(
SourceFiles,
AllSourceFilesRequest(
(tgt.get(Sources) for tgt in request.targets),
for_sources_types=request.valid_sources_types,
enable_codegen=True,
strip_source_roots=False,
),
)
source_root_objs = await MultiGet(
Get(
SourceRoot,
SourceRootRequest,
SourceRootRequest.for_file(representative_path_from_address(tgt.address)),
)
for tgt in request.targets
if tgt.has_field(PythonSources) or tgt.has_field(ResourcesSources)
)
source_root_paths = {source_root_obj.path for source_root_obj in source_root_objs}
return UnstrippedPythonSources(sources.snapshot, tuple(sorted(source_root_paths)))
def rules():
return [
prepare_stripped_python_sources,
prepare_unstripped_python_sources,
*determine_source_files.rules(),
# TODO: remove this as soon as we have a usage of it.
RootRule(UnstrippedPythonSourcesRequest),
]