app/main.py (view raw)
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 |
from __future__ import annotations
import logging
from contextlib import asynccontextmanager
from datetime import UTC, datetime, timedelta
from pathlib import Path
from fastapi import FastAPI
from fastapi.responses import PlainTextResponse
from fastapi.staticfiles import StaticFiles
from sqlalchemy import text
from app.api import pages, routes
from app.collectors import import_all_sources
from app.config import get_settings
from app.db import get_engine, get_session_factory, init_db
from app.db.queries import count_trackers, latest_successful_import_at, set_app_state
from app.i18n import encode_bootstrap_note
from app.scheduler import start_scheduler, stop_scheduler
logger = logging.getLogger(__name__)
def _configure_logging(level: str) -> None:
# force=True: uvicorn configures logging before lifespan runs, so without it
# basicConfig is a no-op and app logs (scheduler, imports) never reach stdout.
logging.basicConfig(
level=getattr(logging, level.upper(), logging.INFO),
format="%(asctime)s %(levelname)s [%(name)s] %(message)s",
force=True,
)
async def _run_migrations() -> None:
"""Apply Alembic migrations; fall back to create_all if needed."""
try:
from alembic import command
from alembic.config import Config
cfg = Config(str(Path(__file__).resolve().parents[1] / "alembic.ini"))
settings = get_settings()
cfg.set_main_option("sqlalchemy.url", settings.database_url)
await __import__("asyncio").to_thread(command.upgrade, cfg, "head")
except Exception: # noqa: BLE001
logger.exception("Alembic upgrade failed; falling back to create_all")
await init_db()
@asynccontextmanager
async def lifespan(app: FastAPI):
settings = get_settings()
_configure_logging(settings.log_level)
get_engine(settings)
await _run_migrations()
factory = get_session_factory()
async with factory() as session:
total = await count_trackers(session)
last_import = await latest_successful_import_at(session)
max_age = timedelta(hours=settings.import_max_age_hours)
if last_import is not None and last_import.tzinfo is None:
last_import = last_import.replace(tzinfo=UTC)
stale = last_import is None or (datetime.now(UTC) - last_import) > max_age
if total == 0 or stale:
try:
await import_all_sources(session, settings)
await set_app_state(session, "bootstrap_note", "")
except Exception as exc: # noqa: BLE001
kind = "empty_import" if total == 0 else "stale_import"
logger.warning("bootstrap import failed (%s): %s", kind, exc)
await set_app_state(session, "bootstrap_note", encode_bootstrap_note(kind, exc))
if settings.run_scheduler:
start_scheduler(settings)
yield
stop_scheduler()
def create_app() -> FastAPI:
settings = get_settings()
app = FastAPI(
title=settings.app_name,
description=settings.app_description,
version=settings.app_version,
lifespan=lifespan,
)
static_dir = Path(__file__).resolve().parent / "static"
app.mount("/static", StaticFiles(directory=str(static_dir)), name="static")
app.include_router(pages.router)
app.include_router(routes.router, prefix="/api/v1")
@app.get("/healthz", response_class=PlainTextResponse)
async def healthz() -> str:
return "ok"
@app.get("/readyz", response_class=PlainTextResponse)
async def readyz() -> PlainTextResponse:
try:
engine = get_engine()
async with engine.connect() as conn:
await conn.execute(text("SELECT 1"))
return PlainTextResponse("ready")
except Exception: # noqa: BLE001
return PlainTextResponse("not ready", status_code=503)
return app
app = create_app()
def run() -> None:
import uvicorn
settings = get_settings()
uvicorn.run(
"app.main:app",
host=settings.host,
port=settings.port,
workers=1,
log_level=settings.log_level.lower(),
)
if __name__ == "__main__":
run()
|