Repository navigation
Expand file tree
/
Copy pathuv.py
More file actions
120 lines (96 loc) · 3.7 KB
/
Copy pathuv.py
File metadata and controls
120 lines (96 loc) · 3.7 KB
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
"""uv detection, installation, and `uv sync` execution."""
from __future__ import annotations
import os
import platform
import shutil
import subprocess
import urllib.request
from pathlib import Path
from . import strings
from .errors import UvError
UV_UNIX_INSTALLER = "https://astral.sh/uv/install.sh"
UV_WINDOWS_INSTALLER = "https://astral.sh/uv/install.ps1"
def find_uv() -> str | None:
found = shutil.which("uv")
if found:
return found
for candidate in _common_uv_paths():
if candidate.exists():
return str(candidate)
return None
def execute_uv_install(*, verbose: bool = False) -> str:
install_uv(verbose=verbose)
uv = find_uv()
if uv:
return uv
raise UvError(strings.ERROR_UV_NOT_ON_PATH)
def install_uv(*, verbose: bool = False) -> None:
if platform.system().lower() == "windows":
_run_windows_installer(verbose=verbose)
else:
_run_unix_installer(verbose=verbose)
def run_uv_sync(uv_executable: str, project_dir: Path, *, verbose: bool = False) -> None:
_run([uv_executable, "sync"], cwd=project_dir, isolated_project=True, verbose=verbose)
def _run_unix_installer(*, verbose: bool = False) -> None:
try:
with urllib.request.urlopen(UV_UNIX_INSTALLER, timeout=30) as response:
script = response.read()
except OSError as exc:
raise UvError(strings.ERROR_UV_INSTALLER_FETCH.format(reason=exc)) from exc
_run(["sh"], input_bytes=script, verbose=verbose)
def _run_windows_installer(*, verbose: bool = False) -> None:
powershell = shutil.which("powershell") or shutil.which("pwsh")
if not powershell:
raise UvError(strings.ERROR_UV_NEEDS_POWERSHELL)
try:
with urllib.request.urlopen(UV_WINDOWS_INSTALLER, timeout=30) as response:
script = response.read()
except OSError as exc:
raise UvError(strings.ERROR_UV_INSTALLER_FETCH.format(reason=exc)) from exc
_run([powershell, "-ExecutionPolicy", "ByPass", "-Command", "-"], input_bytes=script, verbose=verbose)
def _run(
command: list[str],
*,
cwd: Path | None = None,
input_bytes: bytes | None = None,
isolated_project: bool = False,
verbose: bool = False,
) -> None:
try:
subprocess.run(
command,
cwd=cwd,
env=_isolated_project_env() if isolated_project else None,
input=input_bytes,
check=True,
capture_output=not verbose,
)
except FileNotFoundError as exc:
raise UvError(strings.ERROR_UV_COMMAND_NOT_FOUND.format(cmd=command[0])) from exc
except subprocess.CalledProcessError as exc:
joined = " ".join(command)
message = strings.ERROR_UV_COMMAND_FAILED.format(returncode=exc.returncode, cmd=joined)
if not verbose:
captured = _format_captured(exc.stderr, exc.stdout)
if captured:
message = f"{message}\n\n{captured}"
raise UvError(message) from exc
def _format_captured(stderr: bytes | None, stdout: bytes | None) -> str:
parts: list[str] = []
for stream in (stderr, stdout):
if not stream:
continue
decoded = stream.decode("utf-8", errors="replace").rstrip()
if decoded:
parts.append(decoded)
return "\n\n".join(parts)
def _common_uv_paths() -> tuple[Path, ...]:
home = Path.home()
if os.name == "nt":
return (home / ".local" / "bin" / "uv.exe", home / ".cargo" / "bin" / "uv.exe")
return (home / ".local" / "bin" / "uv", home / ".cargo" / "bin" / "uv")
def _isolated_project_env() -> dict[str, str]:
env = os.environ.copy()
for name in ("VIRTUAL_ENV", "CONDA_PREFIX", "PYTHONPATH"):
env.pop(name, None)
return env