presentation.routers.api.v1.balance_snapshots

src.presentation.routers.api.v1.balance_snapshots

Balance snapshots resource handlers.

Handler functions for balance snapshot/history endpoints. Routes are registered via ROUTE_REGISTRY in routes/registry.py.

Handlers

get_latest_snapshots - Get latest snapshots for user get_balance_history - Get balance history for account list_balance_snapshots - List balance snapshots for account

Reference

• docs/architecture/api-design-patterns.md
• docs/architecture/error-handling-architecture.md

Classes

Functions

get_latest_snapshots async

get_latest_snapshots(
    request: Request,
    current_user: AuthenticatedUser,
    handler: GetLatestBalanceSnapshotsHandler = Depends(
        handler_factory(GetLatestBalanceSnapshotsHandler)
    ),
) -> LatestSnapshotsResponse | JSONResponse

Get latest balance snapshot for each account.

GET /api/v1/balance-snapshots → 200 OK

Returns portfolio summary with one snapshot per account.

Source code in src/presentation/routers/api/v1/balance_snapshots.py

async def get_latest_snapshots(
    request: Request,
    current_user: AuthenticatedUser,
    handler: GetLatestBalanceSnapshotsHandler = Depends(
        handler_factory(GetLatestBalanceSnapshotsHandler)
    ),
) -> LatestSnapshotsResponse | JSONResponse:
    """Get latest balance snapshot for each account.

    GET /api/v1/balance-snapshots → 200 OK

    Returns portfolio summary with one snapshot per account.
    """
    query = GetLatestBalanceSnapshots(user_id=current_user.user_id)
    result = await handler.handle(query)

    if isinstance(result, Failure):
        app_error = _map_snapshot_error(result.error)
        return ErrorResponseBuilder.from_application_error(
            error=app_error,
            request=request,
            trace_id=get_trace_id() or "",
        )

    return LatestSnapshotsResponse.from_dto(result.value)

get_balance_history async

get_balance_history(
    request: Request,
    current_user: AuthenticatedUser,
    account_id: Annotated[
        UUID, Path(description="Account UUID")
    ],
    start_date: Annotated[
        datetime, Query(description="Start of date range")
    ],
    end_date: Annotated[
        datetime, Query(description="End of date range")
    ],
    source: Annotated[
        str | None,
        Query(
            description="Filter by snapshot source (account_sync, manual_sync, etc.)"
        ),
    ] = None,
    handler: GetBalanceHistoryHandler = Depends(
        handler_factory(GetBalanceHistoryHandler)
    ),
) -> BalanceHistoryResponse | JSONResponse

Get balance history for an account.

GET /api/v1/accounts/{id}/balance-history → 200 OK

Returns snapshots ordered chronologically for charting.

Source code in src/presentation/routers/api/v1/balance_snapshots.py

async def get_balance_history(
    request: Request,
    current_user: AuthenticatedUser,
    account_id: Annotated[UUID, Path(description="Account UUID")],
    start_date: Annotated[datetime, Query(description="Start of date range")],
    end_date: Annotated[datetime, Query(description="End of date range")],
    source: Annotated[
        str | None,
        Query(
            description="Filter by snapshot source (account_sync, manual_sync, etc.)"
        ),
    ] = None,
    handler: GetBalanceHistoryHandler = Depends(
        handler_factory(GetBalanceHistoryHandler)
    ),
) -> BalanceHistoryResponse | JSONResponse:
    """Get balance history for an account.

    GET /api/v1/accounts/{id}/balance-history → 200 OK

    Returns snapshots ordered chronologically for charting.
    """
    query = GetBalanceHistory(
        account_id=account_id,
        user_id=current_user.user_id,
        start_date=start_date,
        end_date=end_date,
        source=source,
    )
    result = await handler.handle(query)

    if isinstance(result, Failure):
        app_error = _map_snapshot_error(result.error)
        return ErrorResponseBuilder.from_application_error(
            error=app_error,
            request=request,
            trace_id=get_trace_id() or "",
        )

    return BalanceHistoryResponse.from_dto(result.value)

list_balance_snapshots async

list_balance_snapshots(
    request: Request,
    current_user: AuthenticatedUser,
    account_id: Annotated[
        UUID, Path(description="Account UUID")
    ],
    limit: Annotated[
        int | None,
        Query(
            description="Maximum number of snapshots to return",
            ge=1,
            le=100,
        ),
    ] = 30,
    source: Annotated[
        str | None,
        Query(description="Filter by snapshot source"),
    ] = None,
    handler: ListBalanceSnapshotsByAccountHandler = Depends(
        handler_factory(
            ListBalanceSnapshotsByAccountHandler
        )
    ),
) -> BalanceHistoryResponse | JSONResponse

List recent balance snapshots for an account.

GET /api/v1/accounts/{id}/balance-snapshots → 200 OK

Returns snapshots ordered by captured_at descending (most recent first).

Source code in src/presentation/routers/api/v1/balance_snapshots.py

async def list_balance_snapshots(
    request: Request,
    current_user: AuthenticatedUser,
    account_id: Annotated[UUID, Path(description="Account UUID")],
    limit: Annotated[
        int | None,
        Query(description="Maximum number of snapshots to return", ge=1, le=100),
    ] = 30,
    source: Annotated[
        str | None,
        Query(description="Filter by snapshot source"),
    ] = None,
    handler: ListBalanceSnapshotsByAccountHandler = Depends(
        handler_factory(ListBalanceSnapshotsByAccountHandler)
    ),
) -> BalanceHistoryResponse | JSONResponse:
    """List recent balance snapshots for an account.

    GET /api/v1/accounts/{id}/balance-snapshots → 200 OK

    Returns snapshots ordered by captured_at descending (most recent first).
    """
    query = ListBalanceSnapshotsByAccount(
        account_id=account_id,
        user_id=current_user.user_id,
        limit=limit,
        source=source,
    )
    result = await handler.handle(query)

    if isinstance(result, Failure):
        app_error = _map_snapshot_error(result.error)
        return ErrorResponseBuilder.from_application_error(
            error=app_error,
            request=request,
            trace_id=get_trace_id() or "",
        )

    return BalanceHistoryResponse.from_dto(result.value)