Forge-Engine/db/supports/support_connection.py

import contextlib
import sqlite3
from typing import TYPE_CHECKING

from .support_path import support_path

if TYPE_CHECKING:
    from flask import Flask


def connect_support(slug: str) -> sqlite3.Connection:
    """Connexion SQLite dédiée à un support.db — copie volontairement
    indépendante de db/connection.py::connect (mêmes réglages WAL/verrou/
    row_factory, système différent, "aucun import croisé avec le système
    jeu", voir docs/plan/PLAN.md)."""
    conn = sqlite3.connect(support_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é identique à db/connection.py::_track_for_teardown
    (voir son commentaire pour le raisonnement complet) — dupliqué plutôt
    qu'importé pour la même raison d'indépendance totale entre les deux
    systèmes."""
    try:
        from flask import g, has_app_context
    except ImportError:
        return
    if not has_app_context():
        return
    if not hasattr(g, "_forge_support_db_connections"):
        g._forge_support_db_connections = []
    g._forge_support_db_connections.append(conn)


def install_support_teardown_safety_net(app: "Flask") -> None:
    """Mirroir de db/connection.py::install_teardown_safety_net pour les
    connexions support.db — appelé séparément par
    core/db_teardown_guard.py."""

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

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