Forge-Engine/db/connection.py

import contextlib
import sqlite3
from typing import TYPE_CHECKING

from .db_path import db_path

if TYPE_CHECKING:
    from flask import Flask


def connect(slug: str) -> sqlite3.Connection:
    # timeout=10 : si une autre connexion tient un verrou d'écriture au même
    # instant (deux requêtes qui arrivent presque en même temps, ex. deux
    # onglets, ou le navigateur qui recharge plusieurs ressources), sqlite3
    # réessaie pendant 10 secondes au lieu de lever immédiatement "database
    # is locked". Le mode WAL (Write-Ahead Logging) va plus loin : il permet
    # à des lectures de se faire PENDANT qu'une écriture est en cours
    # ailleurs, ce qui est la cause la plus fréquente de ce blocage avec le
    # serveur de développement Flask.
    conn = sqlite3.connect(db_path(slug), timeout=10)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA foreign_keys = ON")
    conn.execute("PRAGMA journal_mode = WAL")
    conn.execute("PRAGMA busy_timeout = 8000")
    _track_for_teardown(conn)
    return conn


def _track_for_teardown(conn: sqlite3.Connection) -> None:
    """Filet de sécurité : chaque fonction de db/ ouvre sa propre connexion
    et est censée la fermer elle-même (conn.close()) avant de rendre la
    main — mais si une exception survient ENTRE l'ouverture et cette
    fermeture (une erreur de programmation, une contrainte violée...), le
    conn.close() prévu n'est jamais atteint. En mode debug (voir app.py),
    le débogueur Werkzeug garde alors la trace complète de l'erreur en
    mémoire pour l'inspection interactive — ce qui inclut la variable
    locale `conn`, empêchant le ramasse-miettes Python de la libérer et
    donc SQLite de relâcher son verrou d'écriture. Toute requête suivante
    qui écrit se heurte alors à "database is locked" jusqu'au redémarrage
    du serveur, même longtemps après l'erreur d'origine. En enregistrant
    ici la connexion sur le contexte de la requête Flask en cours (quand il
    y en a un), on garantit sa fermeture à la fin de la requête via
    _close_leaked_connections ci-dessous, que la requête ait réussi ou
    planté — sans rien changer au comportement des ~80 fonctions qui
    ferment déjà correctement leur connexion (fermer une connexion SQLite
    déjà fermée ne fait rien)."""
    try:
        from flask import g, has_app_context
    except ImportError:
        return
    if not has_app_context():
        return
    if not hasattr(g, "_forge_db_connections"):
        g._forge_db_connections = []
    g._forge_db_connections.append(conn)


def install_teardown_safety_net(app: "Flask") -> None:
    """Enregistre le filet de sécurité (voir _track_for_teardown) sur
    l'appli Flask passée en paramètre — jamais importée ici : `db/` est la
    couche la plus basse du moteur (voir pyproject.toml, contrat
    import-linter) et ne doit dépendre d'aucun autre paquet. C'est
    core/db_teardown_guard.py, dans la couche de câblage, qui appelle
    cette fonction avec `core.flask_app.app`."""

    @app.teardown_request
    def _close_leaked_connections(exception: BaseException | None = None) -> None:  # noqa: ARG001 - signature imposée par Flask
        from flask import g

        for conn in getattr(g, "_forge_db_connections", ()):
            with contextlib.suppress(sqlite3.Error):
                conn.close()