-
-
Notifications
You must be signed in to change notification settings - Fork 607
/
rules.py
147 lines (129 loc) · 5.13 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
# 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 Tuple
from pants.backend.python.rules import download_pex_bin, pex, python_sources
from pants.backend.python.rules.pex import (
Pex,
PexInterpreterConstraints,
PexRequest,
PexRequirements,
)
from pants.backend.python.rules.python_sources import (
UnstrippedPythonSources,
UnstrippedPythonSourcesRequest,
)
from pants.backend.python.subsystems import python_native_code, subprocess_environment
from pants.backend.python.subsystems.subprocess_environment import SubprocessEncodingEnvironment
from pants.backend.python.target_types import PythonSources
from pants.backend.python.typecheck.mypy.subsystem import MyPy
from pants.core.goals.typecheck import TypecheckRequest, TypecheckResult, TypecheckResults
from pants.core.util_rules import determine_source_files, pants_bin, strip_source_roots
from pants.engine.addresses import Addresses
from pants.engine.fs import (
CreateDigest,
Digest,
FileContent,
GlobMatchErrorBehavior,
MergeDigests,
PathGlobs,
)
from pants.engine.process import FallibleProcessResult, Process
from pants.engine.rules import collect_rules, rule
from pants.engine.selectors import Get, MultiGet
from pants.engine.target import FieldSetWithOrigin, TransitiveTargets
from pants.engine.unions import UnionRule
from pants.python.python_setup import PythonSetup
from pants.util.strutil import pluralize
@dataclass(frozen=True)
class MyPyFieldSet(FieldSetWithOrigin):
required_fields = (PythonSources,)
sources: PythonSources
class MyPyRequest(TypecheckRequest):
field_set_type = MyPyFieldSet
def generate_args(mypy: MyPy, *, file_list_path: str) -> Tuple[str, ...]:
args = []
if mypy.config:
args.append(f"--config-file={mypy.config}")
args.extend(mypy.args)
args.append(f"@{file_list_path}")
return tuple(args)
# TODO(#10131): Improve performance, e.g. by leveraging the MyPy cache.
# TODO(#10131): Support plugins and type stubs.
@rule(desc="Lint using MyPy")
async def mypy_lint(
request: MyPyRequest,
mypy: MyPy,
python_setup: PythonSetup,
subprocess_encoding_environment: SubprocessEncodingEnvironment,
) -> TypecheckResults:
if mypy.skip:
return TypecheckResults()
transitive_targets = await Get(
TransitiveTargets, Addresses(fs.address for fs in request.field_sets)
)
prepared_sources_request = Get(
UnstrippedPythonSources, UnstrippedPythonSourcesRequest(transitive_targets.closure),
)
pex_request = Get(
Pex,
PexRequest(
output_filename="mypy.pex",
requirements=PexRequirements(mypy.get_requirement_specs()),
# NB: This only determines what MyPy is run with. The user can specify what version
# their code is with `--python-version`. See
# https://mypy.readthedocs.io/en/stable/config_file.html#platform-configuration. We do
# not auto-configure this for simplicity and to avoid Pants magically setting values for
# users.
interpreter_constraints=PexInterpreterConstraints(mypy.default_interpreter_constraints),
entry_point=mypy.get_entry_point(),
),
)
config_digest_request = Get(
Digest,
PathGlobs(
globs=[mypy.config] if mypy.config else [],
glob_match_error_behavior=GlobMatchErrorBehavior.error,
description_of_origin="the option `--mypy-config`",
),
)
prepared_sources, pex, config_digest = await MultiGet(
prepared_sources_request, pex_request, config_digest_request
)
file_list_path = "__files.txt"
python_files = "\n".join(f for f in prepared_sources.snapshot.files if f.endswith(".py"))
file_list_digest = await Get(
Digest, CreateDigest([FileContent(file_list_path, python_files.encode())]),
)
merged_input_files = await Get(
Digest,
MergeDigests(
[file_list_digest, prepared_sources.snapshot.digest, pex.digest, config_digest]
),
)
process = pex.create_process(
python_setup=python_setup,
subprocess_encoding_environment=subprocess_encoding_environment,
pex_path=pex.output_filename,
pex_args=generate_args(mypy, file_list_path=file_list_path),
input_digest=merged_input_files,
env={"PEX_EXTRA_SYS_PATH": ":".join(prepared_sources.source_roots)},
description=f"Run MyPy on {pluralize(len(prepared_sources.snapshot.files), 'file')}.",
)
result = await Get(FallibleProcessResult, Process, process)
return TypecheckResults(
[TypecheckResult.from_fallible_process_result(result, typechecker_name="MyPy")]
)
def rules():
return [
*collect_rules(),
UnionRule(TypecheckRequest, MyPyRequest),
*download_pex_bin.rules(),
*determine_source_files.rules(),
*pants_bin.rules(),
*pex.rules(),
*python_sources.rules(),
*python_native_code.rules(),
*strip_source_roots.rules(),
*subprocess_environment.rules(),
]