app/db/engine.py

from sqlalchemy import create_engine, event
from sqlalchemy.orm import sessionmaker

from ..config import settings
from .models import Base

_connect_args = {"check_same_thread": False} if settings.db_url.startswith("sqlite") else {}
engine = create_engine(
    settings.db_url,
    connect_args=_connect_args,
    echo=False,
    future=True,
    pool_pre_ping=True,
)

if settings.db_url.startswith("sqlite"):
    @event.listens_for(engine, "connect")
    def _tune_sqlite(dbapi_connection, _):
        """
        Apply SQLite tuning pragmas on every new connection:

        * WAL journaling — better concurrent read/write, dramatically reduces
          ``database is locked`` errors when multiple requests hit history.
        * synchronous=NORMAL — safe with WAL, ~2x faster commits than FULL.
        * foreign_keys=ON — needed for our ON DELETE CASCADE relationships.
        * temp_store=MEMORY — temp B-trees in RAM, faster ordering / grouping.
        * cache_size=-20000 — 20 MB page cache (negative = KiB), big help for
          history pagination workload.
        """
        cursor = dbapi_connection.cursor()
        cursor.execute("PRAGMA journal_mode=WAL")
        cursor.execute("PRAGMA synchronous=NORMAL")
        cursor.execute("PRAGMA foreign_keys=ON")
        cursor.execute("PRAGMA temp_store=MEMORY")
        cursor.execute("PRAGMA cache_size=-20000")
        cursor.close()

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


def init_db() -> None:
    Base.metadata.create_all(engine)