database/sqlalchemy_store.py

"""SQLAlchemy connection helpers for reading SQLite market data."""

from __future__ import annotations

from contextlib import contextmanager
from sqlalchemy import create_engine, event


def _engine(database_url: str):
    engine = create_engine(database_url, connect_args={"timeout": 30})

    @event.listens_for(engine, "connect")
    def _configure_sqlite(dbapi_connection, _):
        cursor = dbapi_connection.cursor()
        cursor.execute("PRAGMA foreign_keys = ON")
        cursor.execute("PRAGMA journal_mode = WAL")
        cursor.execute("PRAGMA busy_timeout = 30000")
        cursor.close()

    return engine


@contextmanager
def read_connection(database_url: str):
    engine = _engine(database_url)
    try:
        with engine.connect() as connection:
            yield connection
    finally:
        engine.dispose()