32 lines
857 B
Python
32 lines
857 B
Python
"""Database access for the FastAPI web app."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Iterator
|
|
from contextlib import contextmanager
|
|
from functools import lru_cache
|
|
|
|
from sqlalchemy.engine import Engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from pipelines.orm.common import get_postgres_engine
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def get_engine() -> Engine:
|
|
"""Return the lazily-created DATA_SCIENCE_DEV SQLAlchemy engine."""
|
|
return get_postgres_engine(name="DATA_SCIENCE_DEV")
|
|
|
|
|
|
def validate_database_connection() -> None:
|
|
"""Fail fast if the configured DATA_SCIENCE_DEV database is unavailable."""
|
|
with get_engine().connect():
|
|
pass
|
|
|
|
|
|
@contextmanager
|
|
def session_scope() -> Iterator[Session]:
|
|
"""Yield a SQLAlchemy session for a read-only request."""
|
|
with Session(get_engine()) as session:
|
|
yield session
|