Ben Stull
9ca07a3f81
- collections.py resolution helpers (default_collection_id, type, initial_state) - registry mirror writes project grouping fields + default-collection corpus fields - auth.project_of_rfc joins collections; project_member_role reads memberships - cache/api_*/funder writers+readers re-keyed to collection_id (cached_prs + denormalised project_id tags unchanged); api_deployment reads type/initial_state from the default collection - projects.py restamp detects bootstrap via collections; initial_state via collection - tests updated to the three-tier schema; test_migration_028 retired (superseded by 029) - add @S1 acceptance test (collection grain + N=1 serving) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
51 lines
1.9 KiB
Python
51 lines
1.9 KiB
Python
"""§22 collection grain — resolution helpers beneath the project tier.
|
|
|
|
In S1 each project has exactly one collection (the default). These helpers
|
|
recover the collection for a project and read the per-corpus fields (`type`,
|
|
`initial_state`) that moved down from `projects` in migration 029. Project-grain
|
|
authz (auth.py) recovers a row's project by joining `collections` on
|
|
`collection_id`.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from . import db
|
|
|
|
DEFAULT_COLLECTION_ID = "default"
|
|
|
|
|
|
def default_collection_id(project_id: str) -> str:
|
|
"""The id of a project's default (S1: sole) collection. Falls back to the
|
|
literal 'default' when the project has no collection row yet."""
|
|
row = db.conn().execute(
|
|
"SELECT id FROM collections WHERE project_id = ? ORDER BY created_at, id LIMIT 1",
|
|
(project_id,),
|
|
).fetchone()
|
|
return row["id"] if row else DEFAULT_COLLECTION_ID
|
|
|
|
|
|
def project_of_collection(collection_id: str) -> str | None:
|
|
"""The project a collection belongs to, or None if unknown."""
|
|
row = db.conn().execute(
|
|
"SELECT project_id FROM collections WHERE id = ?", (collection_id,)
|
|
).fetchone()
|
|
return row["project_id"] if row else None
|
|
|
|
|
|
def collection_initial_state(collection_id: str) -> str:
|
|
"""§22.4b landing state for new entries in a collection. 'super-draft'
|
|
default for an unknown/unset row (today's safe flow)."""
|
|
row = db.conn().execute(
|
|
"SELECT initial_state FROM collections WHERE id = ?", (collection_id,)
|
|
).fetchone()
|
|
if row is None or not row["initial_state"]:
|
|
return "super-draft"
|
|
return row["initial_state"]
|
|
|
|
|
|
def collection_type(collection_id: str) -> str:
|
|
"""The collection's immutable §22.4a type. 'document' default for unknown."""
|
|
row = db.conn().execute(
|
|
"SELECT type FROM collections WHERE id = ?", (collection_id,)
|
|
).fetchone()
|
|
return row["type"] if row and row["type"] else "document"
|