gitea-adminandClaude Fable 5.1
8cac4370bf
AliasRegistry and RemoteRegistry duplicated the lock-file, read-modify- write and atomic-replace logic almost line for line. Move it into a generic JsonRegistry base; each registry now only declares its error type, empty value, parser and serializer. Behaviour, file format, permissions and the corrupt-file guarantee are unchanged. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
99 lines
3.4 KiB
Python
99 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
|
|
from .jsonstore import JsonRegistry
|
|
|
|
|
|
ALIAS_REGISTRY_VERSION = 1
|
|
MAX_ALIAS_LENGTH = 48
|
|
|
|
|
|
class AliasRegistryError(RuntimeError):
|
|
"""A display-name registry error which is safe to show to the user."""
|
|
|
|
|
|
def normalize_alias(label: str) -> str:
|
|
normalized = label.strip()
|
|
if len(normalized) > MAX_ALIAS_LENGTH:
|
|
raise ValueError(f"display name must be at most {MAX_ALIAS_LENGTH} characters")
|
|
if any(ord(character) < 32 or ord(character) == 127 for character in normalized):
|
|
raise ValueError("display name must not contain control characters")
|
|
return normalized
|
|
|
|
|
|
def default_alias_registry_path() -> Path:
|
|
if override := os.environ.get("HERDR_HUB_ALIASES_FILE"):
|
|
return Path(override).expanduser()
|
|
config_home = os.environ.get("XDG_CONFIG_HOME")
|
|
root = Path(config_home).expanduser() if config_home else Path.home() / ".config"
|
|
return root / "herdr-hub" / "aliases.json"
|
|
|
|
|
|
class AliasRegistry(JsonRegistry[dict[str, str]]):
|
|
error_type = AliasRegistryError
|
|
|
|
def __init__(self, path: Path | None = None) -> None:
|
|
super().__init__(path or default_alias_registry_path())
|
|
|
|
def _empty(self) -> dict[str, str]:
|
|
return {}
|
|
|
|
def _parse(self, payload: object) -> dict[str, str]:
|
|
return _parse_aliases(payload, self.path)
|
|
|
|
def _serialize(self, aliases: dict[str, str]) -> object:
|
|
return {
|
|
"version": ALIAS_REGISTRY_VERSION,
|
|
"aliases": dict(sorted(aliases.items())),
|
|
}
|
|
|
|
def set(self, endpoint_key: str, label: str) -> bool:
|
|
_validate_endpoint_key(endpoint_key)
|
|
normalized = normalize_alias(label)
|
|
|
|
def update(aliases: dict[str, str]) -> tuple[dict[str, str], bool]:
|
|
updated = dict(aliases)
|
|
if normalized:
|
|
changed = updated.get(endpoint_key) != normalized
|
|
updated[endpoint_key] = normalized
|
|
else:
|
|
changed = endpoint_key in updated
|
|
updated.pop(endpoint_key, None)
|
|
return updated, changed
|
|
|
|
return self._update(update)
|
|
|
|
|
|
def _parse_aliases(payload: object, path: Path) -> dict[str, str]:
|
|
if (
|
|
not isinstance(payload, dict)
|
|
or payload.get("version") != ALIAS_REGISTRY_VERSION
|
|
):
|
|
raise AliasRegistryError(f"{path} has an unsupported or missing version")
|
|
raw_aliases = payload.get("aliases")
|
|
if not isinstance(raw_aliases, dict):
|
|
raise AliasRegistryError(f"{path} does not contain an aliases object")
|
|
|
|
aliases: dict[str, str] = {}
|
|
for key, raw_label in raw_aliases.items():
|
|
if not isinstance(key, str) or not isinstance(raw_label, str):
|
|
raise AliasRegistryError(f"{path} contains an invalid alias entry")
|
|
try:
|
|
_validate_endpoint_key(key)
|
|
label = normalize_alias(raw_label)
|
|
except ValueError as error:
|
|
raise AliasRegistryError(f"{path}: {error}") from error
|
|
if not label:
|
|
raise AliasRegistryError(f"{path}: display names must not be empty")
|
|
aliases[key] = label
|
|
return dict(sorted(aliases.items()))
|
|
|
|
|
|
def _validate_endpoint_key(key: str) -> None:
|
|
if not key or len(key) > 1024:
|
|
raise ValueError("endpoint key has an invalid length")
|
|
if any(ord(character) < 32 or ord(character) == 127 for character in key):
|
|
raise ValueError("endpoint key must not contain control characters")
|