jhkaktus
f1250881f4
Packages are ordinary Git repositories carrying a sjnetpkg.toml, so sjnetpkg only needs to resolve, clone, track and run them instead of inventing an archive format or build system. This adds the v0.1 CLI (install from registry or Git URL, remove, list/inspect, registry update, run) on top of the Python standard library, together with a test suite. 🤖 Generated with Codebuff Co-Authored-By: Codebuff <noreply@codebuff.com>
102 lines
3.5 KiB
Python
102 lines
3.5 KiB
Python
"""Downloading, reading and querying the package registry.
|
|
|
|
The registry is a plain TOML file mapping package names to Git repositories::
|
|
|
|
[packages.cool-tool]
|
|
repo = "https://github.com/example/cool-tool.git"
|
|
|
|
A copy is kept at ``~/.sjnetpkg/registry.toml`` so packages can be resolved
|
|
even when the remote registry is unreachable.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import tomllib
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
from . import config
|
|
from .config import Paths
|
|
from .errors import SjnetpkgError
|
|
|
|
_DOWNLOAD_TIMEOUT_SECONDS = 30
|
|
|
|
|
|
class RegistryError(SjnetpkgError):
|
|
"""Raised when the registry cannot be read, parsed or downloaded."""
|
|
|
|
|
|
class PackageNotFoundError(RegistryError):
|
|
"""Raised when a package name is not present in the registry."""
|
|
|
|
|
|
def download_registry(url: str, destination: Path) -> None:
|
|
"""Download the registry and atomically replace the local copy.
|
|
|
|
The existing registry is only replaced after the download succeeded *and*
|
|
the content parsed as valid TOML, so a failed download never destroys a
|
|
working local registry.
|
|
"""
|
|
try:
|
|
with urllib.request.urlopen(url, timeout=_DOWNLOAD_TIMEOUT_SECONDS) as response:
|
|
raw = response.read()
|
|
except (urllib.error.URLError, OSError, ValueError) as exc:
|
|
raise RegistryError(f"could not download registry from '{url}': {exc}") from exc
|
|
|
|
try:
|
|
parsed = tomllib.loads(raw.decode("utf-8"))
|
|
except (UnicodeDecodeError, tomllib.TOMLDecodeError) as exc:
|
|
raise RegistryError(f"registry from '{url}' is invalid: {exc}") from exc
|
|
|
|
_parse_packages(parsed, source=url)
|
|
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp = destination.with_name(destination.name + ".tmp")
|
|
try:
|
|
tmp.write_bytes(raw)
|
|
os.replace(tmp, destination)
|
|
finally:
|
|
tmp.unlink(missing_ok=True)
|
|
|
|
|
|
def load_registry(paths: Paths) -> dict[str, str]:
|
|
"""Return the local registry as a mapping of package name to repository URL."""
|
|
if not paths.registry.is_file():
|
|
raise RegistryError("no local registry found. Run 'sjnetpkg -Sy' first.")
|
|
try:
|
|
data = tomllib.loads(paths.registry.read_text(encoding="utf-8"))
|
|
except tomllib.TOMLDecodeError as exc:
|
|
raise RegistryError(f"local registry '{paths.registry}' is invalid: {exc}") from exc
|
|
except OSError as exc:
|
|
raise RegistryError(f"cannot read local registry '{paths.registry}': {exc}") from exc
|
|
return _parse_packages(data, source=str(paths.registry))
|
|
|
|
|
|
def resolve(packages: dict[str, str], name: str) -> str:
|
|
"""Return the repository URL for *name* or raise :class:`PackageNotFoundError`."""
|
|
try:
|
|
return packages[name]
|
|
except KeyError:
|
|
raise PackageNotFoundError(f"package '{name}' was not found.") from None
|
|
|
|
|
|
def update_registry(paths: Paths) -> None:
|
|
"""Download the configured registry into the local copy."""
|
|
download_registry(config.registry_url(paths), paths.registry)
|
|
|
|
|
|
def _parse_packages(data: dict, source: str) -> dict[str, str]:
|
|
packages = data.get("packages")
|
|
if not isinstance(packages, dict):
|
|
raise RegistryError(f"registry from '{source}' has no [packages] table.")
|
|
|
|
result: dict[str, str] = {}
|
|
for name, entry in packages.items():
|
|
repo = entry.get("repo") if isinstance(entry, dict) else None
|
|
if not isinstance(repo, str) or not repo.strip():
|
|
raise RegistryError(f"registry entry '{name}' is missing a 'repo' URL.")
|
|
result[name] = repo.strip()
|
|
return result
|