home-automation/app/main.py

from contextlib import asynccontextmanager
from pathlib import Path

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.interval import IntervalTrigger
from sqlalchemy.orm import Session

from app import models  # noqa: F401
from app.api.routes.auth import router as auth_router
from app.api.routes import pages, status
import app.auth_db as auth_db
from app.api.routes.homeassistant import router as homeassistant_router
from app.api.routes.location import router as location_router
from app.api.routes.poo import router as poo_router
from app.api.routes.public_ip import router as public_ip_router
from app.api.routes.ticktick import router as ticktick_router
from app.config import get_settings
from app.services.auth import AuthBootstrapError, initialize_auth_schema
from app.services.config_page import seed_missing_config_from_bootstrap, sync_app_hostname_from_bootstrap
from app.services.public_ip import check_public_ipv4_and_notify
from scripts.app_db_adopt import AppDatabaseAdoptionError, validate_app_runtime_db
from scripts.location_db_adopt import LocationDatabaseAdoptionError, validate_location_runtime_db
from scripts.poo_db_adopt import PooDatabaseAdoptionError, validate_poo_runtime_db


def _run_scheduled_public_ip_check() -> None:
    session_local = auth_db.get_auth_session_local()
    session: Session = session_local()
    try:
        check_public_ipv4_and_notify(session, bootstrap_settings=get_settings())
    finally:
        session.close()


def ensure_auth_db_ready() -> None:
    session_local = auth_db.get_auth_session_local()
    session: Session = session_local()
    try:
        validate_app_runtime_db(get_settings().app_database_url)
        initialize_auth_schema(session, get_settings())
        seed_missing_config_from_bootstrap(session, get_settings())
        sync_app_hostname_from_bootstrap(session, get_settings())
    except AppDatabaseAdoptionError as exc:
        raise RuntimeError(str(exc)) from exc
    except AuthBootstrapError as exc:
        raise RuntimeError(str(exc)) from exc
    finally:
        session.close()


def ensure_location_db_ready() -> None:
    settings = get_settings()
    if settings.location_sqlite_path is None:
        return

    try:
        validate_location_runtime_db(settings.location_database_url)
    except LocationDatabaseAdoptionError as exc:
        raise RuntimeError(str(exc)) from exc


def ensure_poo_db_ready() -> None:
    settings = get_settings()
    if settings.poo_sqlite_path is None:
        return

    try:
        validate_poo_runtime_db(settings.poo_database_url)
    except PooDatabaseAdoptionError as exc:
        raise RuntimeError(str(exc)) from exc


def ensure_runtime_dirs() -> None:
    settings = get_settings()
    for path in (settings.app_sqlite_path, settings.location_sqlite_path, settings.poo_sqlite_path):
        if path is not None:
            path.parent.mkdir(parents=True, exist_ok=True)


@asynccontextmanager
async def lifespan(_: FastAPI):
    ensure_runtime_dirs()
    ensure_auth_db_ready()
    ensure_location_db_ready()
    ensure_poo_db_ready()
    scheduler = BackgroundScheduler(timezone="UTC")
    scheduler.add_job(
        _run_scheduled_public_ip_check,
        trigger=IntervalTrigger(hours=4),
        id="public-ip-check",
        replace_existing=True,
        max_instances=1,
        coalesce=True,
    )
    scheduler.start()
    yield
    scheduler.shutdown(wait=False)


def create_app() -> FastAPI:
    settings = get_settings()
    app = FastAPI(
        title=settings.app_name,
        debug=settings.app_debug,
        version="0.1.0",
        lifespan=lifespan,
        description=(
            "Home automation backend with auth, runtime config, Home Assistant "
            "integrations, TickTick integration, and SQLite-backed recorders."
        ),
    )

    static_dir = Path(__file__).parent / "static"
    app.mount("/static", StaticFiles(directory=static_dir), name="static")

    app.include_router(status.router)
    app.include_router(auth_router)
    app.include_router(pages.router)
    app.include_router(homeassistant_router)
    app.include_router(location_router)
    app.include_router(poo_router)
    app.include_router(public_ip_router)
    app.include_router(ticktick_router)
    return app


app = create_app()