"""``udb`` CLI launcher bundled with the UDB Python SDK.
{{GENERATED_NOTE}}
Installing ``udb-client`` exposes a console script named ``udb`` (wired in
``pyproject.toml`` as ``udb = "udb_client._cli:main"``). Running it locates a
version-matched ``udb`` binary and execs it, transparently forwarding all
arguments and the exit code. Resolution order:
1. An ``udb`` binary whose ``--version`` matches {{UDB_VERSION}} already on
``PATH`` (but not this launcher itself).
2. A previously downloaded binary cached for this version under the per-user
cache dir.
3. The matching release asset downloaded from
``https://github.com/fahara02/udb/releases/tag/v{{UDB_VERSION}}`` for the
current OS/arch, cached, then exec'd.
Set ``UDB_BIN`` to force a specific binary, or ``UDB_SKIP_DOWNLOAD=1`` to forbid
network fetches (PATH/cache only).
"""
from __future__ import annotations
import os
import platform
import hashlib
import shutil
import stat
import subprocess
import sys
import tempfile
import urllib.request
from pathlib import Path
UDB_VERSION = "{{UDB_VERSION}}"
PROTOCOL_VERSION = "{{PROTOCOL_VERSION}}"
_REPO = "fahara02/udb"
_RELEASE_BASE = f"https://github.com/{_REPO}/releases/download/v{UDB_VERSION}"
def _exe_suffix() -> str:
return ".exe" if os.name == "nt" else ""
def _asset_name() -> str:
"""GitHub release asset name for this platform.
Convention: ``udb-<os>-<arch>[-<variant>]<ext>`` (e.g.
``udb-linux-amd64``, ``udb-windows-amd64.exe``, ``udb-darwin-arm64``).
The published release attaches RAW binaries (no version in the name, no
archive). ``<os>`` is linux|darwin|windows, ``<arch>`` is amd64|arm64.
``UDB_BIN_VARIANT`` appends ``-<variant>`` unless unset/empty/``portable``
(e.g. ``full`` -> ``udb-linux-amd64-full``). ``<ext>`` is ``.exe`` on
Windows, empty elsewhere.
"""
system = platform.system().lower()
os_tag = {"linux": "linux", "darwin": "darwin", "windows": "windows"}.get(system, system)
machine = platform.machine().lower()
arch = {
"x86_64": "amd64",
"amd64": "amd64",
"aarch64": "arm64",
"arm64": "arm64",
}.get(machine, machine)
variant = ""
raw_variant = (os.environ.get("UDB_BIN_VARIANT") or "").strip()
if raw_variant and raw_variant.lower() != "portable":
variant = f"-{raw_variant}"
return f"udb-{os_tag}-{arch}{variant}{_exe_suffix()}"
def _cache_dir() -> Path:
if os.name == "nt":
base = os.environ.get("LOCALAPPDATA") or str(Path.home() / "AppData" / "Local")
else:
base = os.environ.get("XDG_CACHE_HOME") or str(Path.home() / ".cache")
path = Path(base) / "udb" / "bin" / UDB_VERSION
path.mkdir(parents=True, exist_ok=True)
return path
def _binary_version_matches(binary: str) -> bool:
try:
out = subprocess.run(
[binary, "--version"],
capture_output=True,
text=True,
timeout=10,
)
except (OSError, subprocess.SubprocessError):
return False
return UDB_VERSION in (out.stdout + out.stderr)
def _path_binary() -> str | None:
"""An ``udb`` on PATH that is not this launcher and matches the version."""
me = Path(sys.argv[0]).resolve()
found = shutil.which("udb")
if not found:
return None
try:
if Path(found).resolve() == me:
return None
except OSError:
pass
return found if _binary_version_matches(found) else None
def _cached_binary() -> str | None:
candidate = _cache_dir() / f"udb{_exe_suffix()}"
return str(candidate) if candidate.is_file() else None
def _sha256_file(path: str) -> str:
digest = hashlib.sha256()
with open(path, "rb") as fh:
for chunk in iter(lambda: fh.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def _fetch_text(url: str) -> str | None:
try:
with urllib.request.urlopen(url, timeout=20) as resp:
return resp.read().decode("utf-8")
except Exception:
return None
def _expected_sha_from_manifest(text: str, asset: str) -> str | None:
asset_base = os.path.basename(asset)
fallback: str | None = None
for raw in text.splitlines():
line = raw.strip()
if not line or line.startswith("#"):
continue
parts = line.split()
if len(parts) == 1:
fallback = parts[0].lower().lstrip("*")
continue
digest = parts[0].lower()
filename = parts[1].lstrip("*")
if filename == asset or os.path.basename(filename) == asset_base:
return digest
return fallback
def _verify_checksum(path: str, asset: str) -> None:
if os.environ.get("UDB_SKIP_CHECKSUM"):
return
urls = [
f"{_RELEASE_BASE}/{asset}.sha256",
f"{_RELEASE_BASE}/SHA256SUMS",
f"{_RELEASE_BASE}/sha256sums.txt",
]
for url in urls:
text = _fetch_text(url)
if not text:
continue
expected = _expected_sha_from_manifest(text, asset)
if not expected:
continue
actual = _sha256_file(path)
if actual.lower() != expected.lower():
raise SystemExit(
f"udb: checksum mismatch for {asset}: expected {expected}, got {actual}"
)
return
sys.stderr.write(f"udb: warning: checksum manifest unavailable for {asset}\n")
def _download_binary() -> str:
if os.environ.get("UDB_SKIP_DOWNLOAD"):
raise SystemExit(
f"udb {UDB_VERSION} binary not found and UDB_SKIP_DOWNLOAD is set. "
f"Install it from {_RELEASE_BASE} or set UDB_BIN."
)
url = f"{_RELEASE_BASE}/{_asset_name()}"
dest = _cache_dir() / f"udb{_exe_suffix()}"
tmp_fd, tmp_path = tempfile.mkstemp(dir=str(_cache_dir()))
os.close(tmp_fd)
try:
sys.stderr.write(f"udb: downloading {url}\n")
with urllib.request.urlopen(url) as resp, open(tmp_path, "wb") as out:
shutil.copyfileobj(resp, out)
_verify_checksum(tmp_path, _asset_name())
mode = os.stat(tmp_path).st_mode
os.chmod(tmp_path, mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
os.replace(tmp_path, dest)
except Exception as exc: # noqa: BLE001 - surface a clean CLI error
try:
os.unlink(tmp_path)
except OSError:
pass
raise SystemExit(
f"udb: failed to download {url}: {exc}\n"
f"Install the binary manually from {_RELEASE_BASE} or set UDB_BIN."
) from exc
return str(dest)
def _resolve_binary() -> str:
forced = os.environ.get("UDB_BIN")
if forced:
return forced
return _path_binary() or _cached_binary() or _download_binary()
def main(argv: list[str] | None = None) -> int:
"""Locate and exec the version-matched ``udb`` binary, forwarding args."""
args = list(sys.argv[1:] if argv is None else argv)
binary = _resolve_binary()
if os.name == "nt":
# exec semantics on Windows are unreliable; spawn and propagate.
return subprocess.run([binary, *args]).returncode
os.execv(binary, [binary, *args])
return 0 # unreachable on POSIX
if __name__ == "__main__":
raise SystemExit(main())