hs-data/apps/api/app/db/session.py

"""Async SQLAlchemy engine and session factory.

The engine is created lazily so that importing the FastAPI app (e.g. to export
the OpenAPI schema) never requires a live database connection.
"""

from __future__ import annotations

from functools import lru_cache

from sqlalchemy.ext.asyncio import (
    AsyncEngine,
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

from app.config import get_settings


@lru_cache
def get_engine() -> AsyncEngine:
    """Return a lazily-created, cached async engine."""
    settings = get_settings()
    return create_async_engine(settings.database_url, pool_pre_ping=True)


@lru_cache
def get_sessionmaker() -> async_sessionmaker[AsyncSession]:
    """Return a cached async session factory bound to the engine."""
    return async_sessionmaker(get_engine(), expire_on_commit=False)