-
-
Notifications
You must be signed in to change notification settings - Fork 607
/
adhoc_binaries.py
166 lines (131 loc) · 5.71 KB
/
adhoc_binaries.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
# Copyright 2023 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).
from __future__ import annotations
import os
import sys
from dataclasses import dataclass
from textwrap import dedent # noqa: PNT20
from pants.core.subsystems.python_bootstrap import PythonBootstrapSubsystem
from pants.core.util_rules.environments import EnvironmentTarget, LocalEnvironmentTarget
from pants.core.util_rules.system_binaries import SEARCH_PATHS, BashBinary, TarBinary
from pants.engine.env_vars import EnvironmentVars, EnvironmentVarsRequest
from pants.engine.fs import DownloadFile
from pants.engine.internals.native_engine import Digest, FileDigest
from pants.engine.internals.selectors import Get
from pants.engine.platform import Platform
from pants.engine.process import Process, ProcessCacheScope, ProcessResult
from pants.engine.rules import collect_rules, rule
from pants.util.frozendict import FrozenDict
from pants.util.logging import LogLevel
@dataclass(frozen=True)
class PythonBuildStandaloneBinary:
"""A Python interpreter for use by `@rule` code as an alternative to BashBinary scripts.
This interpreter is provided by Python Build Standalone https://gregoryszorc.com/docs/python-build-standalone/main/,
which has a few caveats. Namely it doesn't play nicely with third-party sdists. Meaning Pants'
scripts being run by Python Build Standalone should avoid third-party sdists.
"""
_CACHE_DIRNAME = "python_build_standalone"
_SYMLINK_DIRNAME = ".python-build-standalone"
APPEND_ONLY_CACHES = FrozenDict({_CACHE_DIRNAME: _SYMLINK_DIRNAME})
path: str # The absolute path to a Python executable
# NB: These private types are solely so we can test the docker-path using the local
# environment.
class _PythonBuildStandaloneBinary(PythonBuildStandaloneBinary):
pass
class _DownloadPythonBuildStandaloneBinaryRequest:
pass
@rule
async def get_python_for_scripts(env_tgt: EnvironmentTarget) -> PythonBuildStandaloneBinary:
if env_tgt.val is None or isinstance(env_tgt.val, LocalEnvironmentTarget):
return PythonBuildStandaloneBinary(sys.executable)
result = await Get(_PythonBuildStandaloneBinary, _DownloadPythonBuildStandaloneBinaryRequest())
return PythonBuildStandaloneBinary(result.path)
@rule(desc="Downloading Python for scripts", level=LogLevel.TRACE)
async def download_python_binary(
_: _DownloadPythonBuildStandaloneBinaryRequest,
platform: Platform,
tar_binary: TarBinary,
python_bootstrap: PythonBootstrapSubsystem,
bash: BashBinary,
) -> _PythonBuildStandaloneBinary:
url, fingerprint, bytelen = python_bootstrap.internal_python_build_standalone_info[
platform.value
]
filename = url.rsplit("/", 1)[-1]
python_archive = await Get(
Digest,
DownloadFile(
url,
FileDigest(
fingerprint=fingerprint,
serialized_bytes_length=bytelen,
),
),
)
download_result = await Get(
ProcessResult,
Process(
argv=[tar_binary.path, "-xvf", filename],
input_digest=python_archive,
env={"PATH": os.pathsep.join(SEARCH_PATHS)},
description="Extract Pants' execution Python",
level=LogLevel.DEBUG,
output_directories=("python",),
),
)
installation_root = f"{PythonBuildStandaloneBinary._SYMLINK_DIRNAME}/{download_result.output_digest.fingerprint}"
# NB: This is similar to what we do for every Python provider. We should refactor these into
# some shared code to centralize the behavior.
installation_script = dedent(
f"""\
if [ ! -f "{installation_root}/DONE" ]; then
cp -r python "{installation_root}"
touch "{installation_root}/DONE"
fi
"""
)
env_vars = await Get(EnvironmentVars, EnvironmentVarsRequest(["PATH"]))
await Get(
ProcessResult,
Process(
[bash.path, "-c", installation_script],
level=LogLevel.DEBUG,
input_digest=download_result.output_digest,
description="Install Python for Pants usage",
env={"PATH": env_vars.get("PATH", "")},
append_only_caches=PythonBuildStandaloneBinary.APPEND_ONLY_CACHES,
# Don't cache, we want this to always be run so that we can assume for the rest of the
# session the named_cache destination for this Python is valid, as the Python ecosystem
# mainly assumes absolute paths for Python interpreters.
cache_scope=ProcessCacheScope.PER_SESSION,
),
)
return _PythonBuildStandaloneBinary(f"{installation_root}/bin/python3")
@dataclass(frozen=True)
class GunzipBinaryRequest:
pass
@dataclass(frozen=True)
class GunzipBinary:
python_binary: PythonBuildStandaloneBinary
def extract_archive_argv(self, archive_path: str, extract_path: str) -> tuple[str, ...]:
archive_name = os.path.basename(archive_path)
dest_file_name = os.path.splitext(archive_name)[0]
dest_path = os.path.join(extract_path, dest_file_name)
script = dedent(
f"""
import gzip
import shutil
with gzip.GzipFile(filename={archive_path!r}, mode="rb") as source:
with open({dest_path!r}, "wb") as dest:
shutil.copyfileobj(source, dest)
"""
)
return (self.python_binary.path, "-c", script)
@rule
def find_gunzip(python_binary: PythonBuildStandaloneBinary) -> GunzipBinary:
return GunzipBinary(python_binary)
@rule
async def find_gunzip_wrapper(_: GunzipBinaryRequest, gunzip: GunzipBinary) -> GunzipBinary:
return gunzip
def rules():
return collect_rules()