inventory/crudkit/_sqlite.py
2025-09-08 11:56:29 -05:00

14 lines
512 B
Python

from __future__ import annotations
from sqlalchemy import event
from sqlalchemy.engine import Engine
def apply_sqlite_pragmas(engine: Engine, pragmas: dict[str, str]) -> None:
if not str(engine.url).startswith("sqlite://"):
return
@event.listens_for(engine, "connect")
def set_sqlite_pragma(dbapi_connection, connection_record):
cursor = dbapi_connection.cursor()
for key, value in pragmas.items():
cursor.execute(f"PRAGMA {key}={value}")
cursor.close()