Backend API¶

Application¶

main ¶

Main FastAPI Application - MasterplanOptimiserV3 (GC) Server. Lightweight calendar backend with passkey auth and one-way publish.

lifespan `async` ¶

lifespan(_app: FastAPI)

Run application startup checks through FastAPI's supported lifecycle.

Source code in backend/app/main.py

@asynccontextmanager
async def lifespan(_app: FastAPI):
    """Run application startup checks through FastAPI's supported lifecycle."""

    await startup_event()
    from app.core.retention import retention_scheduler_loop
    from app.services.evidence_archive import evidence_archive_worker_loop

    retention_stop = asyncio.Event()
    retention_task = asyncio.create_task(
        retention_scheduler_loop(retention_stop),
        name="retention-scheduler",
    )
    archive_stop = asyncio.Event()
    archive_task = asyncio.create_task(
        evidence_archive_worker_loop(archive_stop),
        name="evidence-git-uploader",
    )
    try:
        yield
    finally:
        retention_stop.set()
        archive_stop.set()
        await asyncio.gather(retention_task, archive_task)

ha_write_permit_exception_handler `async` ¶

ha_write_permit_exception_handler(request: Request, exc: HAWritePermitError)

Fail closed if a commit outlives its short witness permit.

Source code in backend/app/main.py

@app.exception_handler(HAWritePermitError)
async def ha_write_permit_exception_handler(request: Request, exc: HAWritePermitError):
    """Fail closed if a commit outlives its short witness permit."""

    return JSONResponse(
        status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
        content={"detail": "Writes are paused because ownership cannot be verified.", "code": "HA_OWNERSHIP_UNVERIFIED"},
        headers={"Cache-Control": "no-store", "Retry-After": "5"},
    )

production_exception_handler `async` ¶

production_exception_handler(request: Request, exc: Exception)

Return a generic production error without logging sensitive values.

Source code in backend/app/main.py

@app.exception_handler(Exception)
async def production_exception_handler(request: Request, exc: Exception):
    """Return a generic production error without logging sensitive values."""
    logging.getLogger("api.error").error(json.dumps({
        "event": "request.error",
        "method": request.method,
        "path": request.url.path,
        "error_type": type(exc).__name__,
    }, separators=(",", ":"), sort_keys=True))
    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content={"detail": "Internal server error"},
    )

validation_exception_handler `async` ¶

validation_exception_handler(request: Request, exc: RequestValidationError)

Return actionable field errors without reflecting submitted values.

Source code in backend/app/main.py

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    """Return actionable field errors without reflecting submitted values."""

    detail = []
    for error in exc.errors():
        location = [
            part
            for part in error.get("loc", ())[:8]
            if isinstance(part, (str, int))
        ]
        message = str(error.get("msg") or "Invalid value").strip()[:240]
        error_type = str(error.get("type") or "value_error")[:80]
        detail.append({
            "type": error_type,
            "loc": location,
            "msg": message or "Invalid value",
        })

    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content={"detail": detail or [{
            "type": "value_error",
            "loc": ["body"],
            "msg": "The submitted data is invalid",
        }]},
    )

limit_request_body `async` ¶

limit_request_body(request: Request, call_next)

Reject request bodies larger than the server limit.

Source code in backend/app/main.py

@app.middleware("http")
async def limit_request_body(request: Request, call_next):
    """Reject request bodies larger than the server limit."""

    content_length = request.headers.get("content-length")
    if content_length:
        try:
            parsed_length = int(content_length)
        except ValueError:
            return JSONResponse(
                status_code=status.HTTP_400_BAD_REQUEST,
                content={"detail": "Invalid Content-Length header"},
            )
        if parsed_length < 0:
            return JSONResponse(
                status_code=status.HTTP_400_BAD_REQUEST,
                content={"detail": "Invalid Content-Length header"},
            )
        if parsed_length > _MAX_BODY_BYTES:
            return JSONResponse(
                status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
                content={"detail": "Request body too large"},
            )
    return await call_next(request)

enforce_active_writer `async` ¶

enforce_active_writer(request: Request, call_next)

Reject every mutation when this node is not the durable active writer.

Source code in backend/app/main.py

@app.middleware("http")
async def enforce_active_writer(request: Request, call_next):
    """Reject every mutation when this node is not the durable active writer."""

    if is_ha_enabled() and request.url.path not in {"/health", "/ha/ready", "/ha/status"}:
        if request.method not in {"POST", "PUT", "PATCH", "DELETE"}:
            if request.url.path.startswith("/api/v1/governance/public"):
                # The published controller notice is immutable, non-sensitive
                # and replicated. Keep it readable during witness transitions.
                return await call_next(request)
            root_ha_status_read = (
                request.method == "GET"
                and request.url.path == "/api/v1/admin/ha/status"
            )
            if not control_witness_ready() and not root_ha_status_read:
                return JSONResponse(
                    status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
                    content={"detail": "Live data is paused while service ownership is changing.", "code": "HA_LIVE_READS_PAUSED"},
                    headers={"Cache-Control": "no-store", "Retry-After": "5"},
                )
            return await call_next(request)
        db = SessionLocal()
        try:
            readiness = assess_readiness(db)
        except Exception:
            readiness = None
        finally:
            db.close()
        if readiness is None or not readiness.ready:
            return JSONResponse(
                status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
                content={"detail": "Writes are paused while service ownership is changing.", "code": "HA_WRITES_PAUSED"},
                headers={"Cache-Control": "no-store", "Retry-After": "5"},
            )
        try:
            require_write_permit(force_refresh=True)
        except HAWritePermitError:
            return JSONResponse(
                status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
                content={"detail": "Writes are paused because ownership cannot be verified.", "code": "HA_OWNERSHIP_UNVERIFIED"},
                headers={"Cache-Control": "no-store", "Retry-After": "5"},
            )
    return await call_next(request)

enforce_content_type `async` ¶

enforce_content_type(request: Request, call_next)

Require JSON content type for mutating API requests.

Source code in backend/app/main.py

@app.middleware("http")
async def enforce_content_type(request: Request, call_next):
    """Require JSON content type for mutating API requests."""

    if (
        request.method in _WRITE_METHODS
        and request.url.path.startswith("/api/")
        and request.url.path not in _CONTENT_TYPE_EXEMPT_PATHS
        and not any(request.url.path.startswith(p) for p in _CONTENT_TYPE_EXEMPT_PREFIXES)
    ):
        ct = (request.headers.get("content-type") or "").lower()
        if not ct.startswith("application/json"):
            return JSONResponse(
                status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
                content={"detail": "Content-Type must be application/json"},
            )
    return await call_next(request)

prevent_sensitive_response_caching `async` ¶

prevent_sensitive_response_caching(request: Request, call_next)

Prevent browser and intermediary storage of authenticated API data.

Source code in backend/app/main.py

@app.middleware("http")
async def prevent_sensitive_response_caching(request: Request, call_next):
    """Prevent browser and intermediary storage of authenticated API data."""
    response = await call_next(request)
    if request.url.path.startswith(_NO_STORE_PREFIXES):
        response.headers["Cache-Control"] = "no-store"
        response.headers["Pragma"] = "no-cache"
    return response

log_requests `async` ¶

log_requests(request: Request, call_next)

Log only the bounded, purpose-defined request metadata.

Source code in backend/app/main.py

@app.middleware("http")
async def log_requests(request: Request, call_next):
    """Log only the bounded, purpose-defined request metadata."""

    request_id = uuid.uuid4().hex[:12]
    start = time.perf_counter()
    response = await call_next(request)
    elapsed_ms = (time.perf_counter() - start) * 1000
    _request_logger.info(json.dumps({
        "duration_ms": round(elapsed_ms),
        "event": "request.completed",
        "method": request.method,
        "path": request.url.path,
        "request_id": request_id,
        "status": response.status_code,
        "subject_ref": getattr(request.state, "subject_ref", None),
    }, separators=(",", ":"), sort_keys=True))
    response.headers["X-Request-ID"] = request_id
    return response

health_check `async` ¶

health_check()

Liveness / readiness probe.

Source code in backend/app/main.py

@app.get("/health", tags=["health"])
async def health_check():
    """Liveness / readiness probe."""
    db_ok = False
    try:
        db = SessionLocal()
        db.execute(text("SELECT 1"))
        db.close()
        db_ok = True
    except Exception:
        pass
    payload = {"status": "ok" if db_ok else "degraded", "version": app.version, "db": db_ok}
    if not db_ok:
        return JSONResponse(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            content=payload,
            headers={"Cache-Control": "no-store", "Retry-After": "5"},
        )
    return payload

ha_ready `async` ¶

ha_ready()

Return 200 only for the current writable cluster generation.

Source code in backend/app/main.py

@app.get("/ha/ready", tags=["health"], response_class=PlainTextResponse)
async def ha_ready():
    """Return 200 only for the current writable cluster generation."""

    db = SessionLocal()
    try:
        readiness = assess_readiness(db)
        if readiness.ready:
            record_heartbeat(db)
            return PlainTextResponse(
                "ready\n",
                status_code=status.HTTP_200_OK,
                headers={"Cache-Control": "no-store"},
            )
    except Exception:
        pass
    finally:
        db.close()
    return PlainTextResponse(
        "unavailable\n",
        status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
        headers={"Cache-Control": "no-store", "Retry-After": "5"},
    )

ha_status `async` ¶

ha_status()

Expose a sanitised status shell without requiring database ownership.

Source code in backend/app/main.py

@app.get("/ha/status", tags=["health"])
async def ha_status():
    """Expose a sanitised status shell without requiring database ownership."""

    return JSONResponse(
        status_code=status.HTTP_200_OK,
        content=public_service_status(),
        headers={"Cache-Control": "no-store"},
    )

startup_event `async` ¶

startup_event()

Create tables, seed admin, housekeep.

Source code in backend/app/main.py

async def startup_event():
    """Create tables, seed admin, housekeep."""
    if engine.dialect.name == "postgresql":
        with engine.connect() as connection:
            in_recovery = connection.execute(text("SELECT pg_is_in_recovery()")).scalar()
        if in_recovery:
            raise RuntimeError("The application backend refuses a recovery database")
        # Both symmetric nodes keep the backend available for health checks.
        # Request and commit fencing, rather than process startup, decides
        # which one may expose application data or accept writes.

    # Import models so SQLAlchemy sees them
    from app.models.ha import HAClusterState, HAProtectionOperation  # noqa
    from app.models.event import Event  # noqa
    from app.models.published import (  # noqa
        PublishedTask,
        PublishedPerson,
        PublishedPersonUnavailability,
        TaskEdit,
        PublishSnapshot,
        PublishedGeneralScheduleCategory,
        PublishedGeneralScheduleItem,
        GeneralSchedulePublishState,
    )
    from app.models.user import (  # noqa
        User, WebAuthnCredential, PasskeyChallenge, PasskeyCeremony,
        ExchangeCode, AuthSession, ActivationLink, ActivationEmailDelivery,
    )
    from app.models.notification import PushSubscription, Announcement, ScheduleChange  # noqa
    from app.models.server_setting import ServerSetting  # noqa
    from app.models.audit import AuditLog  # noqa
    from app.models.public_schedule_link import (  # noqa
        PublicScheduleLink,
        PublicScheduleLinkView,
    )
    from app.models.governance import (  # noqa
        DataPolicyAcknowledgement,
        GovernancePublication,
        InstanceGovernanceProfile,
    )
    from app.models.deletion import (  # noqa
        DeletionApprovalChallenge,
        DeletionChecklistApproval,
        DeletionCase,
        DeletionSubjectScope,
        DesktopDeletionWorkOrder,
    )
    from app.models.evidence import (  # noqa
        BackupInventoryRecord,
        EvidenceChainState,
        EvidenceKey,
        EvidenceKeyRegistrationChallenge,
        EvidenceOperation,
        EvidenceArchiveSubmission,
        PrivacyActionReceipt,
    )
    from app.models.retention import RetentionSchedulerState  # noqa

    # A peer process stays healthy enough for monitoring and promotion, but it
    # must not perform *any* schema, cleanup or bootstrap writes. Committed
    # migrations prepare both local databases during deployment. It must still
    # verify the replicated database/evidence pair before reporting healthy.
    if is_ha_enabled() and not control_witness_ready():
        verification_db = SessionLocal()
        try:
            from app.core.evidence import verify_existing
            verify_existing(verification_db)
        finally:
            verification_db.rollback()
            verification_db.close()
        print("[Startup] Schema verification and housekeeping skipped on a non-holder HA node")
        return

    # create_all is an engine-level DDL operation and therefore does not pass
    # through the Session commit fence. Require an online permit explicitly.
    require_write_permit(force_refresh=True)
    Base.metadata.create_all(bind=engine)
    print("[Startup] Database tables created / verified")

    db = SessionLocal()
    try:
        # Create default root admin
        from app.core.security import create_default_admin
        create_default_admin(db)

        # Evidence initialisation is idempotent and mandatory.
        from app.core.evidence import initialise as initialise_evidence
        initialise_evidence(db)

        from app.core.retention import run_retention_cycle

        counts = run_retention_cycle(db)
        print(
            "[Startup] Retention cycle complete "
            f"({sum(counts.values())} bounded action(s))"
        )

    except Exception as exc:
        logging.getLogger(__name__).critical(
            "Startup initialisation failed (%s)",
            type(exc).__name__,
        )
        raise
    finally:
        db.close()

root `async` ¶

root()

Serve the static frontend when bundled, otherwise return API metadata.

Source code in backend/app/main.py

@app.get("/")
async def root():
    """Serve the static frontend when bundled, otherwise return API metadata."""

    _static = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "static")
    _index = os.path.join(_static, "index.html")
    if os.path.isfile(_index):
        from starlette.responses import FileResponse
        return FileResponse(_index)
    return {"message": "Masterplan Calendar API", "version": "1.0.0"}

API Router¶

router ¶

API v1 Router - wire all sub-routers.

Activation¶

activation ¶

Activation-token validation for passkey registration.

ActivationValidateResponse ¶

Bases: BaseModel

Response returned when an activation token is checked.

Source code in backend/app/api/v1/activation.py

class ActivationValidateResponse(BaseModel):
    """Response returned when an activation token is checked."""

    valid: bool
    username: Optional[str] = None
    display_name: Optional[str] = None
    purpose: Optional[
        Literal["initial_setup", "additional_passkey", "credential_reset"]
    ] = None
    logo_color_1: Optional[str] = None
    logo_color_2: Optional[str] = None

ActivationTokenRequest ¶

Bases: BaseModel

Activation token submitted without exposing it in the request URL.

Source code in backend/app/api/v1/activation.py

class ActivationTokenRequest(BaseModel):
    """Activation token submitted without exposing it in the request URL."""

    token: str = Field(..., min_length=20, max_length=256)

validate_token ¶

validate_token(body: ActivationTokenRequest, request: Request, db: Session = Depends(get_db))

Check whether an activation token is valid.

Source code in backend/app/api/v1/activation.py

@router.post("/validate", response_model=ActivationValidateResponse)
@limiter.limit(
    runtime_limit("passkey_requests_per_minute"),
    key_func=client_ip_rate_key,
)
def validate_token(
    body: ActivationTokenRequest,
    request: Request,
    db: Session = Depends(get_db),
):
    """Check whether an activation token is valid."""
    link = validate_activation_token(body.token, db)
    if link is None:
        return ActivationValidateResponse(valid=False)

    user = db.query(User).filter(User.id == link.user_id).first()
    if not user or not user.is_active:
        return ActivationValidateResponse(valid=False)

    return ActivationValidateResponse(
        valid=True,
        username=user.username,
        display_name=user.display_name,
        purpose=link.purpose,
    )

Admin¶

admin ¶

Admin endpoints - event CRUD (with secret generation) and user management.