Backend API¶

Application¶

main ¶

Main FastAPI Application Desktop-only backend API for Google Calendar integrated masterplan optimisation.

validation_exception_handler `async` ¶

validation_exception_handler(request: Request, exc: RequestValidationError)

Return a sanitized validation error response for malformed requests.

Source code in backend/app/main.py

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    """Return a sanitized validation error response for malformed requests."""
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content={"detail": "Validation error"},
    )

check_desktop_token `async` ¶

check_desktop_token(request: Request, call_next)

Require Electron's per-launch desktop token for non-exempt routes.

Source code in backend/app/main.py

@app.middleware("http")
async def check_desktop_token(request: Request, call_next):
    """Require Electron's per-launch desktop token for non-exempt routes."""
    if _DESKTOP_AUTH_TOKEN and not _is_auth_exempt(request):
        token = request.headers.get("x-desktop-token")
        if token != _DESKTOP_AUTH_TOKEN:
            return JSONResponse(status_code=403, content={"detail": "Forbidden"})
    return await call_next(request)

limit_request_body `async` ¶

limit_request_body(request: Request, call_next)

Reject oversized local API requests before they reach route handlers.

Source code in backend/app/main.py

@app.middleware("http")
async def limit_request_body(request: Request, call_next):
    """Reject oversized local API requests before they reach route handlers."""
    if not request.url.path.startswith("/compute"):
        content_length = request.headers.get("content-length")
        if content_length and int(content_length) > MAX_BODY_BYTES:
            return JSONResponse(status_code=413, content={"detail": "Request body too large"})
    return await call_next(request)

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
    return {"status": "ok" if db_ok else "degraded", "version": app.version, "db": db_ok}

startup_event `async` ¶

startup_event()

Initialise a current-schema database without compatibility migrations.

Source code in backend/app/main.py

@app.on_event("startup")
async def startup_event():
    """Initialise a current-schema database without compatibility migrations."""
    # Import all models to ensure they're registered with SQLAlchemy
    from app.models import (
        Person, Event, Task, Assignment,
        Location, Group, Capability, Theme, OptimizationJob,
        MasterplanLayout, GoogleCalendarConnection,
        AppSettings, EventPublishState,
        AudienceTeam, SessionElement, GeneralSchedulePublishState,
        ProcessorEvidenceKey,
    )

    inspector = inspect(engine)
    tables = set(inspector.get_table_names())
    if "persons" in tables:
        person_columns = {column["name"] for column in inspector.get_columns("persons")}
        event_columns = {column["name"] for column in inspector.get_columns("events")}
        if (
            "evidence_subject_id" not in person_columns
            or "global_data" in person_columns
            or "evidence_id" not in event_columns
        ):
            raise RuntimeError(
                "This database uses the retired desktop schema. Run the one-off "
                "convert_current_desktop_v2.py tool against a copy, then start the "
                "application with its converted output."
            )

    # Create tables only for a new or already-current database.
    Base.metadata.create_all(bind=engine)

    # Clear any stuck optimisation jobs from previous sessions
    db = SessionLocal()
    try:
        from sqlalchemy import or_
        from datetime import datetime

        stuck_jobs = db.query(OptimizationJob).filter(
            or_(
                OptimizationJob.status == "running",
                OptimizationJob.status == "pending"
            )
        ).all()

        if stuck_jobs:
            print(f"[Startup] Found {len(stuck_jobs)} stuck optimisation job(s) from previous session")
            for job in stuck_jobs:
                job.status = "failed"
                job.error_message = "Job was interrupted by application restart"
                job.completed_at = datetime.utcnow()
            db.commit()
            print(f"[Startup] Cleared all stuck optimisation jobs")

    except Exception as e:
        print(f"Error during startup: {e}")
    finally:
        db.close()

root `async` ¶

root()

Health check endpoint

Source code in backend/app/main.py

@app.get("/")
async def root():
    """Health check endpoint"""
    return {
        "message": "Masterplan Optimiser API",
        "version": "2.0.0",
        "environment": settings.ENVIRONMENT,
        "mode": "Desktop"
    }

App Settings¶

app_settings ¶

App Settings API Endpoints Manages application-wide configuration like Google OAuth credentials and solver tuning parameters.

SolverSettingsPayload ¶

Bases: BaseModel

Writable solver tuning values supplied by the settings UI.

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

class SolverSettingsPayload(BaseModel):
    """Writable solver tuning values supplied by the settings UI."""

    max_time_seconds: float = Field(ge=1, le=3600, default=30.0)
    break_threshold_min: int = Field(ge=1, le=240, default=30)
    break_recovery_bonus: float = Field(ge=-100, le=0, default=-3.0)
    fatigue_scale: int = Field(ge=1, le=10000, default=100)

SolverSettingsResponse ¶

Bases: BaseModel

Solver tuning values returned to the frontend.

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

class SolverSettingsResponse(BaseModel):
    """Solver tuning values returned to the frontend."""

    max_time_seconds: float
    break_threshold_min: int
    break_recovery_bonus: float
    fatigue_scale: int

GoogleOAuthPayload ¶

Bases: BaseModel

Google OAuth client credentials saved by the desktop app.

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

class GoogleOAuthPayload(BaseModel):
    """Google OAuth client credentials saved by the desktop app."""

    client_id: str
    client_secret: str

GoogleOAuthStatus ¶

Bases: BaseModel

Configuration status for locally stored Google OAuth credentials.

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

class GoogleOAuthStatus(BaseModel):
    """Configuration status for locally stored Google OAuth credentials."""

    configured: bool
    client_id_preview: str | None = None
    credential_storage_available: bool = True
    client_secret_available: bool = False

PublishTargetPayload ¶

Bases: BaseModel

Requested publish target for schedule export actions.

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

class PublishTargetPayload(BaseModel):
    """Requested publish target for schedule export actions."""

    targets: list[PublishDestination] = Field(default_factory=list, max_length=3)

PublishTargetResponse ¶

Bases: BaseModel

Current publish target saved in local app settings.

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

class PublishTargetResponse(BaseModel):
    """Current publish target saved in local app settings."""

    targets: list[PublishDestination]

ShortcutSettingsPayload ¶

Bases: BaseModel

Keyboard shortcut override map keyed by frontend shortcut id.

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

class ShortcutSettingsPayload(BaseModel):
    """Keyboard shortcut override map keyed by frontend shortcut id."""

    shortcuts: dict[str, str] = Field(default_factory=dict)

ShortcutSettingsResponse ¶

Bases: BaseModel

Persisted keyboard shortcut override map returned to the frontend.

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

class ShortcutSettingsResponse(BaseModel):
    """Persisted keyboard shortcut override map returned to the frontend."""

    shortcuts: dict[str, str]

get_solver_settings ¶

get_solver_settings(db: Session) -> dict

Read solver settings from AppSettings table, falling back to defaults.

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

def get_solver_settings(db: Session) -> dict:
    """Read solver settings from AppSettings table, falling back to defaults."""
    result = {}
    for key, default in SOLVER_DEFAULTS.items():
        row = db.query(AppSettings).filter(AppSettings.key == key).first()
        if row and row.value is not None:
            # Coerce to the correct type based on the default
            try:
                if isinstance(default, float):
                    result[key] = float(row.value)
                elif isinstance(default, int):
                    result[key] = int(float(row.value))
                else:
                    result[key] = row.value
            except (ValueError, TypeError):
                result[key] = default
        else:
            result[key] = default
    return result

get_solver_settings_endpoint `async` ¶

get_solver_settings_endpoint(db: Session = Depends(get_db))

Get current solver tuning parameters.

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

@router.get("/solver", response_model=SolverSettingsResponse)
async def get_solver_settings_endpoint(db: Session = Depends(get_db)):
    """Get current solver tuning parameters."""
    s = get_solver_settings(db)
    return SolverSettingsResponse(
        max_time_seconds=s["solver_max_time_seconds"],
        break_threshold_min=s["solver_break_threshold_min"],
        break_recovery_bonus=s["solver_break_recovery_bonus"],
        fatigue_scale=s["solver_fatigue_scale"],
    )

set_solver_settings `async` ¶

set_solver_settings(payload: SolverSettingsPayload, db: Session = Depends(get_db))

Save solver tuning parameters.

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

@router.put("/solver", response_model=SolverSettingsResponse)
async def set_solver_settings(payload: SolverSettingsPayload, db: Session = Depends(get_db)):
    """Save solver tuning parameters."""
    mapping = {
        "solver_max_time_seconds": str(payload.max_time_seconds),
        "solver_break_threshold_min": str(payload.break_threshold_min),
        "solver_break_recovery_bonus": str(payload.break_recovery_bonus),
        "solver_fatigue_scale": str(payload.fatigue_scale),
    }
    for key, value in mapping.items():
        row = db.query(AppSettings).filter(AppSettings.key == key).first()
        if row:
            row.value = value
        else:
            db.add(AppSettings(key=key, value=value))
    db.commit()
    return SolverSettingsResponse(
        max_time_seconds=payload.max_time_seconds,
        break_threshold_min=payload.break_threshold_min,
        break_recovery_bonus=payload.break_recovery_bonus,
        fatigue_scale=payload.fatigue_scale,
    )

reset_solver_settings `async` ¶

reset_solver_settings(db: Session = Depends(get_db))

Reset solver parameters to defaults.

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

@router.delete("/solver")
async def reset_solver_settings(db: Session = Depends(get_db)):
    """Reset solver parameters to defaults."""
    db.query(AppSettings).filter(
        AppSettings.key.in_(list(SOLVER_DEFAULTS.keys()))
    ).delete(synchronize_session=False)
    db.commit()
    return {
        "status": "success",
        "message": "Solver settings reset to defaults",
        "defaults": {
            "max_time_seconds": SOLVER_DEFAULTS["solver_max_time_seconds"],
            "break_threshold_min": SOLVER_DEFAULTS["solver_break_threshold_min"],
            "break_recovery_bonus": SOLVER_DEFAULTS["solver_break_recovery_bonus"],
            "fatigue_scale": SOLVER_DEFAULTS["solver_fatigue_scale"],
        },
    }

get_google_oauth_status `async` ¶

get_google_oauth_status(db: Session = Depends(get_db))

Check whether Google OAuth credentials are configured.

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

@router.get("/google-oauth", response_model=GoogleOAuthStatus)
async def get_google_oauth_status(db: Session = Depends(get_db)):
    """Check whether Google OAuth credentials are configured."""
    configured, client_id, secret_available = google_oauth_configured(db)
    preview = _mask(client_id) if client_id else None

    return GoogleOAuthStatus(
        configured=configured,
        client_id_preview=preview,
        credential_storage_available=credential_store_available(),
        client_secret_available=secret_available,
    )

set_google_oauth `async` ¶

set_google_oauth(payload: GoogleOAuthPayload, db: Session = Depends(get_db))

Save or update Google OAuth credentials.

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

@router.put("/google-oauth", response_model=GoogleOAuthStatus)
async def set_google_oauth(payload: GoogleOAuthPayload, db: Session = Depends(get_db)):
    """Save or update Google OAuth credentials."""
    if not payload.client_id.strip() or not payload.client_secret.strip():
        raise HTTPException(status_code=400, detail="Both client_id and client_secret are required")

    try:
        set_google_oauth_credentials(
            db,
            payload.client_id.strip(),
            payload.client_secret.strip(),
        )
    except SecureCredentialStoreUnavailable as e:
        db.rollback()
        raise HTTPException(status_code=503, detail=str(e))

    db.commit()
    return GoogleOAuthStatus(
        configured=True,
        client_id_preview=_mask(payload.client_id.strip()),
        credential_storage_available=credential_store_available(),
        client_secret_available=True,
    )

delete_google_oauth `async` ¶

delete_google_oauth(db: Session = Depends(get_db))

Remove stored Google OAuth credentials.

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

@router.delete("/google-oauth")
async def delete_google_oauth(db: Session = Depends(get_db)):
    """Remove stored Google OAuth credentials."""
    try:
        delete_google_oauth_credentials(db)
    except SecureCredentialStoreUnavailable as e:
        raise HTTPException(status_code=503, detail=str(e))
    db.commit()
    return {"status": "success", "message": "Google OAuth credentials removed"}

get_publish_target `async` ¶

get_publish_target(db: Session = Depends(get_db))

Get the current publish target. Defaults to 'none'.

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

@router.get("/publish-target", response_model=PublishTargetResponse)
async def get_publish_target(db: Session = Depends(get_db)):
    """Get the current publish target. Defaults to 'none'."""
    row = db.query(AppSettings).filter(AppSettings.key == _KEY_PUBLISH_TARGET).first()
    targets = _normalise_publish_targets(row.value if row else None)
    return PublishTargetResponse(targets=targets)

set_publish_target `async` ¶

set_publish_target(payload: PublishTargetPayload, db: Session = Depends(get_db))

Set the publish target.

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

@router.put("/publish-target", response_model=PublishTargetResponse)
async def set_publish_target(payload: PublishTargetPayload, db: Session = Depends(get_db)):
    """Set the publish target."""
    targets = _normalise_publish_targets(payload.targets)
    if len(targets) != len(set(payload.targets)) or len(targets) != len(payload.targets):
        raise HTTPException(
            status_code=400,
            detail="targets must contain unique values from 'google', 'mp-backend', and 'pdf'",
        )
    stored = json.dumps(targets, separators=(",", ":"))
    row = db.query(AppSettings).filter(AppSettings.key == _KEY_PUBLISH_TARGET).first()
    if row:
        row.value = stored
    else:
        db.add(AppSettings(key=_KEY_PUBLISH_TARGET, value=stored))
    db.commit()
    return PublishTargetResponse(targets=targets)

get_shortcuts `async` ¶

get_shortcuts(db: Session = Depends(get_db))

Get keyboard shortcut overrides. Defaults are held by the frontend.

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

@router.get("/shortcuts", response_model=ShortcutSettingsResponse)
async def get_shortcuts(db: Session = Depends(get_db)):
    """Get keyboard shortcut overrides. Defaults are held by the frontend."""
    row = db.query(AppSettings).filter(AppSettings.key == _KEY_KEYBOARD_SHORTCUTS).first()
    return ShortcutSettingsResponse(shortcuts=_load_shortcut_overrides(row))

set_shortcuts `async` ¶

set_shortcuts(payload: ShortcutSettingsPayload, db: Session = Depends(get_db))

Save keyboard shortcut overrides.

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

@router.put("/shortcuts", response_model=ShortcutSettingsResponse)
async def set_shortcuts(payload: ShortcutSettingsPayload, db: Session = Depends(get_db)):
    """Save keyboard shortcut overrides."""
    shortcuts = _coerce_shortcut_map(payload.shortcuts)
    encoded = json.dumps(shortcuts, sort_keys=True)
    row = db.query(AppSettings).filter(AppSettings.key == _KEY_KEYBOARD_SHORTCUTS).first()
    if row:
        row.value = encoded
    else:
        db.add(AppSettings(key=_KEY_KEYBOARD_SHORTCUTS, value=encoded))
    db.commit()
    return ShortcutSettingsResponse(shortcuts=shortcuts)

reset_shortcuts `async` ¶

reset_shortcuts(db: Session = Depends(get_db))

Clear keyboard shortcut overrides.

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

@router.delete("/shortcuts")
async def reset_shortcuts(db: Session = Depends(get_db)):
    """Clear keyboard shortcut overrides."""
    db.query(AppSettings).filter(AppSettings.key == _KEY_KEYBOARD_SHORTCUTS).delete(synchronize_session=False)
    db.commit()
    return {"status": "success", "message": "Keyboard shortcuts reset to defaults"}

Data Management¶

data_management ¶

Data Management API - export, import, copy-from-event, delete-event, factory-reset.