KATA — API Layer (Phase II)

FastAPI service that turns the Phase-I system of record and Pydantic contracts into a governed, runnable API. Every cross-cutting compliance concern is implemented once, in the core layer, and enforced uniformly across routers.

What “governed” means here — four guarantees

  1. Authentication — OAuth2 password flow → JWT (HS256). Passwords are PBKDF2-HMAC-SHA256 @ 600k iterations (stdlib only; no native crypto build).
  2. RBAC — a single role→permission matrix (app/core/rbac.py) is the only lever for access. The PII boundary lives there: analyst and program_manager never hold participant:read_pii.
  3. Atomic auditrecord_audit() adds the audit row to the same transaction as the mutation, so it commits iff the mutation commits. The acting user is passed explicitly (FastAPI runs sync deps and sync routes in separate threadpool context copies, so a contextvar set in a dependency would not reliably reach the route). request_id/IP ride contextvars set in the ASGI middleware’s base context, which every copy inherits. PII reads are audited too.
  4. De-identification boundary — analytics emits only a salted, non-reversible participant_code, an age band (never DOB), and aggregates. No code path in analytics.py selects a name, guardian, or address.

Safety workflows (structural, non-bypassable)

Trigger Effect Gate Statute
injury_concussion incident opens Return-to-Play @ stage 1 incident can’t close until RTP cleared by a named clinician ORC 3707.511
mandated_report incident requires report_basis (422 without) can’t close without a recorded recipient ORC 2151.421
crisis incident auto-sets escalated, status escalated minor-safety / crisis policy

These live in app/api/incidents.py::_apply_safety_workflow and the close gate. There is no request shape that skips them.

Run it

python -m venv .venv && . .venv/bin/activate
pip install -r requirements-dev.txt
./run_dev.sh                       # sqlite + auto-created schema + reload
# http://localhost:8000/docs  (OpenAPI)  •  /health

Production: set KATA_DATABASE_URL (Postgres), a 32+ byte KATA_SECRET_KEY, KATA_DEID_SALT; install psycopg; leave KATA_AUTO_CREATE_SCHEMA=false (Alembic owns the schema). See .env.example.

Test

pytest -q          # 14 tests: RBAC/PII, audit atomicity+actor, safety gates, de-id

The harness runs entirely on in-memory SQLite (StaticPool) — the portable GUID type and generic JSON columns let the same models run on SQLite (tests) and Postgres (prod).

Routes

POST   /auth/token              issue JWT            (audited: auth:login)
GET    /auth/me                 current identity
POST   /participants            create               participant:write   201, audited
GET    /participants/roster     de-PII'd roster      participant:read_roster
GET    /participants/{id}       full PII             participant:read_pii  (read audited)
POST   /sessions                session + attendance session:write       audited
GET    /sessions                list                 session:read
POST   /enrollments            enroll in cohort     enrollment:write    audited
PATCH  /enrollments/{id}/end   withdraw/complete    enrollment:write    audited
GET    /enrollments            list (participant/cohort)  enrollment:read
GET    /enrollments/cohorts    cohort roster counts enrollment:read
POST   /incidents               create + safety wf   incident:write      audited
PATCH  /incidents/{id}/close    gated close          incident:write
PATCH  /incidents/{id}/rtp      advance/clear RTP    incident:escalate
POST   /consent                create consent       consent:write       audited
GET    /consent                list (by participant) consent:read
GET    /consent/participant/{id}/status  capture gate verdict  consent:read
POST   /consent/{id}/record-session  spend 5122.04 budget consent:write  audited
PATCH  /consent/{id}/revoke    revoke               consent:write       audited
POST   /goals                  create goal          goal:write          audited
GET    /goals                  list (by participant) goal:read
PATCH  /goals/{id}             update status/progress goal:write        audited
GET    /knowledge/interventions read-only spine      knowledge:read
GET    /analytics/participation de-identified only   analytics:read_deid
GET    /health

Layout

app/
  core/    config, security (pbkdf2+jwt), rbac, audit, deid, context, errors
  db/      base (engine/session/get_db), models (GUID-portable SQLAlchemy 2.0)
  schemas/ Pydantic v2 request/response contracts
  api/     deps (auth + require_permission) + routers + aggregate router
  main.py  app factory + pure-ASGI RequestContextMiddleware
tests/     conftest (per-role seed) + 4 behavioural suites

Reconcile before Alembic

app/db/models.py mirrors the documented Phase-I schema (22+ tables). Diff its column names against your actual db/schema.sql before generating the first migration. In Postgres prefer JSONB over JSON and native uuid over the SQLite-portable GUID storage. The de-id boundary is currently computed at the API layer; in production, back it with a Postgres view that has column-level grants, so de-identification is enforced by the database, not just the app.

Phase II — complete

Operational surface done: auth, participants, enrollments/cohorts, sessions (consent-gated), incidents (safety-gated), consent (ORC 5122.04), goals, knowledge, analytics. 34 tests passing.

Beyond Phase II (separate tracks)

  • Alembic migrations + a CI step that greps routers to assert every mutating route calls record_audit.
  • Wire to the Phase-I RAG layer (app/rag/) for knowledge retrieval beyond the structured interventions table.