app/api/deps.py

"""
API dependencies.
"""
from typing import Generator, Optional

from fastapi import Cookie, Header
from sqlalchemy.orm import Session

from app.core.database import get_db_for_session, get_default_session


def get_session_id(
    x_session_id: Optional[str] = Header(None),
    numidium_session: Optional[str] = Cookie(None)
) -> Optional[str]:
    """Return the session id from header or cookie."""
    return x_session_id or numidium_session


def get_scoped_db(
    x_session_id: Optional[str] = Header(None),
    numidium_session: Optional[str] = Cookie(None)
) -> Generator[Session, None, None]:
    """
    Provide a session-scoped DB if available, otherwise the default DB.
    """
    session_id = x_session_id or numidium_session
    if session_id:
        db = get_db_for_session(session_id)
    else:
        db = get_default_session()
    try:
        yield db
    finally:
        db.close()