fire-planner/fire_planner/api/dependencies.py

"""Shared FastAPI dependencies — DB session per request."""
from __future__ import annotations

from collections.abc import AsyncIterator

from fastapi import Request
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker


async def get_session(request: Request) -> AsyncIterator[AsyncSession]:
    """Yield an AsyncSession bound to the engine on app.state.

    The engine + session factory are wired up in `app.lifespan`. Tests
    swap them out via dependency_overrides.
    """
    factory: async_sessionmaker[AsyncSession] = request.app.state.session_factory
    async with factory() as session:
        yield session


async def get_wf_sync_session(request: Request) -> AsyncIterator[AsyncSession | None]:
    """Yield an AsyncSession bound to the wealthfolio_sync mirror.

    Powers the lazy refresh of `account_snapshot` (the disk cache for
    /networth). Yields `None` when the factory is not wired so endpoints
    can fall back to whatever is already cached — relevant for the in-memory
    SQLite test runs that don't wire wf_sync. Tests that DO want to exercise
    the refresh override this with their own factory via
    `app.dependency_overrides`.
    """
    factory: async_sessionmaker[AsyncSession] | None = getattr(
        request.app.state, "wf_sync_session_factory", None)
    if factory is None:
        yield None
        return
    async with factory() as session:
        yield session