1import json
2import os
3import sys
4import tempfile
5import subprocess
6from contextlib import contextmanager
7from os.path import abspath
8from os.path import join as pjoin
9from typing import TYPE_CHECKING, Any, Iterator, Mapping, Optional, Sequence
10import warnings
11
12from ._in_process import _in_proc_script_path
13
14if TYPE_CHECKING:
15 from typing import Protocol
16
17 class SubprocessRunner(Protocol):
18 """A protocol for the subprocess runner."""
19
20 def __call__(
21 self,
22 cmd: Sequence[str],
23 cwd: Optional[str] = None,
24 extra_environ: Optional[Mapping[str, str]] = None,
25 ) -> None:
26 ...
27
28
29def write_json(obj: Mapping[str, Any], path: str, **kwargs) -> None:
30 with open(path, "w", encoding="utf-8") as f:
31 json.dump(obj, f, **kwargs)
32
33
34def read_json(path: str) -> Mapping[str, Any]:
35 with open(path, encoding="utf-8") as f:
36 return json.load(f)
37
38
39class BuildBackendWarning(UserWarning):
40 """Will be emitted for every UserWarning emitted by the hook process."""
41
42
43class BackendUnavailable(Exception):
44 """Will be raised if the backend cannot be imported in the hook process."""
45
46 def __init__(
47 self,
48 traceback: str,
49 message: Optional[str] = None,
50 backend_name: Optional[str] = None,
51 backend_path: Optional[Sequence[str]] = None,
52 ) -> None:
53 # Preserving arg order for the sake of API backward compatibility.
54 self.backend_name = backend_name
55 self.backend_path = backend_path
56 self.traceback = traceback or ""
57 super().__init__(
58 f"{message or 'Error while importing backend'}"
59 + (f"\n\nTraceback:\n{self.traceback}" if self.traceback else "")
60 )
61
62
63class HookMissing(Exception):
64 """Will be raised on missing hooks (if a fallback can't be used)."""
65
66 def __init__(self, hook_name: str) -> None:
67 super().__init__(hook_name)
68 self.hook_name = hook_name
69
70
71class UnsupportedOperation(Exception):
72 """May be raised by build_sdist if the backend indicates that it can't."""
73
74 def __init__(self, traceback: str) -> None:
75 self.traceback = traceback
76
77
78def default_subprocess_runner(
79 cmd: Sequence[str],
80 cwd: Optional[str] = None,
81 extra_environ: Optional[Mapping[str, str]] = None,
82) -> None:
83 """The default method of calling the wrapper subprocess.
84
85 This uses :func:`subprocess.run` under the hood.
86 """
87 env = os.environ.copy()
88 if extra_environ:
89 env.update(extra_environ)
90
91 subprocess.run(cmd, cwd=cwd, env=env, check=True)
92
93
94def quiet_subprocess_runner(
95 cmd: Sequence[str],
96 cwd: Optional[str] = None,
97 extra_environ: Optional[Mapping[str, str]] = None,
98) -> None:
99 """Call the subprocess while suppressing output.
100
101 This uses :func:`subprocess.run` with `stdout=PIPE, stderr=STDOUT` under the hood.
102 """
103 env = os.environ.copy()
104 if extra_environ:
105 env.update(extra_environ)
106
107 # Capturing output in case it fails (available on the exception object)
108 subprocess.run(
109 cmd,
110 cwd=cwd,
111 env=env,
112 stdout=subprocess.PIPE,
113 stderr=subprocess.STDOUT,
114 check=True,
115 )
116
117
118def norm_and_check(source_tree: str, requested: str) -> str:
119 """Normalise and check a backend path.
120
121 Ensure that the requested backend path is specified as a relative path,
122 and resolves to a location under the given source tree.
123
124 Return an absolute version of the requested path.
125 """
126 if os.path.isabs(requested):
127 raise ValueError("paths must be relative")
128
129 abs_source = os.path.abspath(source_tree)
130 abs_requested = os.path.normpath(os.path.join(abs_source, requested))
131 norm_source = os.path.normcase(abs_source)
132 norm_requested = os.path.normcase(abs_requested)
133 if os.path.commonpath([norm_source, norm_requested]) != norm_source:
134 raise ValueError("paths must be inside source tree")
135
136 return abs_requested
137
138
139class BuildBackendHookCaller:
140 """A wrapper to call the build backend hooks for a source directory."""
141
142 def __init__(
143 self,
144 source_dir: str,
145 build_backend: str,
146 backend_path: Optional[Sequence[str]] = None,
147 runner: Optional["SubprocessRunner"] = None,
148 python_executable: Optional[str] = None,
149 ) -> None:
150 """
151 :param source_dir: The source directory to invoke the build backend for
152 :param build_backend: The build backend spec
153 :param backend_path: Additional path entries for the build backend spec
154 :param runner: The :ref:`subprocess runner <Subprocess Runners>` to use
155 :param python_executable:
156 The Python executable used to invoke the build backend
157 """
158 if runner is None:
159 runner = default_subprocess_runner
160
161 self.source_dir = abspath(source_dir)
162 self.build_backend = build_backend
163 if backend_path:
164 backend_path = [norm_and_check(self.source_dir, p) for p in backend_path]
165 self.backend_path = backend_path
166 self._subprocess_runner = runner
167 if not python_executable:
168 python_executable = sys.executable
169 self.python_executable = python_executable
170
171 @contextmanager
172 def subprocess_runner(self, runner: "SubprocessRunner") -> Iterator[None]:
173 """A context manager for temporarily overriding the default
174 :ref:`subprocess runner <Subprocess Runners>`.
175
176 :param runner: The new subprocess runner to use within the context.
177
178 .. warning::
179
180 This context manager temporarily mutates the hook caller instance and
181 is not thread-safe. Callers that need to run hooks concurrently
182 should create separate :class:`BuildBackendHookCaller` instances, or
183 provide the subprocess runner when constructing the caller.
184
185 .. code-block:: python
186
187 hook_caller = BuildBackendHookCaller(...)
188 with hook_caller.subprocess_runner(quiet_subprocess_runner):
189 ...
190 """
191 prev = self._subprocess_runner
192 self._subprocess_runner = runner
193 try:
194 yield
195 finally:
196 self._subprocess_runner = prev
197
198 def _supported_features(self) -> Sequence[str]:
199 """Return the list of optional features supported by the backend."""
200 return self._call_hook("_supported_features", {})
201
202 def get_requires_for_build_wheel(
203 self,
204 config_settings: Optional[Mapping[str, Any]] = None,
205 ) -> Sequence[str]:
206 """Get additional dependencies required for building a wheel.
207
208 :param config_settings: The configuration settings for the build backend
209 :returns: A list of :pep:`dependency specifiers <508>`.
210
211 .. admonition:: Fallback
212
213 If the build backend does not defined a hook with this name, an
214 empty list will be returned.
215 """
216 return self._call_hook(
217 "get_requires_for_build_wheel", {"config_settings": config_settings}
218 )
219
220 def prepare_metadata_for_build_wheel(
221 self,
222 metadata_directory: str,
223 config_settings: Optional[Mapping[str, Any]] = None,
224 _allow_fallback: bool = True,
225 ) -> str:
226 """Prepare a ``*.dist-info`` folder with metadata for this project.
227
228 :param metadata_directory: The directory to write the metadata to
229 :param config_settings: The configuration settings for the build backend
230 :param _allow_fallback:
231 Whether to allow the fallback to building a wheel and extracting
232 the metadata from it. Should be passed as a keyword argument only.
233
234 :returns: Name of the newly created subfolder within
235 ``metadata_directory``, containing the metadata.
236
237 .. admonition:: Fallback
238
239 If the build backend does not define a hook with this name and
240 ``_allow_fallback`` is truthy, the backend will be asked to build a
241 wheel via the ``build_wheel`` hook and the dist-info extracted from
242 that will be returned.
243 """
244 return self._call_hook(
245 "prepare_metadata_for_build_wheel",
246 {
247 "metadata_directory": abspath(metadata_directory),
248 "config_settings": config_settings,
249 "_allow_fallback": _allow_fallback,
250 },
251 )
252
253 def build_wheel(
254 self,
255 wheel_directory: str,
256 config_settings: Optional[Mapping[str, Any]] = None,
257 metadata_directory: Optional[str] = None,
258 ) -> str:
259 """Build a wheel from this project.
260
261 :param wheel_directory: The directory to write the wheel to
262 :param config_settings: The configuration settings for the build backend
263 :param metadata_directory: The directory to reuse existing metadata from
264 :returns:
265 The name of the newly created wheel within ``wheel_directory``.
266
267 .. admonition:: Interaction with fallback
268
269 If the ``build_wheel`` hook was called in the fallback for
270 :meth:`prepare_metadata_for_build_wheel`, the build backend would
271 not be invoked. Instead, the previously built wheel will be copied
272 to ``wheel_directory`` and the name of that file will be returned.
273 """
274 if metadata_directory is not None:
275 metadata_directory = abspath(metadata_directory)
276 return self._call_hook(
277 "build_wheel",
278 {
279 "wheel_directory": abspath(wheel_directory),
280 "config_settings": config_settings,
281 "metadata_directory": metadata_directory,
282 },
283 )
284
285 def get_requires_for_build_editable(
286 self,
287 config_settings: Optional[Mapping[str, Any]] = None,
288 ) -> Sequence[str]:
289 """Get additional dependencies required for building an editable wheel.
290
291 :param config_settings: The configuration settings for the build backend
292 :returns: A list of :pep:`dependency specifiers <508>`.
293
294 .. admonition:: Fallback
295
296 If the build backend does not defined a hook with this name, an
297 empty list will be returned.
298 """
299 return self._call_hook(
300 "get_requires_for_build_editable", {"config_settings": config_settings}
301 )
302
303 def prepare_metadata_for_build_editable(
304 self,
305 metadata_directory: str,
306 config_settings: Optional[Mapping[str, Any]] = None,
307 _allow_fallback: bool = True,
308 ) -> str:
309 """Prepare a ``*.dist-info`` folder with metadata for this project.
310
311 :param metadata_directory: The directory to write the metadata to
312 :param config_settings: The configuration settings for the build backend
313 :param _allow_fallback:
314 Whether to allow the fallback to building a wheel and extracting
315 the metadata from it. Should be passed as a keyword argument only.
316 :returns: Name of the newly created subfolder within
317 ``metadata_directory``, containing the metadata.
318
319 .. admonition:: Fallback
320
321 If the build backend does not define a hook with this name and
322 ``_allow_fallback`` is truthy, the backend will be asked to build a
323 wheel via the ``build_editable`` hook and the dist-info
324 extracted from that will be returned.
325 """
326 return self._call_hook(
327 "prepare_metadata_for_build_editable",
328 {
329 "metadata_directory": abspath(metadata_directory),
330 "config_settings": config_settings,
331 "_allow_fallback": _allow_fallback,
332 },
333 )
334
335 def build_editable(
336 self,
337 wheel_directory: str,
338 config_settings: Optional[Mapping[str, Any]] = None,
339 metadata_directory: Optional[str] = None,
340 ) -> str:
341 """Build an editable wheel from this project.
342
343 :param wheel_directory: The directory to write the wheel to
344 :param config_settings: The configuration settings for the build backend
345 :param metadata_directory: The directory to reuse existing metadata from
346 :returns:
347 The name of the newly created wheel within ``wheel_directory``.
348
349 .. admonition:: Interaction with fallback
350
351 If the ``build_editable`` hook was called in the fallback for
352 :meth:`prepare_metadata_for_build_editable`, the build backend
353 would not be invoked. Instead, the previously built wheel will be
354 copied to ``wheel_directory`` and the name of that file will be
355 returned.
356 """
357 if metadata_directory is not None:
358 metadata_directory = abspath(metadata_directory)
359 return self._call_hook(
360 "build_editable",
361 {
362 "wheel_directory": abspath(wheel_directory),
363 "config_settings": config_settings,
364 "metadata_directory": metadata_directory,
365 },
366 )
367
368 def get_requires_for_build_sdist(
369 self,
370 config_settings: Optional[Mapping[str, Any]] = None,
371 ) -> Sequence[str]:
372 """Get additional dependencies required for building an sdist.
373
374 :returns: A list of :pep:`dependency specifiers <508>`.
375 """
376 return self._call_hook(
377 "get_requires_for_build_sdist", {"config_settings": config_settings}
378 )
379
380 def build_sdist(
381 self,
382 sdist_directory: str,
383 config_settings: Optional[Mapping[str, Any]] = None,
384 ) -> str:
385 """Build an sdist from this project.
386
387 :returns:
388 The name of the newly created sdist within ``sdist_directory``.
389 """
390 return self._call_hook(
391 "build_sdist",
392 {
393 "sdist_directory": abspath(sdist_directory),
394 "config_settings": config_settings,
395 },
396 )
397
398 def _call_hook(self, hook_name: str, kwargs: Mapping[str, Any]) -> Any:
399 extra_environ = {"_PYPROJECT_HOOKS_BUILD_BACKEND": self.build_backend}
400
401 if self.backend_path:
402 extra_environ["_PYPROJECT_HOOKS_BACKEND_PATH_JSON"] = json.dumps(
403 self.backend_path
404 )
405
406 with tempfile.TemporaryDirectory() as td:
407 hook_input = {"kwargs": kwargs}
408 write_json(hook_input, pjoin(td, "input.json"), indent=2)
409
410 # Run the hook in a subprocess
411 with _in_proc_script_path() as script:
412 python = self.python_executable
413 self._subprocess_runner(
414 [python, abspath(str(script)), hook_name, td],
415 cwd=self.source_dir,
416 extra_environ=extra_environ,
417 )
418
419 data = read_json(pjoin(td, "output.json"))
420 if data.get("unsupported"):
421 raise UnsupportedOperation(data.get("traceback", ""))
422 if data.get("no_backend"):
423 raise BackendUnavailable(
424 data.get("traceback", ""),
425 message=data.get("backend_error", ""),
426 backend_name=self.build_backend,
427 backend_path=self.backend_path,
428 )
429 if data.get("hook_missing"):
430 raise HookMissing(data.get("missing_hook_name") or hook_name)
431
432 for w in data.get("warnings", []):
433 warnings.warn_explicit(
434 message=w["message"],
435 category=BuildBackendWarning,
436 filename=w["filename"],
437 lineno=w["lineno"],
438 )
439 return data["return_val"]