destinations.py

Go to the documentation of this file.

"""Named Telegram destination metadata store.
 
Dependencies
------------
- oc_connect.audit (timestamp, utc_now - the created_at_utc stamp of a destination)
- oc_connect.workspace (state_dir - the workspace-anchored directory of the destination store)
- pydantic (BaseModel, ConfigDict - the frozen strict TelegramDestination model)
"""
from __future__ import annotations
 
import json
import os
import secrets as _secrets
from collections.abc import Callable
from datetime import datetime
from pathlib import Path
 
from pydantic import BaseModel, ConfigDict
 
from ...audit import timestamp, utc_now
from ...workspace import state_dir
 
DESTINATION_FILENAME = "destinations.json"
 
 
def default_destination_path(root: str | Path | None = None) -> Path:
    """Return the workspace-anchored destination store path (CONNECT-02 §7.1)."""
    return state_dir(root) / DESTINATION_FILENAME
 
 
class TelegramDestination(BaseModel):
    """Pinned operator identity and chat for one named Telegram target."""
 
    model_config = ConfigDict(frozen=True, extra="forbid", strict=True)
 
    name: str
    connector: str
    chat_id: str
    operator_account_id: str
    credential_name: str
    created_at_utc: str
 
 
class DestinationFileError(ValueError):
    """The destination store is unreadable or malformed."""
 
 
class DestinationStore:
    """Atomic JSON store for non-secret destination metadata."""
 
    def __init__(
        self,
        path: str | Path | None = None,
        *,
        now: Callable[[], datetime] = utc_now,
        root: str | Path | None = None,
    ) -> None:
        self.path = Path(path) if path is not None else default_destination_path(root)
        self._now = now
 
    def _load(self) -> dict[str, TelegramDestination]:
        if not self.path.is_file():
            return {}
        try:
            payload = json.loads(self.path.read_text(encoding="utf-8"))
            return {
                destination.name: destination
                for destination in (
                    TelegramDestination.model_validate(raw)
                    for raw in payload.get("destinations", [])
                )
            }
        except (OSError, UnicodeDecodeError, json.JSONDecodeError, TypeError, ValueError) as exc:
            raise DestinationFileError(
                f"could not parse Telegram destinations at {self.path} "
                f"({type(exc).__name__})"
            ) from None
 
    def _write(self, destinations: dict[str, TelegramDestination]) -> None:
        self.path.parent.mkdir(parents=True, exist_ok=True)
        temporary = self.path.with_name(
            f".tmp-{self.path.name}-{_secrets.token_hex(8)}"
        )
        payload = {
            "version": 1,
            "destinations": [
                destinations[key].model_dump(mode="json")
                for key in sorted(destinations)
            ],
        }
        try:
            with temporary.open("w", encoding="utf-8", newline="\n") as handle:
                json.dump(payload, handle, indent=2, sort_keys=True)
                handle.write("\n")
                handle.flush()
                os.fsync(handle.fileno())
            os.replace(temporary, self.path)
        finally:
            if temporary.exists():
                try:
                    temporary.unlink()
                except OSError:
                    pass
 
    def record(
        self,
        *,
        name: str,
        chat_id: str,
        operator_account_id: str,
        credential_name: str,
    ) -> TelegramDestination:
        """Create or replace one named destination."""
        fields = (name, chat_id, operator_account_id, credential_name)
        if not all(value.strip() for value in fields):
            raise ValueError("destination fields must not be blank")
        destination = TelegramDestination(
            name=name,
            connector="telegram",
            chat_id=chat_id,
            operator_account_id=operator_account_id,
            credential_name=credential_name,
            created_at_utc=timestamp(self._now()),
        )
        destinations = self._load()
        destinations[name] = destination
        self._write(destinations)
        return destination
 
    def get(self, name: str) -> TelegramDestination:
        """Return a named destination or fail with its public name."""
        try:
            return self._load()[name]
        except KeyError:
            raise KeyError(f"Telegram destination {name!r} is not configured") from None
 
    def list_destinations(self) -> tuple[TelegramDestination, ...]:
        """Return destinations in stable name order."""
        destinations = self._load()
        return tuple(destinations[key] for key in sorted(destinations))
 
 
__all__ = [
    "DESTINATION_FILENAME",
    "DestinationFileError",
    "DestinationStore",
    "TelegramDestination",
    "default_destination_path",
]