Skip to content

infrastructure.providers.schwab.api.transactions_api

src.infrastructure.providers.schwab.api.transactions_api

Schwab Transactions API client.

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

Endpoints

GET /trader/v1/accounts/{accountNumber}/transactions - Get all transactions GET /trader/v1/accounts/{accountNumber}/transactions/{transactionId} - Get specific transaction

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

Attributes

Classes

SchwabTransactionsAPI

Bases: BaseProviderAPIClient

HTTP client for Schwab Trader API transactions 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 = SchwabTransactionsAPI( ... base_url="https://api.schwabapi.com/trader/v1", ... timeout=30.0, ... ) result = await api.get_transactions( ... access_token="...", ... account_number="12345678", ... start_date=date(2024, 1, 1), ... )

Source code in src/infrastructure/providers/schwab/api/transactions_api.py 
 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
152
153
154
155
156
157
158
159
160
161
162
163
164
165
class SchwabTransactionsAPI(BaseProviderAPIClient):
    """HTTP client for Schwab Trader API transactions 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 = SchwabTransactionsAPI(
        ...     base_url="https://api.schwabapi.com/trader/v1",
        ...     timeout=30.0,
        ... )
        >>> result = await api.get_transactions(
        ...     access_token="...",
        ...     account_number="12345678",
        ...     start_date=date(2024, 1, 1),
        ... )
    """

    def __init__(
        self,
        *,
        base_url: str,
        timeout: float = PROVIDER_TIMEOUT_DEFAULT,
    ) -> None:
        """Initialize Schwab Transactions 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_transactions(
        self,
        access_token: str,
        account_number: str,
        *,
        start_date: date | None = None,
        end_date: date | None = None,
        transaction_type: str | None = None,
    ) -> Result[list[dict[str, Any]], ProviderError]:
        """Fetch transactions for a specific account.

        Args:
            access_token: Valid Schwab access token (Bearer token).
            account_number: Schwab account number.
            start_date: Beginning of date range (ISO format YYYY-MM-DD).
            end_date: End of date range (ISO format YYYY-MM-DD).
            transaction_type: Filter by transaction type (e.g., "TRADE", "DIVIDEND").

        Returns:
            Success(list[dict]): List of transaction 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.
        """
        masked_account = (
            f"****{account_number[-4:]}" if len(account_number) >= 4 else "****"
        )

        self._logger.debug(
            "schwab_transactions_api_get_transactions_started",
            account_number_masked=masked_account,
            start_date=str(start_date) if start_date else None,
            end_date=str(end_date) if end_date else None,
            transaction_type=transaction_type,
        )

        # Build query params
        params: dict[str, str] = {}
        if start_date:
            params["startDate"] = start_date.isoformat()
        if end_date:
            params["endDate"] = end_date.isoformat()
        if transaction_type:
            params["types"] = transaction_type

        return await self._execute_and_parse_list(
            method="GET",
            path=f"/accounts/{account_number}/transactions",
            headers=self._build_headers(access_token),
            params=params if params else None,
            operation="get_transactions",
        )

    async def get_transaction(
        self,
        access_token: str,
        account_number: str,
        transaction_id: str,
    ) -> Result[dict[str, Any], ProviderError]:
        """Fetch a specific transaction by ID.

        Args:
            access_token: Valid Schwab access token (Bearer token).
            account_number: Schwab account number.
            transaction_id: Schwab transaction ID.

        Returns:
            Success(dict): Single transaction 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.
        """
        masked_account = (
            f"****{account_number[-4:]}" if len(account_number) >= 4 else "****"
        )

        self._logger.debug(
            "schwab_transactions_api_get_transaction_started",
            account_number_masked=masked_account,
            transaction_id=transaction_id,
        )

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

    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/transactions_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 Transactions 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_transactions async 
get_transactions(
    access_token: str,
    account_number: str,
    *,
    start_date: date | None = None,
    end_date: date | None = None,
    transaction_type: str | None = None
) -> Result[list[dict[str, Any]], ProviderError]

Fetch transactions for a specific account.

Parameters:

Name Type Description Default
access_token str

Valid Schwab access token (Bearer token).

 required
account_number str

Schwab account number.

 required
start_date date | None

Beginning of date range (ISO format YYYY-MM-DD).

 None
end_date date | None

End of date range (ISO format YYYY-MM-DD).

 None
transaction_type str | None

Filter by transaction type (e.g., "TRADE", "DIVIDEND").

 None

Returns:

Name Type Description
Success list[dict]

List of transaction 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/transactions_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
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
async def get_transactions(
    self,
    access_token: str,
    account_number: str,
    *,
    start_date: date | None = None,
    end_date: date | None = None,
    transaction_type: str | None = None,
) -> Result[list[dict[str, Any]], ProviderError]:
    """Fetch transactions for a specific account.

    Args:
        access_token: Valid Schwab access token (Bearer token).
        account_number: Schwab account number.
        start_date: Beginning of date range (ISO format YYYY-MM-DD).
        end_date: End of date range (ISO format YYYY-MM-DD).
        transaction_type: Filter by transaction type (e.g., "TRADE", "DIVIDEND").

    Returns:
        Success(list[dict]): List of transaction 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.
    """
    masked_account = (
        f"****{account_number[-4:]}" if len(account_number) >= 4 else "****"
    )

    self._logger.debug(
        "schwab_transactions_api_get_transactions_started",
        account_number_masked=masked_account,
        start_date=str(start_date) if start_date else None,
        end_date=str(end_date) if end_date else None,
        transaction_type=transaction_type,
    )

    # Build query params
    params: dict[str, str] = {}
    if start_date:
        params["startDate"] = start_date.isoformat()
    if end_date:
        params["endDate"] = end_date.isoformat()
    if transaction_type:
        params["types"] = transaction_type

    return await self._execute_and_parse_list(
        method="GET",
        path=f"/accounts/{account_number}/transactions",
        headers=self._build_headers(access_token),
        params=params if params else None,
        operation="get_transactions",
    )
get_transaction async 
get_transaction(
    access_token: str,
    account_number: str,
    transaction_id: str,
) -> Result[dict[str, Any], ProviderError]

Fetch a specific transaction by ID.

Parameters:

Name Type Description Default
access_token str

Valid Schwab access token (Bearer token).

 required
account_number str

Schwab account number.

 required
transaction_id str

Schwab transaction ID.

 required

Returns:

Name Type Description
Success dict

Single transaction 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/transactions_api.py 
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
async def get_transaction(
    self,
    access_token: str,
    account_number: str,
    transaction_id: str,
) -> Result[dict[str, Any], ProviderError]:
    """Fetch a specific transaction by ID.

    Args:
        access_token: Valid Schwab access token (Bearer token).
        account_number: Schwab account number.
        transaction_id: Schwab transaction ID.

    Returns:
        Success(dict): Single transaction 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.
    """
    masked_account = (
        f"****{account_number[-4:]}" if len(account_number) >= 4 else "****"
    )

    self._logger.debug(
        "schwab_transactions_api_get_transaction_started",
        account_number_masked=masked_account,
        transaction_id=transaction_id,
    )

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