refactor(fc2c-i): single shared async engine with lifecycle dispose

backend/app/__init__.py

@@ -21,4 +21,9 @@ def create_app() -> Quart:
     # Registered last so /api/* routes win over the SPA catch-all.
     app.register_blueprint(frontend_bp)
 
+    @app.after_serving
+    async def _dispose_db_engine() -> None:
+        from .extensions import dispose_engine
+        await dispose_engine()
+
     return app

backend/app/extensions.py

@@ -1,14 +1,52 @@
-"""Singleton extension instances; bound to the app in create_app()."""
+"""Process-wide singleton async engine + session helper.
 
-from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker, create_async_engine
+Previously every API blueprint lazily built its own engine via make_engine(),
+so the process ran N connection pools. This module owns ONE engine for the
+whole app; create_app() disposes it on shutdown via after_serving.
+"""
+
+from contextlib import asynccontextmanager
+
+from sqlalchemy.ext.asyncio import (
+    AsyncEngine,
+    async_sessionmaker,
+    create_async_engine,
+)
 
 from .config import get_config
 
-
+_engine: AsyncEngine | None = None
-def make_engine() -> AsyncEngine:
+_sessionmaker: async_sessionmaker | None = None
-    cfg = get_config()
-    return create_async_engine(cfg.database_url, pool_pre_ping=True, future=True)
 
 
-def make_session_factory(engine: AsyncEngine) -> async_sessionmaker:
+def get_engine() -> AsyncEngine:
-    return async_sessionmaker(engine, expire_on_commit=False)
+    global _engine, _sessionmaker
+    if _engine is None:
+        _engine = create_async_engine(
+            get_config().database_url, pool_pre_ping=True, future=True
+        )
+        _sessionmaker = async_sessionmaker(_engine, expire_on_commit=False)
+    return _engine
+
+
+def _get_sessionmaker() -> async_sessionmaker:
+    if _sessionmaker is None:
+        get_engine()
+    assert _sessionmaker is not None
+    return _sessionmaker
+
+
+@asynccontextmanager
+async def get_session():
+    """`async with get_session() as session:` — one shared pool."""
+    Session = _get_sessionmaker()
+    async with Session() as session:
+        yield session
+
+
+async def dispose_engine() -> None:
+    global _engine, _sessionmaker
+    if _engine is not None:
+        await _engine.dispose()
+    _engine = None
+    _sessionmaker = None

tests/test_extensions_session.py

@@ -0,0 +1,31 @@
+"""Unit: the shared engine is a memoized singleton; dispose resets it.
+
+create_async_engine does NOT open a connection, so identity checks need no
+DB and run in CI (not integration-marked).
+"""
+
+import pytest
+
+from backend.app import extensions
+
+
+@pytest.mark.asyncio
+async def test_get_engine_is_singleton():
+    a = extensions.get_engine()
+    b = extensions.get_engine()
+    assert a is b
+    await extensions.dispose_engine()
+
+
+@pytest.mark.asyncio
+async def test_dispose_engine_allows_fresh_engine():
+    first = extensions.get_engine()
+    await extensions.dispose_engine()
+    second = extensions.get_engine()
+    assert first is not second
+    await extensions.dispose_engine()
+
+
+def test_get_session_returns_async_context_manager():
+    cm = extensions.get_session()
+    assert hasattr(cm, "__aenter__") and hasattr(cm, "__aexit__")