xpltdco/chrysopedia
2
0
Fork 0
Code Issues 2 Pull requests Projects Releases Packages Wiki Activity Actions

feat: Added consent API router with 5 endpoints (list, get, upsert with…

Browse source 
- "backend/routers/consent.py"
- "backend/schemas.py"
- "backend/main.py"

GSD-Task: S03/T02
This commit is contained in:
jlightner 2026-04-03 22:11:36 +00:00
parent 8487af0282
commit db135f738e
6 changed files with 471 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.gsd/milestones/M019/slices/S03/S03-PLAN.md
View file

@ -62,7 +62,7 @@
- Estimate: 30m
- Files: backend/models.py, alembic/versions/017_add_consent_tables.py
- Verify: cd backend && python -c "from models import VideoConsent, ConsentAuditLog, ConsentField; print('OK')"
- [ ] **T02: Create consent schemas, router, and wire into main.py** — Build the consent API: Pydantic schemas for request/response, a new router with 5 endpoints (list, get, update, history, admin summary), ownership verification, and registration in main.py.
- [x] **T02: Added consent API router with 5 endpoints (list, get, upsert with audit trail, history, admin summary) and Pydantic schemas, wired into main.py** — Build the consent API: Pydantic schemas for request/response, a new router with 5 endpoints (list, get, update, history, admin summary), ownership verification, and registration in main.py.
## Steps

16
.gsd/milestones/M019/slices/S03/tasks/T01-VERIFY.json Normal file
View file

@ -0,0 +1,16 @@
{
"schemaVersion": 1,
"taskId": "T01",
"unitId": "M019/S03/T01",
"timestamp": 1775254167886,
"passed": true,
"discoverySource": "task-plan",
"checks": [
{
"command": "cd backend",
"exitCode": 0,
"durationMs": 5,
"verdict": "pass"
}
]
}

82
.gsd/milestones/M019/slices/S03/tasks/T02-SUMMARY.md Normal file
View file

@ -0,0 +1,82 @@
---
id: T02
parent: S03
milestone: M019
provides: []
requires: []
affects: []
key_files: ["backend/routers/consent.py", "backend/schemas.py", "backend/main.py"]
key_decisions: ["Used selectinload for SourceVideo join on list endpoint", "GET single video returns defaults when no consent record exists", "Each changed field gets its own audit entry with incrementing version"]
patterns_established: []
drill_down_paths: []
observability_surfaces: []
duration: ""
verification_result: "Ran task verification: `from routers.consent import router` → 5 routes OK. Ran both slice-level checks: model imports OK, migration file valid Python."
completed_at: 2026-04-03T22:11:33.455Z
blocker_discovered: false
---
# T02: Added consent API router with 5 endpoints (list, get, upsert with audit trail, history, admin summary) and Pydantic schemas, wired into main.py
> Added consent API router with 5 endpoints (list, get, upsert with audit trail, history, admin summary) and Pydantic schemas, wired into main.py
## What Happened
---
id: T02
parent: S03
milestone: M019
key_files:
- backend/routers/consent.py
- backend/schemas.py
- backend/main.py
key_decisions:
- Used selectinload for SourceVideo join on list endpoint
- GET single video returns defaults when no consent record exists
- Each changed field gets its own audit entry with incrementing version
duration: ""
verification_result: passed
completed_at: 2026-04-03T22:11:33.455Z
blocker_discovered: false
---
# T02: Added consent API router with 5 endpoints (list, get, upsert with audit trail, history, admin summary) and Pydantic schemas, wired into main.py
**Added consent API router with 5 endpoints (list, get, upsert with audit trail, history, admin summary) and Pydantic schemas, wired into main.py**
## What Happened
Built the consent API layer on top of T01 models. Added 5 Pydantic schemas to schemas.py, created routers/consent.py with ownership verification helper (admin bypass), 5 endpoints covering CRUD + audit history + admin summary, and registered in main.py. PUT endpoint upserts consent with per-field audit log entries using auto-incrementing versions.
## Verification
Ran task verification: `from routers.consent import router` → 5 routes OK. Ran both slice-level checks: model imports OK, migration file valid Python.
## Verification Evidence
| # | Command | Exit Code | Verdict | Duration |
|---|---------|-----------|---------|----------|
| 1 | `cd backend && python -c "from routers.consent import router; print(f'{len(router.routes)} routes OK')"` | 0 | ✅ pass | 500ms |
| 2 | `cd backend && python -c "from models import VideoConsent, ConsentAuditLog, ConsentField; print('OK')"` | 0 | ✅ pass | 500ms |
| 3 | `python -c "import importlib.util; spec = importlib.util.spec_from_file_location('m', 'alembic/versions/017_add_consent_tables.py'); ..."` | 0 | ✅ pass | 300ms |
## Deviations
None.
## Known Issues
None.
## Files Created/Modified
- `backend/routers/consent.py`
- `backend/schemas.py`
- `backend/main.py`
## Deviations
None.
## Known Issues
None.

3
backend/main.py
View file

@ -12,7 +12,7 @@ from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from config import get_settings
from routers import auth, creators, health, ingest, pipeline, reports, search, stats, techniques, topics, videos
from routers import auth, consent, creators, health, ingest, pipeline, reports, search, stats, techniques, topics, videos
def _setup_logging() -> None:
@ -79,6 +79,7 @@ app.include_router(health.router)
# Versioned API
app.include_router(auth.router, prefix="/api/v1")
app.include_router(consent.router, prefix="/api/v1")
app.include_router(creators.router, prefix="/api/v1")
app.include_router(ingest.router, prefix="/api/v1")
app.include_router(pipeline.router, prefix="/api/v1")

322
backend/routers/consent.py Normal file
View file

@ -0,0 +1,322 @@
"""Consent router — per-video consent toggles with versioned audit trail.
Creator endpoints (ownership-gated):
GET /consent/videos List consent for the current creator's videos
GET /consent/videos/{video_id} Single video consent status
PUT /consent/videos/{video_id} Upsert consent (partial update, audit logged)
GET /consent/videos/{video_id}/history Audit trail for a video
Admin endpoint:
GET /consent/admin/summary Aggregate consent flag counts
"""
from __future__ import annotations
import logging
import uuid
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from auth import get_current_user, require_role
from database import get_session
from models import (
ConsentAuditLog,
ConsentField,
SourceVideo,
User,
UserRole,
VideoConsent,
)
from schemas import (
ConsentAuditEntry,
ConsentListResponse,
ConsentSummary,
VideoConsentRead,
VideoConsentUpdate,
)
logger = logging.getLogger("chrysopedia.consent")
router = APIRouter(prefix="/consent", tags=["consent"])
# ── Helpers ──────────────────────────────────────────────────────────────────
async def _verify_video_ownership(
video_id: uuid.UUID,
user: User,
session: AsyncSession,
) -> SourceVideo:
"""Load a SourceVideo and verify the user owns it (or is admin).
Returns the SourceVideo on success.
Raises 403 if user has no creator_id or doesn't own the video.
Raises 404 if video doesn't exist.
"""
result = await session.execute(
select(SourceVideo).where(SourceVideo.id == video_id)
)
video = result.scalar_one_or_none()
if video is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Video not found",
)
# Admin bypasses ownership check
if user.role == UserRole.admin:
return video
if user.creator_id is None:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="User is not linked to a creator profile",
)
if video.creator_id != user.creator_id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You do not own this video",
)
return video
def _consent_to_read(consent: VideoConsent, filename: str) -> VideoConsentRead:
"""Map a VideoConsent ORM instance to the read schema."""
return VideoConsentRead(
source_video_id=consent.source_video_id,
video_filename=filename,
creator_id=consent.creator_id,
kb_inclusion=consent.kb_inclusion,
training_usage=consent.training_usage,
public_display=consent.public_display,
updated_at=consent.updated_at,
)
# ── Endpoints ────────────────────────────────────────────────────────────────
@router.get("/videos", response_model=ConsentListResponse)
async def list_video_consents(
current_user: Annotated[User, Depends(get_current_user)],
session: Annotated[AsyncSession, Depends(get_session)],
offset: Annotated[int, Query(ge=0)] = 0,
limit: Annotated[int, Query(ge=1, le=100)] = 50,
):
"""List consent records for the current creator's videos."""
if current_user.creator_id is None and current_user.role != UserRole.admin:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="User is not linked to a creator profile",
)
stmt = (
select(VideoConsent)
.join(SourceVideo, VideoConsent.source_video_id == SourceVideo.id)
.options(selectinload(VideoConsent.source_video))
)
# Non-admin sees only their own videos
if current_user.role != UserRole.admin:
stmt = stmt.where(VideoConsent.creator_id == current_user.creator_id)
# Count
count_stmt = select(func.count()).select_from(stmt.subquery())
total = (await session.execute(count_stmt)).scalar() or 0
# Fetch page
stmt = stmt.order_by(VideoConsent.updated_at.desc())
stmt = stmt.offset(offset).limit(limit)
result = await session.execute(stmt)
consents = result.scalars().all()
items = [
_consent_to_read(c, c.source_video.filename) for c in consents
]
return ConsentListResponse(items=items, total=total)
@router.get("/videos/{video_id}", response_model=VideoConsentRead)
async def get_video_consent(
video_id: uuid.UUID,
current_user: Annotated[User, Depends(get_current_user)],
session: Annotated[AsyncSession, Depends(get_session)],
):
"""Get consent status for a single video."""
video = await _verify_video_ownership(video_id, current_user, session)
result = await session.execute(
select(VideoConsent).where(VideoConsent.source_video_id == video_id)
)
consent = result.scalar_one_or_none()
if consent is None:
# No consent record yet — return defaults
return VideoConsentRead(
source_video_id=video_id,
video_filename=video.filename,
creator_id=video.creator_id,
kb_inclusion=False,
training_usage=False,
public_display=True,
updated_at=video.created_at,
)
return _consent_to_read(consent, video.filename)
@router.put("/videos/{video_id}", response_model=VideoConsentRead)
async def update_video_consent(
video_id: uuid.UUID,
body: VideoConsentUpdate,
current_user: Annotated[User, Depends(get_current_user)],
session: Annotated[AsyncSession, Depends(get_session)],
request: Request,
):
"""Upsert consent for a video. Only non-None fields are changed.
Creates audit log entries for each changed field with incrementing
version numbers.
"""
video = await _verify_video_ownership(video_id, current_user, session)
# Load or create consent record
result = await session.execute(
select(VideoConsent).where(VideoConsent.source_video_id == video_id)
)
consent = result.scalar_one_or_none()
is_new = consent is None
if is_new:
consent = VideoConsent(
source_video_id=video_id,
creator_id=video.creator_id,
updated_by=current_user.id,
)
session.add(consent)
await session.flush() # get consent.id for audit entries
# Determine the next version number
max_version_result = await session.execute(
select(func.coalesce(func.max(ConsentAuditLog.version), 0)).where(
ConsentAuditLog.video_consent_id == consent.id
)
)
next_version = (max_version_result.scalar() or 0) + 1
# Collect client IP for audit
client_ip = request.client.host if request.client else None
# Apply changes and build audit entries
fields_changed: list[str] = []
update_data = body.model_dump(exclude_none=True)
for field_name, new_value in update_data.items():
# Validate field name against the enum
try:
ConsentField(field_name)
except ValueError:
continue
old_value = getattr(consent, field_name)
# Skip if no actual change
if old_value == new_value:
continue
# Update the consent record
setattr(consent, field_name, new_value)
fields_changed.append(field_name)
# Create audit entry
audit_entry = ConsentAuditLog(
video_consent_id=consent.id,
version=next_version,
field_name=field_name,
old_value=old_value if not is_new else None,
new_value=new_value,
changed_by=current_user.id,
ip_address=client_ip,
)
session.add(audit_entry)
next_version += 1
if fields_changed:
consent.updated_by = current_user.id
await session.commit()
await session.refresh(consent)
logger.info(
"Consent updated: video_id=%s fields_changed=%s user=%s",
video_id,
fields_changed,
current_user.id,
)
else:
# No actual changes — still commit if we created a new record
if is_new:
await session.commit()
await session.refresh(consent)
else:
# Nothing changed, no audit entries
pass
return _consent_to_read(consent, video.filename)
@router.get("/videos/{video_id}/history", response_model=list[ConsentAuditEntry])
async def get_consent_history(
video_id: uuid.UUID,
current_user: Annotated[User, Depends(get_current_user)],
session: Annotated[AsyncSession, Depends(get_session)],
):
"""Get the audit trail for a video's consent changes."""
await _verify_video_ownership(video_id, current_user, session)
# Find the consent record
result = await session.execute(
select(VideoConsent).where(VideoConsent.source_video_id == video_id)
)
consent = result.scalar_one_or_none()
if consent is None:
return []
# Fetch audit entries ordered by version
audit_result = await session.execute(
select(ConsentAuditLog)
.where(ConsentAuditLog.video_consent_id == consent.id)
.order_by(ConsentAuditLog.version.asc())
)
return audit_result.scalars().all()
@router.get(
"/admin/summary",
response_model=ConsentSummary,
dependencies=[Depends(require_role(UserRole.admin))],
)
async def consent_admin_summary(
session: Annotated[AsyncSession, Depends(get_session)],
):
"""Aggregate consent flag counts across all videos (admin only)."""
result = await session.execute(
select(
func.count().label("total"),
func.sum(VideoConsent.kb_inclusion.cast(int)).label("kb"),
func.sum(VideoConsent.training_usage.cast(int)).label("tu"),
func.sum(VideoConsent.public_display.cast(int)).label("pd"),
)
)
row = result.one()
return ConsentSummary(
total_videos=row.total or 0,
kb_inclusion_granted=row.kb or 0,
training_usage_granted=row.tu or 0,
public_display_granted=row.pd or 0,
)

48
backend/schemas.py
View file

@ -545,3 +545,51 @@ class UpdateProfileRequest(BaseModel):
display_name: str | None = Field(None, min_length=1, max_length=255)
current_password: str | None = None
new_password: str | None = Field(None, min_length=8, max_length=128)
# ── Consent ──────────────────────────────────────────────────────────────────
class VideoConsentUpdate(BaseModel):
"""Partial update — only non-None fields trigger changes."""
kb_inclusion: bool | None = None
training_usage: bool | None = None
public_display: bool | None = None
class VideoConsentRead(BaseModel):
"""Current consent state for a video."""
model_config = ConfigDict(from_attributes=True)
source_video_id: uuid.UUID
video_filename: str = ""
creator_id: uuid.UUID
kb_inclusion: bool = False
training_usage: bool = False
public_display: bool = True
updated_at: datetime
class ConsentAuditEntry(BaseModel):
"""Single audit trail entry for a consent change."""
model_config = ConfigDict(from_attributes=True)
version: int
field_name: str
old_value: bool | None = None
new_value: bool
changed_by: uuid.UUID
created_at: datetime
class ConsentListResponse(BaseModel):
"""Paginated list of video consent records."""
items: list[VideoConsentRead] = Field(default_factory=list)
total: int = 0
class ConsentSummary(BaseModel):
"""Aggregate consent flag counts across all videos."""
total_videos: int = 0
kb_inclusion_granted: int = 0
training_usage_granted: int = 0
public_display_granted: int = 0