| """
|
| Database connection and session management.
|
| Ensures SQLite WAL mode and foreign key constraint enforcement.
|
| """
|
|
|
| import os
|
| from collections.abc import Generator
|
|
|
| from sqlalchemy import create_engine, event
|
| from sqlalchemy.engine import Engine
|
| from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
|
|
|
| DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./placement_engine.db")
|
|
|
|
|
| connect_args = {}
|
| if DATABASE_URL.startswith("sqlite"):
|
| connect_args = {"check_same_thread": False}
|
|
|
| engine = create_engine(
|
| DATABASE_URL,
|
| connect_args=connect_args,
|
| echo=os.getenv("DEBUG", "false").lower() == "true",
|
| )
|
|
|
|
|
|
|
| @event.listens_for(Engine, "connect")
|
| def set_sqlite_pragma(dbapi_connection: object, connection_record: object) -> None:
|
|
|
| if hasattr(dbapi_connection, "cursor"):
|
| cursor = dbapi_connection.cursor()
|
| cursor.execute("PRAGMA foreign_keys=ON")
|
| cursor.execute("PRAGMA journal_mode=WAL")
|
| cursor.execute("PRAGMA synchronous=NORMAL")
|
| cursor.close()
|
|
|
|
|
| SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
|
|
|
| class Base(DeclarativeBase):
|
| pass
|
|
|
|
|
| def get_db() -> Generator[Session, None, None]:
|
| """Dependency for API route handlers."""
|
| db = SessionLocal()
|
| try:
|
| yield db
|
| finally:
|
| db.close()
|
|
|
