| """Backing store for traces: plain files under ``TRACE_DIR``. |
| |
| On a Space, ``TRACE_DIR`` points inside a **private HF bucket mounted as a |
| read-write volume**. The platform mounts it and handles authentication, so the |
| app holds no credentials at all — there is no HF token in the environment for a |
| bug or a dependency to leak. Locally, ``TRACE_DIR`` is just a directory. |
| |
| The mount is a network filesystem, so writes go through a temp file and an |
| atomic rename: a reader must never observe a half-written trace. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import json |
| import logging |
| import os |
| import secrets |
| from pathlib import Path |
| from typing import Any |
|
|
| log = logging.getLogger("tracebin.storage") |
|
|
|
|
| class NotFound(Exception): |
| pass |
|
|
|
|
| class FileStore: |
| def __init__(self, root: str) -> None: |
| self.root = Path(root) |
| self.root.mkdir(parents=True, exist_ok=True) |
|
|
| def _path(self, trace_id: str) -> Path: |
| |
| |
| return self.root / trace_id[:2] / f"{trace_id}.json" |
|
|
| def put(self, trace_id: str, record: dict[str, Any]) -> None: |
| path = self._path(trace_id) |
| path.parent.mkdir(parents=True, exist_ok=True) |
| tmp = path.parent / f".{trace_id}.{secrets.token_hex(4)}.tmp" |
| try: |
| tmp.write_text(json.dumps(record), encoding="utf-8") |
| os.replace(tmp, path) |
| except OSError: |
| tmp.unlink(missing_ok=True) |
| raise |
|
|
| def get(self, trace_id: str) -> dict[str, Any]: |
| try: |
| return json.loads(self._path(trace_id).read_text(encoding="utf-8")) |
| except FileNotFoundError as exc: |
| raise NotFound(trace_id) from exc |
|
|
|
|
| def build_store() -> FileStore: |
| root = os.environ.get("TRACE_DIR", "./data").strip() or "./data" |
| store = FileStore(root) |
| if not os.access(root, os.W_OK): |
| |
| raise RuntimeError(f"TRACE_DIR {root!r} is not writable") |
| log.info("storage: %s", root) |
| return store |
|
|
