presentation.routers.api.v1.holdings

src.presentation.routers.api.v1.holdings

Holdings resource handlers.

Handler functions for holdings (positions) management endpoints. Routes are registered via ROUTE_REGISTRY in routes/registry.py.

Handlers

list_holdings - List all holdings for user list_holdings_by_account - List holdings for an account sync_holdings - Sync holdings from provider

Reference

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

Classes

Functions

list_holdings async

list_holdings(
    request: Request,
    current_user: AuthenticatedUser,
    active_only: Annotated[
        bool,
        Query(description="Only return active holdings"),
    ] = True,
    asset_type: Annotated[
        str | None,
        Query(
            description="Filter by asset type (e.g., equity, etf, option)"
        ),
    ] = None,
    symbol: Annotated[
        str | None,
        Query(
            description="Filter by security symbol (e.g., AAPL)"
        ),
    ] = None,
    handler: ListHoldingsByUserHandler = Depends(
        handler_factory(ListHoldingsByUserHandler)
    ),
) -> HoldingListResponse | JSONResponse

List all holdings for the authenticated user.

GET /api/v1/holdings → 200 OK

Parameters:

Name	Type	Description	Default
request	Request	FastAPI request object.	required
current_user	AuthenticatedUser	Authenticated user (from JWT).	required
active_only	Annotated[bool, Query(description='Only return active holdings')]	Filter to only active holdings.	True
asset_type	Annotated[str | None, Query(description='Filter by asset type (e.g., equity, etf, option)')]	Filter by asset type.	None
symbol	Annotated[str | None, Query(description='Filter by security symbol (e.g., AAPL)')]	Filter by security symbol.	None
handler	ListHoldingsByUserHandler	List holdings handler (injected).	Depends(handler_factory(ListHoldingsByUserHandler))

Returns:

Type	Description
HoldingListResponse | JSONResponse	HoldingListResponse with list of holdings.
HoldingListResponse | JSONResponse	JSONResponse with RFC 9457 error on failure.

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

async def list_holdings(
    request: Request,
    current_user: AuthenticatedUser,
    active_only: Annotated[
        bool,
        Query(description="Only return active holdings"),
    ] = True,
    asset_type: Annotated[
        str | None,
        Query(description="Filter by asset type (e.g., equity, etf, option)"),
    ] = None,
    symbol: Annotated[
        str | None,
        Query(description="Filter by security symbol (e.g., AAPL)"),
    ] = None,
    handler: ListHoldingsByUserHandler = Depends(
        handler_factory(ListHoldingsByUserHandler)
    ),
) -> HoldingListResponse | JSONResponse:
    """List all holdings for the authenticated user.

    GET /api/v1/holdings → 200 OK

    Args:
        request: FastAPI request object.
        current_user: Authenticated user (from JWT).
        active_only: Filter to only active holdings.
        asset_type: Filter by asset type.
        symbol: Filter by security symbol.
        handler: List holdings handler (injected).

    Returns:
        HoldingListResponse with list of holdings.
        JSONResponse with RFC 9457 error on failure.
    """
    query = ListHoldingsByUser(
        user_id=current_user.user_id,
        active_only=active_only,
        asset_type=asset_type,
        symbol=symbol,
    )
    result = await handler.handle(query)

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

    return HoldingListResponse.from_dto(result.value)

list_holdings_by_account async

list_holdings_by_account(
    request: Request,
    current_user: AuthenticatedUser,
    account_id: Annotated[
        UUID, Path(description="Account UUID")
    ],
    active_only: Annotated[
        bool,
        Query(description="Only return active holdings"),
    ] = True,
    asset_type: Annotated[
        str | None,
        Query(
            description="Filter by asset type (e.g., equity, etf, option)"
        ),
    ] = None,
    handler: ListHoldingsByAccountHandler = Depends(
        handler_factory(ListHoldingsByAccountHandler)
    ),
) -> HoldingListResponse | JSONResponse

List holdings for a specific account.

GET /api/v1/accounts/{id}/holdings → 200 OK

Parameters:

Name	Type	Description	Default
request	Request	FastAPI request object.	required
current_user	AuthenticatedUser	Authenticated user (from JWT).	required
account_id	Annotated[UUID, Path(description='Account UUID')]	Account UUID.	required
active_only	Annotated[bool, Query(description='Only return active holdings')]	Filter to only active holdings.	True
asset_type	Annotated[str | None, Query(description='Filter by asset type (e.g., equity, etf, option)')]	Filter by asset type.	None
handler	ListHoldingsByAccountHandler	List holdings by account handler (injected).	Depends(handler_factory(ListHoldingsByAccountHandler))

Returns:

Type	Description
HoldingListResponse | JSONResponse	HoldingListResponse with list of holdings.
HoldingListResponse | JSONResponse	JSONResponse with RFC 9457 error on failure.

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

async def list_holdings_by_account(
    request: Request,
    current_user: AuthenticatedUser,
    account_id: Annotated[UUID, Path(description="Account UUID")],
    active_only: Annotated[
        bool,
        Query(description="Only return active holdings"),
    ] = True,
    asset_type: Annotated[
        str | None,
        Query(description="Filter by asset type (e.g., equity, etf, option)"),
    ] = None,
    handler: ListHoldingsByAccountHandler = Depends(
        handler_factory(ListHoldingsByAccountHandler)
    ),
) -> HoldingListResponse | JSONResponse:
    """List holdings for a specific account.

    GET /api/v1/accounts/{id}/holdings → 200 OK

    Args:
        request: FastAPI request object.
        current_user: Authenticated user (from JWT).
        account_id: Account UUID.
        active_only: Filter to only active holdings.
        asset_type: Filter by asset type.
        handler: List holdings by account handler (injected).

    Returns:
        HoldingListResponse with list of holdings.
        JSONResponse with RFC 9457 error on failure.
    """
    query = ListHoldingsByAccount(
        account_id=account_id,
        user_id=current_user.user_id,
        active_only=active_only,
        asset_type=asset_type,
    )
    result = await handler.handle(query)

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

    return HoldingListResponse.from_dto(result.value)

sync_holdings async

sync_holdings(
    request: Request,
    current_user: AuthenticatedUser,
    account_id: Annotated[
        UUID, Path(description="Account UUID")
    ],
    data: SyncHoldingsRequest,
    handler: SyncHoldingsHandler = Depends(
        handler_factory(SyncHoldingsHandler)
    ),
) -> SyncHoldingsResponse | JSONResponse

Sync holdings from provider for an account.

POST /api/v1/accounts/{id}/holdings/syncs → 201 Created

Fetches holdings from the provider and upserts them to the database. Returns sync statistics (created/updated/deactivated counts).

Parameters:

Name	Type	Description	Default
request	Request	FastAPI request object.	required
current_user	AuthenticatedUser	Authenticated user (from JWT).	required
account_id	Annotated[UUID, Path(description='Account UUID')]	Account UUID.	required
data	SyncHoldingsRequest	Sync request with force flag.	required
handler	SyncHoldingsHandler	Sync holdings handler (injected).	Depends(handler_factory(SyncHoldingsHandler))

Returns:

Type	Description
SyncHoldingsResponse | JSONResponse	SyncHoldingsResponse with sync statistics.
SyncHoldingsResponse | JSONResponse	JSONResponse with RFC 9457 error on failure.

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

async def sync_holdings(
    request: Request,
    current_user: AuthenticatedUser,
    account_id: Annotated[UUID, Path(description="Account UUID")],
    data: SyncHoldingsRequest,
    handler: SyncHoldingsHandler = Depends(handler_factory(SyncHoldingsHandler)),
) -> SyncHoldingsResponse | JSONResponse:
    """Sync holdings from provider for an account.

    POST /api/v1/accounts/{id}/holdings/syncs → 201 Created

    Fetches holdings from the provider and upserts them to the database.
    Returns sync statistics (created/updated/deactivated counts).

    Args:
        request: FastAPI request object.
        current_user: Authenticated user (from JWT).
        account_id: Account UUID.
        data: Sync request with force flag.
        handler: Sync holdings handler (injected).

    Returns:
        SyncHoldingsResponse with sync statistics.
        JSONResponse with RFC 9457 error on failure.
    """
    command = SyncHoldings(
        account_id=account_id,
        user_id=current_user.user_id,
        force=data.force,
    )
    result = await handler.handle(command)

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

    # Convert handler result to response
    sync_result = result.value
    return SyncHoldingsResponse(
        created=sync_result.created,
        updated=sync_result.updated,
        unchanged=sync_result.unchanged,
        errors=sync_result.errors,
        message=sync_result.message,
        holdings_created=sync_result.created,
        holdings_updated=sync_result.updated,
        holdings_deactivated=sync_result.deactivated,
    )