backend/app/database.py

"""Couche base de données — SQLite via SQLAlchemy 2.0.

POC : SQLite, zéro infrastructure. La couche est volontairement isolée
pour pouvoir migrer vers PostgreSQL + PostGIS + pgvector plus tard sans
toucher au reste de l'application.
"""
from collections.abc import Generator

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker

from . import config


class Base(DeclarativeBase):
    pass


# check_same_thread n'est valable que pour SQLite. pool_pre_ping évite les
# connexions Postgres mortes (utile sur Neon/serverless qui se met en veille).
_is_sqlite = config.DATABASE_URL.startswith("sqlite")
engine = create_engine(
    config.DATABASE_URL,
    connect_args={"check_same_thread": False} if _is_sqlite else {},
    pool_pre_ping=not _is_sqlite,
)

# Audit (round 3) : SQLite ignore les FK par défaut → les violations d'intégrité
# référentielle passaient inaperçues en test alors qu'elles sont FATALES sur Neon
# Postgres (NO ACTION). On force PRAGMA foreign_keys=ON pour que les tests
# reproduisent fidèlement la prod et attrapent les régressions de suppression.
if _is_sqlite:
    from sqlalchemy import event

    @event.listens_for(engine, "connect")
    def _sqlite_fk_pragma(dbapi_connection, connection_record):  # noqa: ANN001
        cur = dbapi_connection.cursor()
        cur.execute("PRAGMA foreign_keys=ON")
        cur.close()

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)


def init_db() -> None:
    """Met la base à jour via Alembic (migrations propres, plus de CHATS_RESET_DB).

    - Base vierge → applique toutes les migrations (crée les tables).
    - Schéma existant sans Alembic (ex. ancienne base create_all) → l'« adopte »
      (stamp) sans tout recréer.
    - Schéma en retard → applique les migrations manquantes (ALTER, etc.).
    CHATS_RESET_DB=1 reste dispo en secours (⚠️ efface tout).
    """
    import os

    from . import models  # noqa: F401  (enregistre les modèles sur Base)

    if os.environ.get("CHATS_RESET_DB", "").lower() in {"1", "true", "yes"}:
        Base.metadata.drop_all(bind=engine)
        _drop_alembic_version()
    _run_migrations()


def _drop_alembic_version() -> None:
    from sqlalchemy import text

    with engine.begin() as conn:
        conn.execute(text("DROP TABLE IF EXISTS alembic_version"))


def _run_migrations() -> None:
    from pathlib import Path

    from alembic import command
    from alembic.config import Config
    from alembic.runtime.migration import MigrationContext
    from sqlalchemy import inspect

    base = Path(__file__).resolve().parent.parent  # .../backend
    cfg = Config(str(base / "alembic.ini"))
    cfg.set_main_option("script_location", str(base / "migrations"))
    try:
        with engine.connect() as conn:
            current = MigrationContext.configure(conn).get_current_revision()
        has_tables = inspect(engine).has_table("cats")
        if current is None and has_tables:
            # Schéma déjà en place mais jamais versionné : c'est forcément une
            # ancienne base create_all figée au schéma INITIAL (787cf3d14385,
            # phone VARCHAR(32)). La stamper directement à « head » lui ferait
            # sauter tous les ALTER intermédiaires (dont l'élargissement de phone
            # en Text). On stampe donc la révision de BASE puis on applique le
            # retard normalement.
            base_rev = _base_revision(cfg) or "787cf3d14385"
            command.stamp(cfg, base_rev)      # adopter le schéma initial
            command.upgrade(cfg, "head")      # rattraper les ALTER manquants
        else:
            command.upgrade(cfg, "head")      # vierge → crée tout ; sinon applique le retard
    except Exception:
        Base.metadata.create_all(bind=engine)  # filet de sécurité


def _base_revision(cfg) -> str | None:  # noqa: ANN001 (Config est un type Alembic)
    """Retourne la première révision de l'historique (down_revision is None)."""
    from alembic.script import ScriptDirectory

    script = ScriptDirectory.from_config(cfg)
    bases = script.get_bases()
    return bases[0] if bases else None


def get_db() -> Generator[Session, None, None]:
    """Dépendance FastAPI : fournit une session et la ferme proprement."""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()