theo-agent-dashboard

Unnamed repository; edit this file 'description' to name the repository.
Log | Files | Refs | README

main.py (2199B)


      1 """Theo Agent Dashboard — FastAPI application."""
      2 from __future__ import annotations
      3 
      4 from contextlib import asynccontextmanager
      5 from collections.abc import AsyncIterator
      6 from pathlib import Path
      7 
      8 from fastapi import FastAPI
      9 from fastapi.middleware.cors import CORSMiddleware
     10 from fastapi.staticfiles import StaticFiles
     11 from fastapi.responses import FileResponse
     12 
     13 from app.db import Database
     14 from app.routes import register_routes
     15 from app.routes.topics import set_db
     16 from app.routes.attachments import set_db as set_attachment_db
     17 
     18 _DB: Database | None = None
     19 
     20 
     21 async def _init_db() -> Database:  # pragma: no mutate: block
     22     """Create and migrate the SQLite database."""
     23     db = Database("data/theo.db")
     24     await db.connect()
     25     migrations_dir = Path(__file__).parent / "migrations"
     26     await db.run_migrations(migrations_dir)
     27     return db
     28 
     29 
     30 @asynccontextmanager
     31 async def lifespan(app: FastAPI) -> AsyncIterator[None]:  # pragma: no mutate: block
     32     """Application lifespan: startup and shutdown."""
     33     global _DB  # noqa: PLW0603
     34     _DB = await _init_db()
     35     app.state.db = _DB
     36     set_db(_DB)
     37     set_attachment_db(_DB)
     38     yield
     39     if _DB:
     40         await _DB.close()
     41 
     42 
     43 app = FastAPI(
     44     title="Theo Agent Dashboard",
     45     version="0.1.0",
     46     lifespan=lifespan,
     47 )
     48 
     49 app.add_middleware(
     50     CORSMiddleware,
     51     allow_origins=["http://localhost:5173", "https://dash.packetloaf.privatedns.org"],
     52     allow_credentials=True,
     53     allow_methods=["*"],
     54     allow_headers=["*"],
     55 )
     56 
     57 register_routes(app)
     58 
     59 
     60 @app.get("/api/health")
     61 async def health() -> dict[str, str]:  # pragma: no mutate: block
     62     """Health check endpoint."""
     63     return {"status": "ok"}
     64 
     65 
     66 _STATIC_DIR = Path(__file__).parent.parent / "static"
     67 
     68 if _STATIC_DIR.is_dir():
     69     app.mount("/assets", StaticFiles(directory=str(_STATIC_DIR / "assets")), name="static-assets")
     70 
     71     @app.get("/{full_path:path}")
     72     async def serve_spa(full_path: str):  # pragma: no mutate: block
     73         """Serve frontend SPA — all non-API routes return index.html."""
     74         file_path = _STATIC_DIR / full_path
     75         if file_path.is_file():
     76             return FileResponse(file_path)
     77         return FileResponse(_STATIC_DIR / "index.html")