-
-
Notifications
You must be signed in to change notification settings - Fork 607
/
target_types.py
253 lines (217 loc) · 9.34 KB
/
target_types.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
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).
import os.path
import re
from dataclasses import dataclass
from enum import Enum
from typing import Match, Optional, Tuple, cast
from pants.backend.python.dependency_inference.module_mapper import (
PythonModuleOwners,
PythonModuleOwnersRequest,
)
from pants.backend.python.dependency_inference.rules import PythonInferSubsystem, import_rules
from pants.backend.python.subsystems.setup import PythonSetup
from pants.backend.python.target_types import PythonResolveField
from pants.core.goals.package import OutputPathField
from pants.engine.addresses import Address
from pants.engine.fs import GlobMatchErrorBehavior, PathGlobs, Paths
from pants.engine.rules import Get, MultiGet, collect_rules, rule
from pants.engine.target import (
COMMON_TARGET_FIELDS,
AsyncFieldMixin,
Dependencies,
DependenciesRequest,
ExplicitlyProvidedDependencies,
InjectDependenciesRequest,
InjectedDependencies,
InvalidFieldException,
SecondaryOwnerMixin,
StringField,
Target,
WrappedTarget,
)
from pants.engine.unions import UnionRule
from pants.source.filespec import Filespec
from pants.source.source_root import SourceRoot, SourceRootRequest
from pants.util.docutil import doc_url
class PythonGoogleCloudFunctionHandlerField(StringField, AsyncFieldMixin, SecondaryOwnerMixin):
alias = "handler"
required = True
value: str
help = (
"Entry point to the Google Cloud Function handler.\n\nYou can specify a full module like "
"'path.to.module:handler_func' or use a shorthand to specify a file name, using the same "
"syntax as the `sources` field, e.g. 'cloud_function.py:handler_func'.\n\nYou must use the file "
"name shorthand for file arguments to work with this target."
)
@classmethod
def compute_value(cls, raw_value: Optional[str], address: Address) -> str:
value = cast(str, super().compute_value(raw_value, address))
if ":" not in value:
raise InvalidFieldException(
f"The `{cls.alias}` field in target at {address} must end in the "
f"format `:my_handler_func`, but was {value}."
)
return value
@property
def filespec(self) -> Filespec:
path, _, func = self.value.partition(":")
if not path.endswith(".py"):
return {"includes": []}
full_glob = os.path.join(self.address.spec_path, path)
return {"includes": [full_glob]}
@dataclass(frozen=True)
class ResolvedPythonGoogleHandler:
val: str
file_name_used: bool
@dataclass(frozen=True)
class ResolvePythonGoogleHandlerRequest:
field: PythonGoogleCloudFunctionHandlerField
@rule(desc="Determining the handler for a `python_google_cloud_function` target")
async def resolve_python_google_cloud_function_handler(
request: ResolvePythonGoogleHandlerRequest,
) -> ResolvedPythonGoogleHandler:
handler_val = request.field.value
field_alias = request.field.alias
address = request.field.address
path, _, func = handler_val.partition(":")
# If it's already a module, simply use that. Otherwise, convert the file name into a module
# path.
if not path.endswith(".py"):
return ResolvedPythonGoogleHandler(handler_val, file_name_used=False)
# Use the engine to validate that the file exists and that it resolves to only one file.
full_glob = os.path.join(address.spec_path, path)
handler_paths = await Get(
Paths,
PathGlobs(
[full_glob],
glob_match_error_behavior=GlobMatchErrorBehavior.error,
description_of_origin=f"{address}'s `{field_alias}` field",
),
)
# We will have already raised if the glob did not match, i.e. if there were no files. But
# we need to check if they used a file glob (`*` or `**`) that resolved to >1 file.
if len(handler_paths.files) != 1:
raise InvalidFieldException(
f"Multiple files matched for the `{field_alias}` {repr(handler_val)} for the target "
f"{address}, but only one file expected. Are you using a glob, rather than a file "
f"name?\n\nAll matching files: {list(handler_paths.files)}."
)
handler_path = handler_paths.files[0]
source_root = await Get(
SourceRoot,
SourceRootRequest,
SourceRootRequest.for_file(handler_path),
)
stripped_source_path = os.path.relpath(handler_path, source_root.path)
module_base, _ = os.path.splitext(stripped_source_path)
normalized_path = module_base.replace(os.path.sep, ".")
return ResolvedPythonGoogleHandler(f"{normalized_path}:{func}", file_name_used=True)
class PythonGoogleCloudFunctionDependencies(Dependencies):
supports_transitive_excludes = True
class InjectPythonCloudFunctionHandlerDependency(InjectDependenciesRequest):
inject_for = PythonGoogleCloudFunctionDependencies
@rule(desc="Inferring dependency from the python_google_cloud_function `handler` field")
async def inject_cloud_function_handler_dependency(
request: InjectPythonCloudFunctionHandlerDependency,
python_infer_subsystem: PythonInferSubsystem,
python_setup: PythonSetup,
) -> InjectedDependencies:
if not python_infer_subsystem.entry_points:
return InjectedDependencies()
original_tgt = await Get(WrappedTarget, Address, request.dependencies_field.address)
explicitly_provided_deps, handler = await MultiGet(
Get(ExplicitlyProvidedDependencies, DependenciesRequest(original_tgt.target[Dependencies])),
Get(
ResolvedPythonGoogleHandler,
ResolvePythonGoogleHandlerRequest(
original_tgt.target[PythonGoogleCloudFunctionHandlerField]
),
),
)
module, _, _func = handler.val.partition(":")
owners = await Get(
PythonModuleOwners,
PythonModuleOwnersRequest(
module, resolve=original_tgt.target[PythonResolveField].normalized_value(python_setup)
),
)
address = original_tgt.target.address
explicitly_provided_deps.maybe_warn_of_ambiguous_dependency_inference(
owners.ambiguous,
address,
# If the handler was specified as a file, like `app.py`, we know the module must
# live in the python_google_cloud_function's directory or subdirectory, so the owners must be ancestors.
owners_must_be_ancestors=handler.file_name_used,
import_reference="module",
context=(
f"The python_google_cloud_function target {address} has the field "
f"`handler={repr(original_tgt.target[PythonGoogleCloudFunctionHandlerField].value)}`, which maps "
f"to the Python module `{module}`"
),
)
maybe_disambiguated = explicitly_provided_deps.disambiguated(
owners.ambiguous, owners_must_be_ancestors=handler.file_name_used
)
unambiguous_owners = owners.unambiguous or (
(maybe_disambiguated,) if maybe_disambiguated else ()
)
return InjectedDependencies(unambiguous_owners)
class PythonGoogleCloudFunctionRuntimes(Enum):
PYTHON_37 = "python37"
PYTHON_38 = "python38"
PYTHON_39 = "python39"
class PythonGoogleCloudFunctionRuntime(StringField):
PYTHON_RUNTIME_REGEX = r"^python(?P<major>\d)(?P<minor>\d+)$"
alias = "runtime"
required = True
valid_choices = PythonGoogleCloudFunctionRuntimes
help = (
"The identifier of the Google Cloud Function runtime to target (pythonXY). See "
"https://cloud.google.com/functions/docs/concepts/python-runtime."
)
@classmethod
def compute_value(cls, raw_value: Optional[str], address: Address) -> str:
value = cast(str, super().compute_value(raw_value, address))
if not re.match(cls.PYTHON_RUNTIME_REGEX, value):
raise InvalidFieldException(
f"The `{cls.alias}` field in target at {address} must be of the form pythonXY, "
f"but was {value}."
)
return value
def to_interpreter_version(self) -> Tuple[int, int]:
"""Returns the Python version implied by the runtime, as (major, minor)."""
mo = cast(Match, re.match(self.PYTHON_RUNTIME_REGEX, self.value))
return int(mo.group("major")), int(mo.group("minor"))
class GoogleCloudFunctionTypes(Enum):
EVENT = "event"
HTTP = "http"
class PythonGoogleCloudFunctionType(StringField):
alias = "type"
required = True
valid_choices = GoogleCloudFunctionTypes
help = (
"The trigger type of the cloud function. Can either be 'event' or 'http'. "
"See https://cloud.google.com/functions/docs/concepts/python-runtime for reference to --trigger-http."
)
class PythonGoogleCloudFunction(Target):
alias = "python_google_cloud_function"
core_fields = (
*COMMON_TARGET_FIELDS,
OutputPathField,
PythonGoogleCloudFunctionDependencies,
PythonGoogleCloudFunctionHandlerField,
PythonGoogleCloudFunctionRuntime,
PythonGoogleCloudFunctionType,
PythonResolveField,
)
help = (
"A self-contained Python function suitable for uploading to Google Cloud Function.\n\n"
f"See {doc_url('python-google-cloud-function')}."
)
def rules():
return (
*collect_rules(),
*import_rules(),
UnionRule(InjectDependenciesRequest, InjectPythonCloudFunctionHandlerDependency),
)