"""FastAPI application factory.""" from __future__ import annotations import logging from contextlib import asynccontextmanager from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from app import db as dbmod from app.config import get_settings from app.routes import admin, chat, health, widget log = logging.getLogger(__name__) @asynccontextmanager async def lifespan(app: FastAPI): settings = get_settings() if not dbmod.is_initialized(): dbmod.init_engine(settings.database_url) if not settings.testing: # Dev convenience for sqlite; Postgres uses Alembic migrations. if dbmod.get_engine().dialect.name == "sqlite": await dbmod.create_all() # Warm the embedding model so the first chat isn't slow. try: from app.embeddings import embed_query await embed_query("warmup") except Exception: # noqa: BLE001 - warmup is best-effort log.warning("embedding warmup failed", exc_info=True) # Ensure a 'default' tenant exists + GDPR retention sweep at boot. try: from app.retention import purge_old_sessions from app.tenancy import ensure_default_tenant async with dbmod.get_sessionmaker()() as session: await ensure_default_tenant(session) await purge_old_sessions(session, settings.session_retention_days) await session.commit() except Exception: # noqa: BLE001 - best-effort log.warning("startup sweep failed", exc_info=True) yield def create_app() -> FastAPI: settings = get_settings() if not dbmod.is_initialized(): dbmod.init_engine(settings.database_url) if not settings.testing and not settings.admin_token: log.warning("ADMIN_TOKEN is empty — the admin panel/API will reject all requests") app = FastAPI(title="Shopify Support Bot", version="0.1.0", lifespan=lifespan) app.add_middleware( CORSMiddleware, allow_origins=settings.allowed_origins, allow_credentials=False, allow_methods=["*"], allow_headers=["*"], ) app.include_router(health.router) app.include_router(widget.router) app.include_router(chat.router) app.include_router(admin.router) return app app = create_app()