Function _configure_sqlite_immediate_transactions

api/database.py:339–365

Configure engine for IMMEDIATE transactions via event hooks. Per SQLAlchemy docs: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html This replaces fragile pysqlite implicit transaction handling with explicit BEGIN IMMEDIATE at transaction start. Benefits: - Acquires write lock

(engine)

def _configure_sqlite_immediate_transactions(engine) -> None:
    """Configure engine for IMMEDIATE transactions via event hooks.

    Per SQLAlchemy docs: https://docs.sqlalchemy.org/en/20/dialects/sqlite.html

    This replaces fragile pysqlite implicit transaction handling with explicit
    BEGIN IMMEDIATE at transaction start. Benefits:
    - Acquires write lock immediately, preventing stale reads
    - Works correctly regardless of prior ORM operations
    - Future-proof: won't break when pysqlite legacy mode is removed in Python 3.16
    """
    @event.listens_for(engine, "connect")
    def do_connect(dbapi_connection, connection_record):
        # Disable pysqlite's implicit transaction handling
        dbapi_connection.isolation_level = None

        # Set busy_timeout on raw connection before any transactions
        cursor = dbapi_connection.cursor()
        try:
            cursor.execute("PRAGMA busy_timeout=30000")
        finally:
            cursor.close()

    @event.listens_for(engine, "begin")
    def do_begin(conn):
        # Use IMMEDIATE for all transactions to prevent stale reads
        conn.exec_driver_sql("BEGIN IMMEDIATE")


def create_database(project_dir: Path) -> tuple: