tegwick
5dc2f3cdf6
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a023c0-a0a3-7c03-b395-5a0d2757214d
109 lines
3.3 KiB
Python
109 lines
3.3 KiB
Python
"""Local JSON projection store for the vertical-slice proof (pre-Postgres)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import tempfile
|
|
from dataclasses import asdict, dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from repo_manager.time import utc_now_text
|
|
|
|
|
|
def _now() -> str:
|
|
return utc_now_text()
|
|
|
|
|
|
@dataclass
|
|
class WorkRecordEntry:
|
|
kind: str
|
|
id: str | None
|
|
status: str | None
|
|
title: str | None
|
|
source_path: str
|
|
uuid: str | None = None
|
|
parent_id: str | None = None
|
|
extra: dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
@dataclass
|
|
class RepoIndex:
|
|
slug: str
|
|
repo_root: str
|
|
head_sha: str | None
|
|
observed_at: str
|
|
source_fingerprint: str | None = None
|
|
source_files: list[str] = field(default_factory=list)
|
|
work_records: list[WorkRecordEntry] = field(default_factory=list)
|
|
events: list[dict[str, Any]] = field(default_factory=list)
|
|
schema: str = "repo_manager.index.v1"
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"schema": self.schema,
|
|
"slug": self.slug,
|
|
"repo_root": self.repo_root,
|
|
"head_sha": self.head_sha,
|
|
"observed_at": self.observed_at,
|
|
"source_fingerprint": self.source_fingerprint,
|
|
"source_files": self.source_files,
|
|
"work_records": [asdict(r) for r in self.work_records],
|
|
"events": self.events,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> RepoIndex:
|
|
records = [
|
|
WorkRecordEntry(**{k: v for k, v in r.items() if k in WorkRecordEntry.__dataclass_fields__})
|
|
for r in data.get("work_records", [])
|
|
]
|
|
return cls(
|
|
slug=data["slug"],
|
|
repo_root=data["repo_root"],
|
|
head_sha=data.get("head_sha"),
|
|
observed_at=data.get("observed_at") or _now(),
|
|
source_fingerprint=data.get("source_fingerprint"),
|
|
source_files=list(data.get("source_files") or []),
|
|
work_records=records,
|
|
events=list(data.get("events") or []),
|
|
schema=data.get("schema") or "repo_manager.index.v0",
|
|
)
|
|
|
|
|
|
def default_index_path(repo_root: Path) -> Path:
|
|
return repo_root / ".repo-manager" / "index.json"
|
|
|
|
|
|
def save_index(index: RepoIndex, path: Path | None = None) -> Path:
|
|
path = path or default_index_path(Path(index.repo_root))
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary: Path | None = None
|
|
try:
|
|
with tempfile.NamedTemporaryFile(
|
|
mode="w",
|
|
encoding="utf-8",
|
|
dir=path.parent,
|
|
prefix=f".{path.name}.",
|
|
delete=False,
|
|
) as handle:
|
|
temporary = Path(handle.name)
|
|
handle.write(json.dumps(index.to_dict(), indent=2) + "\n")
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
os.replace(temporary, path)
|
|
temporary = None
|
|
finally:
|
|
if temporary is not None:
|
|
temporary.unlink(missing_ok=True)
|
|
return path
|
|
|
|
|
|
def load_index(path: Path) -> RepoIndex:
|
|
return RepoIndex.from_dict(json.loads(path.read_text(encoding="utf-8")))
|
|
|
|
|
|
def append_event(index: RepoIndex, event: dict[str, Any]) -> None:
|
|
event = {**event, "emitted_at": event.get("emitted_at") or _now()}
|
|
index.events.append(event)
|