Skip to content

infrastructure.providers.schwab.api.accounts_api

src.infrastructure.providers.schwab.api.accounts_api

Schwab Accounts API client.

HTTP client for Schwab Trader API accounts endpoints. Handles HTTP concerns only - returns raw JSON responses.

Endpoints

GET /trader/v1/accounts - Get all linked accounts with balances/positions GET /trader/v1/accounts/{accountNumber} - Get specific account

Reference
  • Schwab Trader API: https://developer.schwab.com
  • docs/architecture/provider-integration-architecture.md

Attributes

Classes

SchwabAccountsAPI

Bases: BaseProviderAPIClient

HTTP client for Schwab Trader API accounts endpoints.

Extends BaseProviderAPIClient with Schwab-specific authentication. Returns raw JSON responses - mapping to domain types is done by mappers.

Thread-safe: Uses httpx.AsyncClient per-request (no shared state).

Example

api = SchwabAccountsAPI( ... base_url="https://api.schwabapi.com/trader/v1", ... timeout=30.0, ... ) result = await api.get_accounts(access_token="...") match result: ... case Success(data): ... print(f"Got {len(data)} accounts") ... case Failure(error): ... print(f"Error: {error.message}")

Source code in src/infrastructure/providers/schwab/api/accounts_api.py 
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
class SchwabAccountsAPI(BaseProviderAPIClient):
    """HTTP client for Schwab Trader API accounts endpoints.

    Extends BaseProviderAPIClient with Schwab-specific authentication.
    Returns raw JSON responses - mapping to domain types is done by mappers.

    Thread-safe: Uses httpx.AsyncClient per-request (no shared state).

    Example:
        >>> api = SchwabAccountsAPI(
        ...     base_url="https://api.schwabapi.com/trader/v1",
        ...     timeout=30.0,
        ... )
        >>> result = await api.get_accounts(access_token="...")
        >>> match result:
        ...     case Success(data):
        ...         print(f"Got {len(data)} accounts")
        ...     case Failure(error):
        ...         print(f"Error: {error.message}")
    """

    def __init__(
        self,
        *,
        base_url: str,
        timeout: float = PROVIDER_TIMEOUT_DEFAULT,
    ) -> None:
        """Initialize Schwab Accounts API client.

        Args:
            base_url: Schwab Trader API base URL (e.g., "https://api.schwabapi.com/trader/v1").
            timeout: HTTP request timeout in seconds.
        """
        super().__init__(
            base_url=base_url,
            provider_name="schwab",
            timeout=timeout,
        )

    async def get_accounts(
        self,
        access_token: str,
        *,
        include_positions: bool = False,
    ) -> Result[list[dict[str, Any]], ProviderError]:
        """Fetch all accounts for the authenticated user.

        Args:
            access_token: Valid Schwab access token (Bearer token).
            include_positions: Include position details in response.

        Returns:
            Success(list[dict]): List of account JSON objects from Schwab.
            Failure(ProviderAuthenticationError): If token is invalid/expired.
            Failure(ProviderRateLimitError): If rate limit exceeded.
            Failure(ProviderUnavailableError): If Schwab API is unreachable.
            Failure(ProviderInvalidResponseError): If response is malformed.
        """
        self._logger.debug(
            "schwab_accounts_api_get_accounts_started",
            include_positions=include_positions,
        )

        params: dict[str, str] | None = (
            {"fields": "positions"} if include_positions else None
        )

        return await self._execute_and_parse_list(
            method="GET",
            path="/accounts",
            headers=self._build_headers(access_token),
            params=params,
            operation="get_accounts",
        )

    async def get_account(
        self,
        access_token: str,
        account_number: str,
        *,
        include_positions: bool = False,
    ) -> Result[dict[str, Any], ProviderError]:
        """Fetch a specific account by account number.

        Args:
            access_token: Valid Schwab access token (Bearer token).
            account_number: Schwab account number to fetch.
            include_positions: Include position details in response.

        Returns:
            Success(dict): Single account JSON object from Schwab.
            Failure(ProviderAuthenticationError): If token is invalid/expired.
            Failure(ProviderRateLimitError): If rate limit exceeded.
            Failure(ProviderUnavailableError): If Schwab API is unreachable.
            Failure(ProviderInvalidResponseError): If response is malformed.
        """
        self._logger.debug(
            "schwab_accounts_api_get_account_started",
            account_number_masked=f"****{account_number[-4:]}"
            if len(account_number) >= 4
            else "****",
            include_positions=include_positions,
        )

        params: dict[str, str] | None = (
            {"fields": "positions"} if include_positions else None
        )

        return await self._execute_and_parse_object(
            method="GET",
            path=f"/accounts/{account_number}",
            headers=self._build_headers(access_token),
            params=params,
            operation="get_account",
        )

    def _build_headers(self, access_token: str) -> dict[str, str]:
        """Build HTTP headers for Schwab API requests.

        Args:
            access_token: Bearer token for authentication.

        Returns:
            Headers dict with Authorization and Accept.
        """
        return {
            "Authorization": f"{BEARER_PREFIX}{access_token}",
            "Accept": "application/json",
        }
Functions
__init__ 
__init__(
    *,
    base_url: str,
    timeout: float = PROVIDER_TIMEOUT_DEFAULT
) -> None

Parameters:

Name Type Description Default
base_url str

Schwab Trader API base URL (e.g., "https://api.schwabapi.com/trader/v1").

 required
timeout float

HTTP request timeout in seconds.

 PROVIDER_TIMEOUT_DEFAULT
Source code in src/infrastructure/providers/schwab/api/accounts_api.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def __init__(
    self,
    *,
    base_url: str,
    timeout: float = PROVIDER_TIMEOUT_DEFAULT,
) -> None:
    """Initialize Schwab Accounts API client.

    Args:
        base_url: Schwab Trader API base URL (e.g., "https://api.schwabapi.com/trader/v1").
        timeout: HTTP request timeout in seconds.
    """
    super().__init__(
        base_url=base_url,
        provider_name="schwab",
        timeout=timeout,
    )
get_accounts async 
get_accounts(
    access_token: str, *, include_positions: bool = False
) -> Result[list[dict[str, Any]], ProviderError]

Fetch all accounts for the authenticated user.

Parameters:

Name Type Description Default
access_token str

Valid Schwab access token (Bearer token).

 required
include_positions bool

Include position details in response.

 False

Returns:

Name Type Description
Success list[dict]

List of account JSON objects from Schwab.
Failure ProviderAuthenticationError

If token is invalid/expired.
Failure ProviderRateLimitError

If rate limit exceeded.
Failure ProviderUnavailableError

If Schwab API is unreachable.
Failure ProviderInvalidResponseError

If response is malformed.
Source code in src/infrastructure/providers/schwab/api/accounts_api.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
async def get_accounts(
    self,
    access_token: str,
    *,
    include_positions: bool = False,
) -> Result[list[dict[str, Any]], ProviderError]:
    """Fetch all accounts for the authenticated user.

    Args:
        access_token: Valid Schwab access token (Bearer token).
        include_positions: Include position details in response.

    Returns:
        Success(list[dict]): List of account JSON objects from Schwab.
        Failure(ProviderAuthenticationError): If token is invalid/expired.
        Failure(ProviderRateLimitError): If rate limit exceeded.
        Failure(ProviderUnavailableError): If Schwab API is unreachable.
        Failure(ProviderInvalidResponseError): If response is malformed.
    """
    self._logger.debug(
        "schwab_accounts_api_get_accounts_started",
        include_positions=include_positions,
    )

    params: dict[str, str] | None = (
        {"fields": "positions"} if include_positions else None
    )

    return await self._execute_and_parse_list(
        method="GET",
        path="/accounts",
        headers=self._build_headers(access_token),
        params=params,
        operation="get_accounts",
    )
get_account async 
get_account(
    access_token: str,
    account_number: str,
    *,
    include_positions: bool = False
) -> Result[dict[str, Any], ProviderError]

Fetch a specific account by account number.

Parameters:

Name Type Description Default
access_token str

Valid Schwab access token (Bearer token).

 required
account_number str

Schwab account number to fetch.

 required
include_positions bool

Include position details in response.

 False

Returns:

Name Type Description
Success dict

Single account JSON object from Schwab.
Failure ProviderAuthenticationError

If token is invalid/expired.
Failure ProviderRateLimitError

If rate limit exceeded.
Failure ProviderUnavailableError

If Schwab API is unreachable.
Failure ProviderInvalidResponseError

If response is malformed.
Source code in src/infrastructure/providers/schwab/api/accounts_api.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
async def get_account(
    self,
    access_token: str,
    account_number: str,
    *,
    include_positions: bool = False,
) -> Result[dict[str, Any], ProviderError]:
    """Fetch a specific account by account number.

    Args:
        access_token: Valid Schwab access token (Bearer token).
        account_number: Schwab account number to fetch.
        include_positions: Include position details in response.

    Returns:
        Success(dict): Single account JSON object from Schwab.
        Failure(ProviderAuthenticationError): If token is invalid/expired.
        Failure(ProviderRateLimitError): If rate limit exceeded.
        Failure(ProviderUnavailableError): If Schwab API is unreachable.
        Failure(ProviderInvalidResponseError): If response is malformed.
    """
    self._logger.debug(
        "schwab_accounts_api_get_account_started",
        account_number_masked=f"****{account_number[-4:]}"
        if len(account_number) >= 4
        else "****",
        include_positions=include_positions,
    )

    params: dict[str, str] | None = (
        {"fields": "positions"} if include_positions else None
    )

    return await self._execute_and_parse_object(
        method="GET",
        path=f"/accounts/{account_number}",
        headers=self._build_headers(access_token),
        params=params,
        operation="get_account",
    )