e178ae406d
- SQLModel tables: flow_definitions/versions/runs, approval_tasks - Alembic setup with autogenerate (sqlmodel types fixed via script.py.mako) - Settings extended (shared root .env), db engine/session helpers - pytest suite: conftest with isolated test DB, 5 model tests green
36 lines
988 B
Python
36 lines
988 B
Python
"""Database engine, sessions, and FastAPI dependency."""
|
|
|
|
from collections.abc import Iterator
|
|
|
|
from sqlalchemy import Engine
|
|
from sqlmodel import Session, SQLModel, create_engine
|
|
|
|
from .settings import settings
|
|
|
|
_engine: Engine | None = None
|
|
|
|
|
|
def get_engine(url: str | None = None) -> Engine:
|
|
"""Process-wide engine singleton (override URL in tests via settings)."""
|
|
global _engine
|
|
if _engine is None:
|
|
_engine = create_engine(url or settings.database_url)
|
|
return _engine
|
|
|
|
|
|
def reset_engine() -> None:
|
|
"""Drop the cached engine (used by the test suite between DB targets)."""
|
|
global _engine
|
|
if _engine is not None:
|
|
_engine.dispose()
|
|
_engine = None
|
|
|
|
|
|
def get_session() -> Iterator[Session]:
|
|
with Session(get_engine()) as session:
|
|
yield session
|
|
|
|
|
|
def create_all(url: str | None = None) -> None:
|
|
"""Create tables directly (dev convenience; migrations own prod schema)."""
|
|
SQLModel.metadata.create_all(get_engine(url))
|