Ben Stull
49981e2d6e
apply_values, EntryGitState, read_entry_from_git, write_entry_files, sidecar_path_for. Plus the SLICE-4 implementation plan. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
312 lines
11 KiB
Python
312 lines
11 KiB
Python
"""§22.4a configurable collection metadata — sidecar storage + dual-read.
|
|
|
|
SLICE-1 of docs/design/2026-06-06-configurable-collection-metadata.md.
|
|
|
|
Entry metadata is collection-configured and stored in a per-entry sidecar,
|
|
`<slug>.meta.yaml`, with the `.md` body kept as pure prose (INV-2). This module
|
|
is the storage/compat layer:
|
|
|
|
- the **dual-read** parser (`read_entry`) — read the sidecar if present, else
|
|
legacy top-of-document frontmatter, with identical resulting records
|
|
(INV-6);
|
|
- sidecar (de)serialization that preserves unknown / forward-compat keys
|
|
(INV-7), reusing `entry`'s canonical field semantics;
|
|
- lenient parsing that never hard-fails a read — a malformed sidecar surfaces
|
|
a flag, not an exception (INV-3).
|
|
|
|
The collection `fields:` schema and per-field validation are SLICE-2; faceted
|
|
filtering and the edit UIs are later slices. This module interprets no field
|
|
values — it only moves metadata between git and in-memory `Entry` records.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
from . import entry as entry_mod
|
|
from .entry import Entry
|
|
|
|
SIDECAR_SUFFIX = ".meta.yaml"
|
|
|
|
|
|
# ----- filename helpers -----
|
|
|
|
def sidecar_name(slug: str) -> str:
|
|
"""The sidecar filename for an entry whose markdown is `<slug>.md`."""
|
|
return f"{slug}{SIDECAR_SUFFIX}"
|
|
|
|
|
|
def is_sidecar(name: str) -> bool:
|
|
return name.endswith(SIDECAR_SUFFIX)
|
|
|
|
|
|
def slug_of_sidecar(name: str) -> str:
|
|
"""The entry stem for a `<slug>.meta.yaml` filename."""
|
|
return name[: -len(SIDECAR_SUFFIX)] if is_sidecar(name) else name
|
|
|
|
|
|
def sidecar_path_for(md_path: str) -> str:
|
|
"""The sidecar path sibling to a `<dir>/<slug>.md` entry file."""
|
|
assert md_path.endswith(".md"), md_path
|
|
return md_path[: -len(".md")] + SIDECAR_SUFFIX
|
|
|
|
|
|
# ----- metadata <-> sidecar -----
|
|
|
|
def metadata_dict(entry: Entry) -> dict[str, Any]:
|
|
"""The full metadata mapping for an entry (known fields + `extra`)."""
|
|
return entry_mod.to_frontmatter_dict(entry)
|
|
|
|
|
|
def sidecar_yaml(entry: Entry) -> str:
|
|
"""Render an entry's metadata as standalone `<slug>.meta.yaml` text."""
|
|
return yaml.safe_dump(
|
|
metadata_dict(entry), sort_keys=False, default_flow_style=False
|
|
)
|
|
|
|
|
|
def parse_sidecar(text: str) -> tuple[dict[str, Any], bool]:
|
|
"""Parse sidecar YAML leniently → `(values, malformed)`.
|
|
|
|
`malformed` is True when the text is not a YAML mapping (a list, a scalar,
|
|
or a YAML syntax error). An empty / whitespace-only sidecar is an empty
|
|
mapping, not malformed. Never raises (INV-3).
|
|
"""
|
|
try:
|
|
raw = yaml.safe_load(text)
|
|
except yaml.YAMLError:
|
|
return {}, True
|
|
if raw is None:
|
|
return {}, False
|
|
if not isinstance(raw, dict):
|
|
return {}, True
|
|
return raw, False
|
|
|
|
|
|
# ----- frontmatter stripping (INV-2) -----
|
|
|
|
def strip_frontmatter(md_text: str) -> str:
|
|
"""Return the prose body of a `.md`, dropping a leading `---…---` block.
|
|
|
|
A migrated entry's `.md` is body-only and passes through unchanged. A
|
|
not-yet-migrated `.md` still carrying frontmatter yields just its body, so
|
|
the dual-read body is the same either way.
|
|
"""
|
|
match = entry_mod.FRONTMATTER_RE.match(md_text)
|
|
if not match:
|
|
return md_text
|
|
return match.group(2).lstrip("\n")
|
|
|
|
|
|
# ----- dual-read (INV-6) -----
|
|
|
|
def read_entry(
|
|
md_text: str, sidecar_text: str | None, *, fallback_slug: str | None = None
|
|
) -> tuple[Entry, bool]:
|
|
"""Read an entry from its `.md` and optional sidecar → `(Entry, malformed)`.
|
|
|
|
- **Sidecar present and well-formed (non-empty):** metadata comes from the
|
|
sidecar; the body is the `.md` stripped of any leading frontmatter. The
|
|
sidecar is the source of truth (INV-1) and wins over stale `.md`
|
|
frontmatter.
|
|
- **Sidecar present but malformed:** the entry still loads from the legacy
|
|
`.md` frontmatter (if any) and is flagged `malformed` (INV-3).
|
|
- **Sidecar present but empty:** it has nothing to override with, so fall
|
|
back to the `.md` frontmatter (not flagged).
|
|
- **No sidecar:** the legacy path — parse the `.md` frontmatter (INV-6).
|
|
|
|
`fallback_slug` (typically the filename stem) backstops the entry's slug
|
|
whenever the metadata source lacks one — so a degenerate sidecar never
|
|
yields a slug-less record the caller has to silently drop (INV-3).
|
|
"""
|
|
def _with_slug(entry: Entry) -> Entry:
|
|
if not entry.slug and fallback_slug:
|
|
entry.slug = fallback_slug
|
|
return entry
|
|
|
|
if sidecar_text is None:
|
|
return _with_slug(entry_mod.parse(md_text)), False
|
|
|
|
values, malformed = parse_sidecar(sidecar_text)
|
|
if malformed or not values:
|
|
# Malformed or empty sidecar: load from the legacy .md so the entry
|
|
# still loads; flag only when the sidecar was actually malformed.
|
|
try:
|
|
entry = entry_mod.parse(md_text)
|
|
except ValueError:
|
|
entry = entry_mod.from_frontmatter({}, strip_frontmatter(md_text))
|
|
return _with_slug(entry), malformed
|
|
|
|
body = strip_frontmatter(md_text)
|
|
return _with_slug(entry_mod.from_frontmatter(values, body)), False
|
|
|
|
|
|
# ----- value editing (SLICE-4) -----
|
|
|
|
def apply_values(entry: Entry, values: dict[str, Any]) -> Entry:
|
|
"""Return a new Entry with `values` merged over the entry's metadata.
|
|
|
|
Known keys (`tags`, `state`, `reviewed_by`, …) land on their typed fields;
|
|
unknown keys land on `extra` (INV-7). The body is carried through unchanged
|
|
— this mutates metadata only. Unspecified keys are preserved.
|
|
"""
|
|
merged = metadata_dict(entry)
|
|
merged.update(values)
|
|
return entry_mod.from_frontmatter(merged, entry.body)
|
|
|
|
|
|
# ----- git-aware read/write (SLICE-4) -----
|
|
|
|
@dataclass
|
|
class EntryGitState:
|
|
"""An entry's on-disk state across its `.md` and optional sidecar.
|
|
|
|
Captured by `read_entry_from_git` and consumed by `write_entry_files` to
|
|
decide create-vs-update for the sidecar and whether the `.md` still needs
|
|
its frontmatter stripped (lazy migration).
|
|
"""
|
|
entry: Entry
|
|
md_text: str
|
|
md_sha: str
|
|
sidecar_text: str | None
|
|
sidecar_sha: str | None
|
|
malformed: bool
|
|
|
|
|
|
async def read_entry_from_git(
|
|
gitea: Any, org: str, repo: str, md_path: str, *, ref: str = "main"
|
|
) -> "EntryGitState | None":
|
|
"""Dual-read an entry from git → `EntryGitState`, or None if the `.md` is
|
|
missing. Reads the `.md` and its sibling sidecar (if any); never raises on
|
|
bad metadata (INV-3)."""
|
|
md = await gitea.read_file(org, repo, md_path, ref=ref)
|
|
if md is None:
|
|
return None
|
|
md_text, md_sha = md
|
|
sc_path = sidecar_path_for(md_path)
|
|
sc = await gitea.read_file(org, repo, sc_path, ref=ref)
|
|
sidecar_text, sidecar_sha = (sc[0], sc[1]) if sc else (None, None)
|
|
stem = md_path.rsplit("/", 1)[-1][: -len(".md")]
|
|
entry, malformed = read_entry(md_text, sidecar_text, fallback_slug=stem)
|
|
return EntryGitState(
|
|
entry=entry, md_text=md_text, md_sha=md_sha,
|
|
sidecar_text=sidecar_text, sidecar_sha=sidecar_sha, malformed=malformed,
|
|
)
|
|
|
|
|
|
def _md_has_frontmatter(md_text: str) -> bool:
|
|
return entry_mod.FRONTMATTER_RE.match(md_text) is not None
|
|
|
|
|
|
def write_entry_files(
|
|
md_path: str, entry: Entry, state: "EntryGitState"
|
|
) -> list[dict[str, Any]]:
|
|
"""Produce `change_files` ops that persist `entry`'s metadata to its sidecar
|
|
and keep the `.md` as pure prose (INV-1/INV-2).
|
|
|
|
- Sidecar: `create` when none existed, else `update` at its prior sha.
|
|
- `.md`: rewritten body-only **only when it still carries frontmatter**
|
|
(lazy migration, INV-6); an already-clean body is left untouched.
|
|
"""
|
|
sc_path = sidecar_path_for(md_path)
|
|
ops: list[dict[str, Any]] = []
|
|
sc_op: dict[str, Any] = {
|
|
"operation": "update" if state.sidecar_sha else "create",
|
|
"path": sc_path,
|
|
"content": sidecar_yaml(entry),
|
|
}
|
|
if state.sidecar_sha:
|
|
sc_op["sha"] = state.sidecar_sha
|
|
ops.append(sc_op)
|
|
if _md_has_frontmatter(state.md_text):
|
|
body = strip_frontmatter(state.md_text)
|
|
new_md = body if (body == "" or body.endswith("\n")) else body + "\n"
|
|
ops.append({
|
|
"operation": "update", "path": md_path,
|
|
"content": new_md, "sha": state.md_sha,
|
|
})
|
|
return ops
|
|
|
|
|
|
# ----- frontmatter -> sidecar migration (PUC-5) -----
|
|
|
|
async def migrate_collection(
|
|
gitea: Any,
|
|
*,
|
|
org: str,
|
|
repo: str,
|
|
subfolder: str = "",
|
|
actor: Any = None,
|
|
branch: str = "main",
|
|
) -> dict[str, Any]:
|
|
"""Migrate a collection's legacy-frontmatter entries to sidecars.
|
|
|
|
Walks `<subfolder>/rfcs`; for each `<slug>.md` that has legacy frontmatter
|
|
and **no** `<slug>.meta.yaml` sibling yet, it stages two file changes —
|
|
create the sidecar (the entry's metadata, unknown keys preserved, INV-7)
|
|
and rewrite the `.md` to body-only (INV-2) — and commits all of them in a
|
|
single ChangeFiles commit (§6.5: one commit per collection).
|
|
|
|
Idempotent: an entry that already has a sidecar is skipped; a second run
|
|
with nothing left to migrate makes no commit. Returns
|
|
`{"migrated": [...], "skipped": [...], "committed": bool}`.
|
|
"""
|
|
rfcs_dir = f"{subfolder}/rfcs" if subfolder else "rfcs"
|
|
listing = await gitea.list_dir(org, repo, rfcs_dir, ref=branch)
|
|
names = {f.get("name") for f in listing if f.get("type") == "file"}
|
|
|
|
ops: list[dict[str, Any]] = []
|
|
migrated: list[str] = []
|
|
skipped: list[str] = []
|
|
for f in listing:
|
|
if f.get("type") != "file" or not f.get("name", "").endswith(".md"):
|
|
continue
|
|
slug = f["name"][:-len(".md")]
|
|
if sidecar_name(slug) in names:
|
|
skipped.append(slug) # already migrated
|
|
continue
|
|
result = await gitea.read_file(org, repo, f["path"], ref=branch)
|
|
if not result:
|
|
continue
|
|
text, sha = result
|
|
try:
|
|
e = entry_mod.parse(text)
|
|
except ValueError:
|
|
# No frontmatter to lift (e.g. an already-clean body without a
|
|
# sidecar) — nothing to migrate; leave it untouched.
|
|
skipped.append(slug)
|
|
continue
|
|
body = strip_frontmatter(text)
|
|
new_md = body if (body == "" or body.endswith("\n")) else body + "\n"
|
|
ops.append({
|
|
"operation": "create",
|
|
"path": f"{rfcs_dir}/{sidecar_name(slug)}",
|
|
"content": sidecar_yaml(e),
|
|
})
|
|
ops.append({
|
|
"operation": "update",
|
|
"path": f["path"],
|
|
"content": new_md,
|
|
"sha": sha,
|
|
})
|
|
migrated.append(slug)
|
|
|
|
committed = False
|
|
if ops:
|
|
n = len(migrated)
|
|
message = f"Migrate {n} entr{'y' if n == 1 else 'ies'} to metadata sidecars (§22.4a)"
|
|
kwargs: dict[str, Any] = {}
|
|
if actor is not None:
|
|
kwargs = {
|
|
"author_name": actor.display_name,
|
|
"author_email": actor.email or f"{actor.gitea_login}@users.noreply",
|
|
}
|
|
await gitea.change_files(
|
|
org, repo, files=ops, message=message, branch=branch, **kwargs
|
|
)
|
|
committed = True
|
|
|
|
return {"migrated": migrated, "skipped": skipped, "committed": committed}
|