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.

    The real funnel tables live in the ``ads`` schema; the ORM models are
    schema-less, so we point every connection's ``search_path`` at the
    configured schema (asyncpg ``server_settings``). Skipped when blank.
    """
    settings = get_settings()
    connect_args: dict = {}
    if settings.db_schema:
        connect_args["server_settings"] = {"search_path": settings.db_schema}
    return create_async_engine(
        settings.database_url, pool_pre_ping=True, connect_args=connect_args
    )


@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)