refactor(api): Pydantic request models for setup & admin routes (R6 S1b)

- setup: SetupAdminIn (KILLS the 3rd named 500 — a non-string password reaching
  argon2/len()), SetupTestEmailIn; setup_config stays a typed dict[str, Any] (a
  genuinely dynamic key→value settings map, not a fixed schema).
- admin: AddInviteIn / SetRoleIn / SetSuspendedIn / AddDemoWhitelistIn. The
  role/status business-rule 400s are unchanged; type errors are now 422.

All 3 ROADMAP-named 500-producers now fixed (channels priority + tag_id, setup
password). Gate: ruff clean, pytest 164 green.
This commit is contained in:
2026-07-26 04:56:55 +02:00
parent 99d65cea23
commit ab83aae8bb
2 changed files with 51 additions and 17 deletions
+27 -9
View File
@@ -3,6 +3,7 @@ Also the demo-account admin surface: the demo email whitelist + a manual state r
from datetime import datetime, timezone
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
from pydantic import BaseModel
from sqlalchemy import delete, func, select
from sqlalchemy.orm import Session
@@ -110,14 +111,18 @@ def deny_invite(
return _serialize(_decide(db, invite_id, admin, approved=False))
class AddInviteIn(BaseModel):
email: str = ""
@router.post("/invites")
def add_invite(
payload: dict,
body: AddInviteIn,
admin: User = Depends(admin_user),
db: Session = Depends(get_db),
) -> dict:
"""Manually whitelist an email (approved straight away). Convenience for the admin."""
email = (payload.get("email") or "").strip().lower()
email = body.email.strip().lower()
if not valid_email(email):
raise HTTPException(status_code=400, detail="Enter a valid email address.")
inv = db.execute(select(Invite).where(Invite.email == email)).scalar_one_or_none()
@@ -157,17 +162,21 @@ def list_users(_: User = Depends(admin_user), db: Session = Depends(get_db)) ->
return [_serialize_user(u) for u in rows]
class SetRoleIn(BaseModel):
role: str
@router.patch("/users/{user_id}/role")
def set_user_role(
user_id: int,
payload: dict,
body: SetRoleIn,
background: BackgroundTasks,
admin: User = Depends(admin_user),
db: Session = Depends(get_db),
) -> dict:
"""Promote/demote a user. Guards: can't change your own role, can't touch the demo
account, and can't remove the last remaining admin. Emails the user when it actually changes."""
role = payload.get("role")
role = body.role
if role not in ("user", "admin"):
raise HTTPException(status_code=400, detail="role must be 'user' or 'admin'")
target = db.get(User, user_id)
@@ -198,10 +207,14 @@ def set_user_role(
return _serialize_user(target)
class SetSuspendedIn(BaseModel):
suspended: bool = False
@router.patch("/users/{user_id}/suspend")
def set_user_suspended(
user_id: int,
payload: dict,
body: SetSuspendedIn,
background: BackgroundTasks,
admin: User = Depends(admin_user),
db: Session = Depends(get_db),
@@ -211,7 +224,7 @@ def set_user_suspended(
suspend yourself, and can't suspend the last active admin (that would lock admin out). On
un-suspend the user is emailed they can return (suspend is announced reactively, on a blocked
sign-in)."""
suspended = bool(payload.get("suspended"))
suspended = body.suspended
target = db.get(User, user_id)
if target is None:
raise HTTPException(status_code=404, detail="No such user")
@@ -296,15 +309,20 @@ def list_demo_whitelist(
return [_serialize_demo(r) for r in rows]
class AddDemoWhitelistIn(BaseModel):
email: str = ""
note: str | None = None
@router.post("/demo/whitelist")
def add_demo_whitelist(
payload: dict,
body: AddDemoWhitelistIn,
admin: User = Depends(admin_user),
db: Session = Depends(get_db),
) -> dict:
"""Add an email that may enter the shared demo account from the login page (no Google
sign-in). Idempotent — re-adding an existing email just returns it."""
email = (payload.get("email") or "").strip().lower()
email = body.email.strip().lower()
if not valid_email(email):
raise HTTPException(status_code=400, detail="Enter a valid email address.")
row = db.execute(
@@ -313,7 +331,7 @@ def add_demo_whitelist(
if row is None:
row = DemoWhitelist(
email=email,
note=(payload.get("note") or "").strip() or None,
note=(body.note or "").strip() or None,
added_by=admin.email,
)
db.add(row)