Ben Stull
282706d7ef
set/add/remove ops reusing the SLICE-4 sidecar write-through; per-entry partial-rejection; contributor+ gated (INV-4); validated at the write boundary. Tests cover one-commit, add/remove, partial reject, authz, and op/field/empty guards. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
208 lines
9.2 KiB
Python
208 lines
9.2 KiB
Python
"""§22.4a SLICE-4/5 — entry metadata edit endpoints.
|
|
|
|
`POST .../rfcs/<slug>/meta` writes schema-defined metadata to an entry's sidecar
|
|
with a direct commit (D7: direct commit for authorized roles), validated against
|
|
the collection's field schema at the write boundary (INV-4), lazy-migrating a
|
|
legacy entry to a clean body-only `.md` on first edit. The Owner-gated
|
|
`metadata.migrate_collection` operator endpoint also lives here (SLICE-4 carried
|
|
work); SLICE-5's bulk endpoint will join it.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, HTTPException, Request
|
|
from pydantic import BaseModel
|
|
|
|
from . import (auth, cache, collections as collections_mod,
|
|
metadata as metadata_mod, metadata_schema,
|
|
projects as projects_mod)
|
|
from .bot import Bot
|
|
from .config import Config
|
|
from .gitea import Gitea, GiteaError
|
|
|
|
|
|
class MetaEditBody(BaseModel):
|
|
values: dict[str, Any]
|
|
|
|
|
|
class BulkMetaBody(BaseModel):
|
|
slugs: list[str]
|
|
op: str
|
|
field: str
|
|
value: Any = None
|
|
|
|
|
|
def _apply_op(entry: Any, op: str, field: str, value: Any) -> Any:
|
|
"""Return the new value for `field` after applying `op` to `entry`.
|
|
|
|
`set` → `value`; `add`/`remove` operate on the entry's current tags-list
|
|
value for `field` (the route restricts add/remove to tags-type fields).
|
|
"""
|
|
if op == "set":
|
|
return value
|
|
current = metadata_mod.metadata_dict(entry).get(field) or []
|
|
if not isinstance(current, list):
|
|
current = [current]
|
|
if op == "add":
|
|
return current if value in current else [*current, value]
|
|
if op == "remove":
|
|
return [x for x in current if x != value]
|
|
return value # unreachable; op validated by the route
|
|
|
|
|
|
def make_router(config: Config, gitea: Gitea, bot: Bot) -> APIRouter:
|
|
router = APIRouter()
|
|
|
|
def _content_repo() -> tuple[str, str]:
|
|
return config.gitea_org, (projects_mod.default_content_repo(config) or "")
|
|
|
|
def _md_path(collection_id: str, slug: str) -> str:
|
|
sub = collections_mod.subfolder_of(collection_id) or ""
|
|
rfcs_dir = f"{sub}/rfcs" if sub else "rfcs"
|
|
return f"{rfcs_dir}/{slug}.md"
|
|
|
|
@router.post("/api/projects/{project_id}/collections/{collection_id}/rfcs/{slug}/meta")
|
|
async def edit_meta(
|
|
project_id: str, collection_id: str, slug: str,
|
|
body: MetaEditBody, request: Request,
|
|
) -> dict[str, Any]:
|
|
viewer = auth.current_user(request)
|
|
if collections_mod.project_of_collection(collection_id) != project_id:
|
|
raise HTTPException(404, "Collection not in project")
|
|
# INV-4: contributor+ on the collection (returns False for anonymous).
|
|
if not auth.can_contribute_in_collection(viewer, collection_id):
|
|
raise HTTPException(403, "Contributor access required to edit metadata")
|
|
col = collections_mod.get_collection(collection_id)
|
|
fields = (col or {}).get("fields") or {}
|
|
if not fields:
|
|
raise HTTPException(422, "Collection declares no editable fields")
|
|
if not body.values:
|
|
raise HTTPException(422, "Provide at least one field value")
|
|
unknown = [k for k in body.values if k not in fields]
|
|
if unknown:
|
|
raise HTTPException(422, f"Unknown field(s): {', '.join(sorted(unknown))}")
|
|
|
|
org, repo = _content_repo()
|
|
md_path = _md_path(collection_id, slug)
|
|
st = await metadata_mod.read_entry_from_git(gitea, org, repo, md_path)
|
|
if st is None:
|
|
raise HTTPException(404, f"{md_path} not found")
|
|
|
|
new_entry = metadata_mod.apply_values(st.entry, body.values)
|
|
problems = metadata_schema.validate(
|
|
metadata_mod.metadata_dict(new_entry), fields)
|
|
if problems:
|
|
raise HTTPException(422, {"problems": [p.as_dict() for p in problems]})
|
|
|
|
files = metadata_mod.write_entry_files(md_path, new_entry, st)
|
|
try:
|
|
await bot.commit_entry_files(
|
|
viewer.as_actor(), org=org, repo=repo, files=files,
|
|
message=f"Edit metadata: {slug}", branch="main")
|
|
except GiteaError as e:
|
|
raise HTTPException(502, f"Gitea: {e.detail}")
|
|
|
|
await cache.refresh_meta_repo(config, gitea)
|
|
return {
|
|
"ok": True, "slug": slug,
|
|
"meta": metadata_mod.metadata_dict(new_entry),
|
|
}
|
|
|
|
@router.post("/api/projects/{project_id}/collections/{collection_id}/meta/bulk")
|
|
async def bulk_meta(
|
|
project_id: str, collection_id: str,
|
|
body: BulkMetaBody, request: Request,
|
|
) -> dict[str, Any]:
|
|
"""§22.4a PUC-2 (SLICE-5): apply one field op to many entries at once.
|
|
|
|
`set` works for any field; `add`/`remove` operate on a tags-type field.
|
|
Each passing entry's metadata is validated at the write boundary
|
|
(INV-4) and its sidecar staged; all stage into **one** commit (D7:
|
|
bulk = 1 commit, reusing the SLICE-4 sidecar write-through). Entries
|
|
that are missing or fail validation are reported in `rejected`; the
|
|
rest in `applied`. A no-op (value unchanged) is applied without writing.
|
|
"""
|
|
viewer = auth.current_user(request)
|
|
if collections_mod.project_of_collection(collection_id) != project_id:
|
|
raise HTTPException(404, "Collection not in project")
|
|
# INV-4: contributor+ on the collection (returns False for anonymous).
|
|
if not auth.can_contribute_in_collection(viewer, collection_id):
|
|
raise HTTPException(403, "Contributor access required to edit metadata")
|
|
col = collections_mod.get_collection(collection_id)
|
|
fields = (col or {}).get("fields") or {}
|
|
if not fields:
|
|
raise HTTPException(422, "Collection declares no editable fields")
|
|
if not body.slugs:
|
|
raise HTTPException(422, "Provide at least one entry")
|
|
if body.op not in ("set", "add", "remove"):
|
|
raise HTTPException(422, f"Unknown op: {body.op}")
|
|
if body.field not in fields:
|
|
raise HTTPException(422, f"Unknown field: {body.field}")
|
|
if body.op in ("add", "remove") and fields[body.field].get("type") != "tags":
|
|
raise HTTPException(422, f"op {body.op} requires a tags field")
|
|
|
|
org, repo = _content_repo()
|
|
applied: list[str] = []
|
|
rejected: list[dict[str, str]] = []
|
|
all_ops: list[dict[str, Any]] = []
|
|
for slug in body.slugs:
|
|
md_path = _md_path(collection_id, slug)
|
|
st = await metadata_mod.read_entry_from_git(gitea, org, repo, md_path)
|
|
if st is None:
|
|
rejected.append({"slug": slug, "reason": "not found"})
|
|
continue
|
|
new_value = _apply_op(st.entry, body.op, body.field, body.value)
|
|
new_entry = metadata_mod.apply_values(st.entry, {body.field: new_value})
|
|
problems = metadata_schema.validate(
|
|
metadata_mod.metadata_dict(new_entry), fields)
|
|
if problems:
|
|
rejected.append({"slug": slug,
|
|
"reason": "; ".join(p.message for p in problems)})
|
|
continue
|
|
applied.append(slug)
|
|
if metadata_mod.metadata_dict(new_entry) != metadata_mod.metadata_dict(st.entry):
|
|
all_ops.extend(metadata_mod.write_entry_files(md_path, new_entry, st))
|
|
|
|
committed = False
|
|
if all_ops:
|
|
n = len(applied)
|
|
msg = f"Bulk {body.op} {body.field}: {n} entr{'y' if n == 1 else 'ies'}"
|
|
try:
|
|
await bot.commit_entry_files(
|
|
viewer.as_actor(), org=org, repo=repo, files=all_ops,
|
|
message=msg, branch="main")
|
|
except GiteaError as e:
|
|
raise HTTPException(502, f"Gitea: {e.detail}")
|
|
committed = True
|
|
await cache.refresh_meta_repo(config, gitea)
|
|
return {"ok": True, "applied": applied,
|
|
"rejected": rejected, "committed": committed}
|
|
|
|
@router.post("/api/projects/{project_id}/collections/{collection_id}/migrate")
|
|
async def migrate(
|
|
project_id: str, collection_id: str, request: Request
|
|
) -> dict[str, Any]:
|
|
"""§22.4a PUC-5: migrate a collection's legacy-frontmatter entries to
|
|
clean body-only `.md` + sidecars, one commit per collection. Owner-gated
|
|
operator action. Safe to ship now that every entry write path is
|
|
sidecar-aware (SLICE-4 carried work). Idempotent."""
|
|
viewer = auth.current_user(request)
|
|
if collections_mod.project_of_collection(collection_id) != project_id:
|
|
raise HTTPException(404, "Collection not in project")
|
|
if not auth.is_collection_superuser(viewer, collection_id):
|
|
raise HTTPException(403, "Owner access required to migrate a collection")
|
|
org, repo = _content_repo()
|
|
subfolder = collections_mod.subfolder_of(collection_id) or ""
|
|
try:
|
|
result = await metadata_mod.migrate_collection(
|
|
gitea, org=org, repo=repo, subfolder=subfolder,
|
|
actor=viewer.as_actor())
|
|
except GiteaError as e:
|
|
raise HTTPException(502, f"Gitea: {e.detail}")
|
|
if result["committed"]:
|
|
await cache.refresh_meta_repo(config, gitea)
|
|
return result
|
|
|
|
return router
|