Class FileTraceStorage

uncommon_route/traces.py:182–270

Append-only JSONL store rotated daily.

class FileTraceStorage(TraceStorage):
    """Append-only JSONL store rotated daily."""

    def __init__(self, base_dir: Path | None = None) -> None:
        self._base_dir = base_dir or (data_dir() / "traces")

    def append(self, record: dict[str, Any]) -> None:
        try:
            self._base_dir.mkdir(parents=True, exist_ok=True, mode=0o700)
            day = _date_str(float(record.get("timestamp", time.time())))
            path = self._base_dir / f"{day}.jsonl"
            line = json.dumps(record, default=str, ensure_ascii=False)
            with open(path, "a", encoding="utf-8") as f:
                f.write(line + "\n")
            try:
                path.chmod(0o600)
            except Exception:
                pass
        except Exception:
            pass

    def load_recent_days(
        self, days: int, *, now: float
    ) -> list[dict[str, Any]]:
        if not self._base_dir.exists():
            return []
        import datetime as _dt
        end = _dt.datetime.fromtimestamp(now, tz=_dt.timezone.utc).date()
        accepted = {
            (end - _dt.timedelta(days=i)).strftime("%Y-%m-%d")
            for i in range(max(1, days))
        }
        files = sorted(
            f for f in self._base_dir.glob("*.jsonl") if f.stem in accepted
        )
        out: list[dict[str, Any]] = []
        for f in files:
            try:
                for line in f.read_text(encoding="utf-8").splitlines():
                    line = line.strip()
                    if not line:
                        continue
                    try:
                        out.append(json.loads(line))
                    except Exception:
                        continue
            except Exception:
                continue
        return out

    def load_for_request(
        self, request_id: str, timestamp: float
    ) -> dict[str, Any] | None:
        if not self._base_dir.exists():
            return None
        import datetime as _dt
        center = _dt.datetime.fromtimestamp(timestamp, tz=_dt.timezone.utc).date()
        candidates = [