§22 S2: collection read helpers (list/get/subfolder)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ben Stull
2026-06-05 13:00:28 -07:00
parent 74476423ba
commit 868391870c
2 changed files with 100 additions and 0 deletions
+36
View File
@@ -48,3 +48,39 @@ def collection_type(collection_id: str) -> str:
"SELECT type FROM collections WHERE id = ?", (collection_id,)
).fetchone()
return row["type"] if row and row["type"] else "document"
def subfolder_of(collection_id: str) -> str:
"""The content-repo subfolder a collection lives under (§22.3). Empty string
for the default collection (entries at the repo root `rfcs/`)."""
row = db.conn().execute(
"SELECT subfolder FROM collections WHERE id = ?", (collection_id,)
).fetchone()
return (row["subfolder"] if row else "") or ""
def get_collection(collection_id: str) -> dict | None:
"""The full collection row as a dict, or None if unknown."""
row = db.conn().execute(
"SELECT id, project_id, type, subfolder, initial_state, visibility, name "
"FROM collections WHERE id = ?",
(collection_id,),
).fetchone()
return dict(row) if row else None
def list_collections(project_id: str, include_unlisted: bool = False) -> list[dict]:
"""Collections in a project, the default first then by name (§22.5). `unlisted`
is omitted from enumeration unless include_unlisted (a direct-id read or the
corpus mirror, which serves every collection)."""
rows = db.conn().execute(
"SELECT id, project_id, type, subfolder, initial_state, visibility, name "
"FROM collections WHERE project_id = ? ORDER BY (id != 'default'), name, id",
(project_id,),
).fetchall()
out: list[dict] = []
for r in rows:
if not include_unlisted and r["visibility"] == "unlisted":
continue
out.append(dict(r))
return out
+64
View File
@@ -0,0 +1,64 @@
"""§22 S2 — collection read helpers: list_collections / get_collection /
subfolder_of."""
from __future__ import annotations
import tempfile
from pathlib import Path
from app import collections as collections_mod, db
from app.config import Config
def _db() -> Config:
cfg = Config(
gitea_url="x", gitea_bot_user="x", gitea_bot_token="x", gitea_org="x",
registry_repo="registry", oauth_client_id="x",
oauth_client_secret="x", app_url="x", secret_key="x",
database_path=Path(tempfile.mkdtemp(prefix="colhelp-")) / "t.db",
owner_gitea_login="x", webhook_secret="x",
)
db.run_migrations(cfg)
if db._CONN is not None:
db._CONN.close()
db._CONN = None
db.init(cfg)
return cfg
def _seed(project_id="ohm"):
db.conn().execute(
"INSERT OR REPLACE INTO projects (id, name, content_repo, visibility, updated_at) "
"VALUES (?, 'Ohm', 'ohm-rfc', 'public', datetime('now'))", (project_id,))
for cid, sub, vis, name in [
("default", "", "public", "Model"),
("features", "features", "public", "Features"),
("secret", "secret", "unlisted", "Secret"),
]:
db.conn().execute(
"INSERT OR REPLACE INTO collections (id, project_id, type, subfolder, initial_state, "
"visibility, name, created_at, updated_at) VALUES (?,?, 'document', ?, "
"'super-draft', ?, ?, datetime('now'), datetime('now'))",
(cid, project_id, sub, vis, name))
def test_list_collections_excludes_unlisted():
_db()
_seed()
ids = [c["id"] for c in collections_mod.list_collections("ohm", include_unlisted=False)]
assert ids == ["default", "features"] # default first, then by name; 'secret' omitted
def test_list_collections_include_unlisted():
_db()
_seed()
ids = {c["id"] for c in collections_mod.list_collections("ohm", include_unlisted=True)}
assert ids == {"default", "features", "secret"}
def test_get_collection_and_subfolder():
_db()
_seed()
assert collections_mod.get_collection("features")["name"] == "Features"
assert collections_mod.subfolder_of("features") == "features"
assert collections_mod.subfolder_of("default") == ""
assert collections_mod.get_collection("nope") is None