Skip to content

statuspro_public_api_client

statuspro_public_api_client

StatusPro Public API Client — Python client for the StatusPro API.

Classes

APIError(message, status_code, error_response=None)

Bases: Exception

Base exception for API errors.

Source code in statuspro_public_api_client/utils.py 
22
23
24
25
26
27
28
29
30
def __init__(
    self,
    message: str,
    status_code: int,
    error_response: ErrorResponse | ValidationErrorResponse | None = None,
):
    super().__init__(message)
    self.status_code = status_code
    self.error_response = error_response

AuthenticatedClient

A Client which has been authenticated for use on secured endpoints

The following are accepted as keyword arguments and will be used to construct httpx Clients internally:

``base_url``: The base URL for the API, all requests are made to a relative path to this URL

``cookies``: A dictionary of cookies to be sent with every request

``headers``: A dictionary of headers to be sent with every request

``timeout``: The maximum amount of a time a request can take. API functions will raise
httpx.TimeoutException if this is exceeded.

``verify_ssl``: Whether or not to verify the SSL certificate of the API server. This should be True in production,
but can be set to False for testing purposes.

``follow_redirects``: Whether or not to follow redirects. Default value is False.

``httpx_args``: A dictionary of additional arguments to be passed to the ``httpx.Client`` and ``httpx.AsyncClient`` constructor.
Functions
__aenter__() async

Enter a context manager for underlying httpx.AsyncClient—you cannot enter twice (see httpx docs)

Source code in statuspro_public_api_client/client.py 
260
261
262
263
async def __aenter__(self) -> "AuthenticatedClient":
    """Enter a context manager for underlying httpx.AsyncClient—you cannot enter twice (see httpx docs)"""
    await self.get_async_httpx_client().__aenter__()
    return self
__aexit__(*args, **kwargs) async

Exit a context manager for underlying httpx.AsyncClient (see httpx docs)

Source code in statuspro_public_api_client/client.py 
265
266
267
async def __aexit__(self, *args: Any, **kwargs: Any) -> None:
    """Exit a context manager for underlying httpx.AsyncClient (see httpx docs)"""
    await self.get_async_httpx_client().__aexit__(*args, **kwargs)
__enter__()

Enter a context manager for self.client—you cannot enter twice (see httpx docs)

Source code in statuspro_public_api_client/client.py 
224
225
226
227
def __enter__(self) -> "AuthenticatedClient":
    """Enter a context manager for self.client—you cannot enter twice (see httpx docs)"""
    self.get_httpx_client().__enter__()
    return self
__exit__(*args, **kwargs)

Exit a context manager for internal httpx.Client (see httpx docs)

Source code in statuspro_public_api_client/client.py 
229
230
231
def __exit__(self, *args: Any, **kwargs: Any) -> None:
    """Exit a context manager for internal httpx.Client (see httpx docs)"""
    self.get_httpx_client().__exit__(*args, **kwargs)
get_async_httpx_client()

Get the underlying httpx.AsyncClient, constructing a new one if not previously set

Source code in statuspro_public_api_client/client.py 
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
def get_async_httpx_client(self) -> httpx.AsyncClient:
    """Get the underlying httpx.AsyncClient, constructing a new one if not previously set"""
    if self._async_client is None:
        self._headers[self.auth_header_name] = (
            f"{self.prefix} {self.token}" if self.prefix else self.token
        )
        self._async_client = httpx.AsyncClient(
            base_url=self._base_url,
            cookies=self._cookies,
            headers=self._headers,
            timeout=self._timeout,
            verify=self._verify_ssl,
            follow_redirects=self._follow_redirects,
            **self._httpx_args,
        )
    return self._async_client
get_httpx_client()

Get the underlying httpx.Client, constructing a new one if not previously set

Source code in statuspro_public_api_client/client.py 
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
def get_httpx_client(self) -> httpx.Client:
    """Get the underlying httpx.Client, constructing a new one if not previously set"""
    if self._client is None:
        self._headers[self.auth_header_name] = (
            f"{self.prefix} {self.token}" if self.prefix else self.token
        )
        self._client = httpx.Client(
            base_url=self._base_url,
            cookies=self._cookies,
            headers=self._headers,
            timeout=self._timeout,
            verify=self._verify_ssl,
            follow_redirects=self._follow_redirects,
            **self._httpx_args,
        )
    return self._client
set_async_httpx_client(async_client)

Manually set the underlying httpx.AsyncClient

NOTE: This will override any other settings on the client, including cookies, headers, and timeout.

Source code in statuspro_public_api_client/client.py 
233
234
235
236
237
238
239
240
241
def set_async_httpx_client(
    self, async_client: httpx.AsyncClient
) -> "AuthenticatedClient":
    """Manually set the underlying httpx.AsyncClient

    **NOTE**: This will override any other settings on the client, including cookies, headers, and timeout.
    """
    self._async_client = async_client
    return self
set_httpx_client(client)

Manually set the underlying httpx.Client

NOTE: This will override any other settings on the client, including cookies, headers, and timeout.

Source code in statuspro_public_api_client/client.py 
199
200
201
202
203
204
205
def set_httpx_client(self, client: httpx.Client) -> "AuthenticatedClient":
    """Manually set the underlying httpx.Client

    **NOTE**: This will override any other settings on the client, including cookies, headers, and timeout.
    """
    self._client = client
    return self
with_cookies(cookies)

Get a new client matching this one with additional cookies

Source code in statuspro_public_api_client/client.py 
183
184
185
186
187
188
189
def with_cookies(self, cookies: dict[str, str]) -> "AuthenticatedClient":
    """Get a new client matching this one with additional cookies"""
    if self._client is not None:
        self._client.cookies.update(cookies)
    if self._async_client is not None:
        self._async_client.cookies.update(cookies)
    return evolve(self, cookies={**self._cookies, **cookies})
with_headers(headers)

Get a new client matching this one with additional headers

Source code in statuspro_public_api_client/client.py 
175
176
177
178
179
180
181
def with_headers(self, headers: dict[str, str]) -> "AuthenticatedClient":
    """Get a new client matching this one with additional headers"""
    if self._client is not None:
        self._client.headers.update(headers)
    if self._async_client is not None:
        self._async_client.headers.update(headers)
    return evolve(self, headers={**self._headers, **headers})
with_timeout(timeout)

Get a new client matching this one with a new timeout configuration

Source code in statuspro_public_api_client/client.py 
191
192
193
194
195
196
197
def with_timeout(self, timeout: httpx.Timeout) -> "AuthenticatedClient":
    """Get a new client matching this one with a new timeout configuration"""
    if self._client is not None:
        self._client.timeout = timeout
    if self._async_client is not None:
        self._async_client.timeout = timeout
    return evolve(self, timeout=timeout)

AuthenticationError(message, status_code, error_response=None)

Bases: APIError

Raised when authentication fails (401).

Source code in statuspro_public_api_client/utils.py 
22
23
24
25
26
27
28
29
30
def __init__(
    self,
    message: str,
    status_code: int,
    error_response: ErrorResponse | ValidationErrorResponse | None = None,
):
    super().__init__(message)
    self.status_code = status_code
    self.error_response = error_response

Client

A class for keeping track of data related to the API

The following are accepted as keyword arguments and will be used to construct httpx Clients internally:

``base_url``: The base URL for the API, all requests are made to a relative path to this URL

``cookies``: A dictionary of cookies to be sent with every request

``headers``: A dictionary of headers to be sent with every request

``timeout``: The maximum amount of a time a request can take. API functions will raise
httpx.TimeoutException if this is exceeded.

``verify_ssl``: Whether or not to verify the SSL certificate of the API server. This should be True in production,
but can be set to False for testing purposes.

``follow_redirects``: Whether or not to follow redirects. Default value is False.

``httpx_args``: A dictionary of additional arguments to be passed to the ``httpx.Client`` and ``httpx.AsyncClient`` constructor.
Functions
__aenter__() async

Enter a context manager for underlying httpx.AsyncClient—you cannot enter twice (see httpx docs)

Source code in statuspro_public_api_client/client.py 
123
124
125
126
async def __aenter__(self) -> "Client":
    """Enter a context manager for underlying httpx.AsyncClient—you cannot enter twice (see httpx docs)"""
    await self.get_async_httpx_client().__aenter__()
    return self
__aexit__(*args, **kwargs) async

Exit a context manager for underlying httpx.AsyncClient (see httpx docs)

Source code in statuspro_public_api_client/client.py 
128
129
130
async def __aexit__(self, *args: Any, **kwargs: Any) -> None:
    """Exit a context manager for underlying httpx.AsyncClient (see httpx docs)"""
    await self.get_async_httpx_client().__aexit__(*args, **kwargs)
__enter__()

Enter a context manager for self.client—you cannot enter twice (see httpx docs)

Source code in statuspro_public_api_client/client.py 
92
93
94
95
def __enter__(self) -> "Client":
    """Enter a context manager for self.client—you cannot enter twice (see httpx docs)"""
    self.get_httpx_client().__enter__()
    return self
__exit__(*args, **kwargs)

Exit a context manager for internal httpx.Client (see httpx docs)

Source code in statuspro_public_api_client/client.py 
97
98
99
def __exit__(self, *args: Any, **kwargs: Any) -> None:
    """Exit a context manager for internal httpx.Client (see httpx docs)"""
    self.get_httpx_client().__exit__(*args, **kwargs)
get_async_httpx_client()

Get the underlying httpx.AsyncClient, constructing a new one if not previously set

Source code in statuspro_public_api_client/client.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
def get_async_httpx_client(self) -> httpx.AsyncClient:
    """Get the underlying httpx.AsyncClient, constructing a new one if not previously set"""
    if self._async_client is None:
        self._async_client = httpx.AsyncClient(
            base_url=self._base_url,
            cookies=self._cookies,
            headers=self._headers,
            timeout=self._timeout,
            verify=self._verify_ssl,
            follow_redirects=self._follow_redirects,
            **self._httpx_args,
        )
    return self._async_client
get_httpx_client()

Get the underlying httpx.Client, constructing a new one if not previously set

Source code in statuspro_public_api_client/client.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
def get_httpx_client(self) -> httpx.Client:
    """Get the underlying httpx.Client, constructing a new one if not previously set"""
    if self._client is None:
        self._client = httpx.Client(
            base_url=self._base_url,
            cookies=self._cookies,
            headers=self._headers,
            timeout=self._timeout,
            verify=self._verify_ssl,
            follow_redirects=self._follow_redirects,
            **self._httpx_args,
        )
    return self._client
set_async_httpx_client(async_client)

Manually set the underlying httpx.AsyncClient

NOTE: This will override any other settings on the client, including cookies, headers, and timeout.

Source code in statuspro_public_api_client/client.py 
101
102
103
104
105
106
107
def set_async_httpx_client(self, async_client: httpx.AsyncClient) -> "Client":
    """Manually set the underlying httpx.AsyncClient

    **NOTE**: This will override any other settings on the client, including cookies, headers, and timeout.
    """
    self._async_client = async_client
    return self
set_httpx_client(client)

Manually set the underlying httpx.Client

NOTE: This will override any other settings on the client, including cookies, headers, and timeout.

Source code in statuspro_public_api_client/client.py 
70
71
72
73
74
75
76
def set_httpx_client(self, client: httpx.Client) -> "Client":
    """Manually set the underlying httpx.Client

    **NOTE**: This will override any other settings on the client, including cookies, headers, and timeout.
    """
    self._client = client
    return self
with_cookies(cookies)

Get a new client matching this one with additional cookies

Source code in statuspro_public_api_client/client.py 
54
55
56
57
58
59
60
def with_cookies(self, cookies: dict[str, str]) -> "Client":
    """Get a new client matching this one with additional cookies"""
    if self._client is not None:
        self._client.cookies.update(cookies)
    if self._async_client is not None:
        self._async_client.cookies.update(cookies)
    return evolve(self, cookies={**self._cookies, **cookies})
with_headers(headers)

Get a new client matching this one with additional headers

Source code in statuspro_public_api_client/client.py 
46
47
48
49
50
51
52
def with_headers(self, headers: dict[str, str]) -> "Client":
    """Get a new client matching this one with additional headers"""
    if self._client is not None:
        self._client.headers.update(headers)
    if self._async_client is not None:
        self._async_client.headers.update(headers)
    return evolve(self, headers={**self._headers, **headers})
with_timeout(timeout)

Get a new client matching this one with a new timeout configuration

Source code in statuspro_public_api_client/client.py 
62
63
64
65
66
67
68
def with_timeout(self, timeout: httpx.Timeout) -> "Client":
    """Get a new client matching this one with a new timeout configuration"""
    if self._client is not None:
        self._client.timeout = timeout
    if self._async_client is not None:
        self._async_client.timeout = timeout
    return evolve(self, timeout=timeout)

RateLimitError(message, status_code, error_response=None)

Bases: APIError

Raised when rate limit is exceeded (429).

Source code in statuspro_public_api_client/utils.py 
22
23
24
25
26
27
28
29
30
def __init__(
    self,
    message: str,
    status_code: int,
    error_response: ErrorResponse | ValidationErrorResponse | None = None,
):
    super().__init__(message)
    self.status_code = status_code
    self.error_response = error_response

ServerError(message, status_code, error_response=None)

Bases: APIError

Raised when a server error occurs (5xx).

Source code in statuspro_public_api_client/utils.py 
22
23
24
25
26
27
28
29
30
def __init__(
    self,
    message: str,
    status_code: int,
    error_response: ErrorResponse | ValidationErrorResponse | None = None,
):
    super().__init__(message)
    self.status_code = status_code
    self.error_response = error_response

StatusProClient(api_key=None, base_url=None, timeout=30.0, max_retries=5, max_pages=100, logger=None, **httpx_kwargs)

Bases: AuthenticatedClient

The pythonic StatusPro API client with automatic resilience and pagination.

Inherits from AuthenticatedClient and can be passed directly to generated API methods without a .client property.

Features: - Automatic retries on network errors and server errors (5xx) - Automatic rate-limit handling (parses Retry-After, falls back to exponential backoff on 429 since StatusPro doesn't emit the header) - Auto-pagination for wrapped list endpoints ({"data": [...], "meta": {...}}) - Uses 100 items per page (StatusPro's max) by default - Raw-array endpoints (/statuses, /orders/{id}/viable-statuses) are passed through - Rich logging and observability

Auto-pagination behavior: - ON by default for GET requests with no page parameter - per_page defaults to 100; caller values are respected (capped at 100) - ANY explicit page param disables auto-pagination - Disabled per-request via extensions={"auto_pagination": False} - max_pages constructor argument caps total pages collected - extensions={"max_items": N} caps total items collected

Usage

async with StatusProClient() as client: from statuspro_public_api_client.api.orders import list_orders

response = await list_orders.asyncio_detailed(client=client)

# One specific page (disables auto-pagination)
response = await list_orders.asyncio_detailed(
    client=client, page=2, per_page=25
)

Parameters:

  • api_key (str | None, default: None ) –

    StatusPro API key. If None, will try to load from STATUSPRO_API_KEY env var, .env file, or ~/.netrc file (in that order).

  • base_url (str | None, default: None ) –

    Base URL for the StatusPro API. Defaults to https://app.orderstatuspro.com/api/v1

  • timeout (float, default: 30.0 ) –

    Request timeout in seconds. Defaults to 30.0.

  • max_retries (int, default: 5 ) –

    Maximum number of retry attempts for failed requests. Defaults to 5.

  • max_pages (int, default: 100 ) –

    Maximum number of pages to collect during auto-pagination. Defaults to 100.

  • logger (Logger | None, default: None ) –

    Any object whose debug/info/warning/error methods accept (msg, *args, **kwargs) — the standard logging.Logger call convention (e.g. logging.Logger, structlog.BoundLogger). If None, creates a default stdlib logger.

  • **httpx_kwargs (Any, default: {} ) –

    Additional arguments passed to the base AsyncHTTPTransport. Common parameters include: - http2 (bool): Enable HTTP/2 support - limits (httpx.Limits): Connection pool limits - verify (bool | str | ssl.SSLContext): SSL certificate verification - cert (str | tuple): Client-side certificates - trust_env (bool): Trust environment variables for proxy configuration - event_hooks (dict): Custom event hooks (will be merged with built-in hooks)

Raises:

  • ValueError –

    If no API key is provided via api_key param, STATUSPRO_API_KEY env var, .env file, or ~/.netrc file.

Note

Transport-related parameters (http2, limits, verify, etc.) are correctly passed to the innermost AsyncHTTPTransport layer, ensuring they take effect even with the layered transport architecture.

Example

async with StatusProClient() as client: ... # All API calls through client get automatic resilience ... response = await some_api_method.asyncio_detailed(client=client)

Source code in statuspro_public_api_client/statuspro_client.py 
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
def __init__(
    self,
    api_key: str | None = None,
    base_url: str | None = None,
    timeout: float = 30.0,
    max_retries: int = 5,
    max_pages: int = 100,
    logger: Logger | None = None,
    **httpx_kwargs: Any,
):
    """
    Initialize the StatusPro API client with automatic resilience features.

    Args:
        api_key: StatusPro API key. If None, will try to load from STATUSPRO_API_KEY env var,
            .env file, or ~/.netrc file (in that order).
        base_url: Base URL for the StatusPro API. Defaults to https://app.orderstatuspro.com/api/v1
        timeout: Request timeout in seconds. Defaults to 30.0.
        max_retries: Maximum number of retry attempts for failed requests. Defaults to 5.
        max_pages: Maximum number of pages to collect during auto-pagination. Defaults to 100.
        logger: Any object whose debug/info/warning/error methods accept
            (msg, *args, **kwargs) — the standard logging.Logger call convention
            (e.g. logging.Logger, structlog.BoundLogger). If None, creates a
            default stdlib logger.
        **httpx_kwargs: Additional arguments passed to the base AsyncHTTPTransport.
            Common parameters include:
            - http2 (bool): Enable HTTP/2 support
            - limits (httpx.Limits): Connection pool limits
            - verify (bool | str | ssl.SSLContext): SSL certificate verification
            - cert (str | tuple): Client-side certificates
            - trust_env (bool): Trust environment variables for proxy configuration
            - event_hooks (dict): Custom event hooks (will be merged with built-in hooks)

    Raises:
        ValueError: If no API key is provided via api_key param, STATUSPRO_API_KEY env var,
            .env file, or ~/.netrc file.

    Note:
        Transport-related parameters (http2, limits, verify, etc.) are correctly
        passed to the innermost AsyncHTTPTransport layer, ensuring they take effect
        even with the layered transport architecture.

    Example:
        >>> async with StatusProClient() as client:
        ...     # All API calls through client get automatic resilience
        ...     response = await some_api_method.asyncio_detailed(client=client)
    """
    load_dotenv()

    # Handle backwards compatibility: accept 'token' kwarg as alias for 'api_key'
    if "token" in httpx_kwargs:
        if api_key is not None:
            raise ValueError("Cannot specify both 'api_key' and 'token' parameters")
        api_key = httpx_kwargs.pop("token")

    # Determine base_url early so we can use it for netrc lookup
    base_url = (
        base_url
        or os.getenv("STATUSPRO_BASE_URL")
        or "https://app.orderstatuspro.com/api/v1"
    )

    # Setup credentials with priority: param > env (including .env) > netrc
    api_key = (
        api_key or os.getenv("STATUSPRO_API_KEY") or self._read_from_netrc(base_url)
    )

    if not api_key:
        raise ValueError(
            "API key required via: api_key param, STATUSPRO_API_KEY env var, "
            ".env file, or ~/.netrc"
        )

    self.logger: Logger = logger or logging.getLogger(__name__)
    self.max_pages = max_pages

    # Warn if SSL verification is disabled — risk of MITM attacks
    if httpx_kwargs.get("verify") is False:
        self.logger.warning(
            "SSL certificate verification is disabled (verify=False). "
            "This exposes the connection to MITM attacks. "
            "Only use this for local development."
        )

    # Domain helper instances (lazy-loaded via properties)
    self._orders: Orders | None = None
    self._statuses: Statuses | None = None
    self._api_namespace: ApiNamespace | None = None

    # Extract client-level parameters that shouldn't go to the transport
    # Event hooks for observability - start with our defaults
    event_hooks: dict[str, list[Callable[[httpx.Response], Awaitable[None]]]] = {
        "response": [
            self._capture_pagination_metadata,
            self._log_response_metrics,
        ]
    }

    # Extract and merge user hooks
    user_hooks = httpx_kwargs.pop("event_hooks", {})
    for event, hooks in user_hooks.items():
        # Normalize to list and add to existing or create new event
        hook_list = cast(
            list[Callable[[httpx.Response], Awaitable[None]]],
            hooks if isinstance(hooks, list) else [hooks],
        )
        if event in event_hooks:
            event_hooks[event].extend(hook_list)
        else:
            event_hooks[event] = hook_list

    # Check if user wants to override the transport entirely
    custom_transport = httpx_kwargs.pop("transport", None) or httpx_kwargs.pop(
        "async_transport", None
    )

    if custom_transport:
        # User provided a custom transport, use it as-is
        transport = custom_transport
    else:
        # Separate transport-specific kwargs from client-specific kwargs
        # Client-specific params that should NOT go to the transport
        client_only_params = ["headers", "cookies", "params", "auth"]
        client_kwargs = {
            k: httpx_kwargs.pop(k)
            for k in list(httpx_kwargs.keys())
            if k in client_only_params
        }

        # Create resilient transport with remaining transport-specific httpx_kwargs
        # These will be passed to the base AsyncHTTPTransport (http2, limits, verify, etc.)
        transport = ResilientAsyncTransport(
            max_retries=max_retries,
            max_pages=max_pages,
            logger=self.logger,
            **httpx_kwargs,  # Pass through http2, limits, verify, cert, trust_env, etc.
        )

        # Put client-specific params back into httpx_kwargs for the parent class
        httpx_kwargs.update(client_kwargs)

    # Initialize the parent AuthenticatedClient
    super().__init__(
        base_url=base_url,
        token=api_key,
        timeout=httpx.Timeout(timeout),
        httpx_args={
            "transport": transport,
            "event_hooks": event_hooks,
            **httpx_kwargs,  # Include any remaining client-level kwargs
        },
    )
Attributes
api property

Thin CRUD wrappers for all API resources. Returns raw attrs models.

orders property

Access order operations (list, lookup, get, update status, etc.).

statuses property

Access status catalog operations.

Functions

ValidationError(message, status_code, error_response=None)

Bases: APIError

Raised when request validation fails (422).

The validation_errors attribute is a dict[str, list[str]] mapping field name to the list of human-readable error messages that StatusPro returned for that field.

Source code in statuspro_public_api_client/utils.py 
45
46
47
48
49
50
51
52
53
def __init__(
    self,
    message: str,
    status_code: int,
    error_response: ValidationErrorResponse | None = None,
):
    super().__init__(message, status_code, error_response)
    errors = getattr(error_response, "errors", None)
    self.validation_errors: dict[str, list[str]] = unwrap_unset(errors, {}) or {}

Functions

get_error_message(response)

Extract message from an error response, if present.

Source code in statuspro_public_api_client/utils.py 
243
244
245
246
247
248
249
250
251
252
253
def get_error_message[T](response: Response[T]) -> str | None:
    """Extract ``message`` from an error response, if present."""
    if response.parsed is None:
        return None

    parsed = response.parsed
    if not isinstance(parsed, ErrorResponse | ValidationErrorResponse):
        return None

    message_raw = getattr(parsed, "message", None)
    return message_raw if isinstance(message_raw, str) and message_raw else None

handle_response(response, *, on_success=None, on_error=None, raise_on_error=False)

Call on_success with parsed data for 2xx, on_error with an APIError otherwise.

Source code in statuspro_public_api_client/utils.py 
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
def handle_response[T](
    response: Response[T],
    *,
    on_success: Callable[[T], Any] | None = None,
    on_error: Callable[[APIError], Any] | None = None,
    raise_on_error: bool = False,
) -> Any:
    """Call on_success with parsed data for 2xx, on_error with an APIError otherwise."""
    try:
        data = unwrap(response, raise_on_error=True)
        if on_success:
            return on_success(data)
        return data
    except APIError as e:
        if raise_on_error:
            raise
        if on_error:
            return on_error(e)
        return None

is_error(response)

True if the response has a 4xx or 5xx status code.

Source code in statuspro_public_api_client/utils.py 
199
200
201
def is_error(response: Response[Any]) -> bool:
    """True if the response has a 4xx or 5xx status code."""
    return response.status_code >= 400

is_success(response)

True if the response has a 2xx status code.

Source code in statuspro_public_api_client/utils.py 
194
195
196
def is_success(response: Response[Any]) -> bool:
    """True if the response has a 2xx status code."""
    return 200 <= response.status_code < 300

unwrap(response, *, raise_on_error=True) 

unwrap(
    response: Response[T], *, raise_on_error: bool = True
) -> T

unwrap(
    response: Response[T], *, raise_on_error: bool = False
) -> T | None

Unwrap a Response and return parsed data, raising typed errors on failure.

Parameters:

  • response (Response[T]) –

    Response object from a generated API call.

  • raise_on_error (bool, default: True ) –

    If True (default), raise an APIError subclass for non-2xx responses; if False, return None.

Returns:

  • T | None –

    The parsed response body.

Raises:

Source code in statuspro_public_api_client/utils.py 
 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
def unwrap[T](response: Response[T], *, raise_on_error: bool = True) -> T | None:
    """Unwrap a Response and return parsed data, raising typed errors on failure.

    Args:
        response: Response object from a generated API call.
        raise_on_error: If True (default), raise an APIError subclass for non-2xx
            responses; if False, return None.

    Returns:
        The parsed response body.

    Raises:
        AuthenticationError: 401
        ValidationError: 422
        RateLimitError: 429
        ServerError: 5xx
        APIError: other non-2xx statuses
    """
    if response.parsed is None:
        if raise_on_error:
            raise APIError(
                f"No parsed response data for status {response.status_code}",
                response.status_code,
            )
        return None

    parsed = response.parsed
    if isinstance(parsed, ErrorResponse | ValidationErrorResponse):
        if not raise_on_error:
            return None

        message_raw = getattr(parsed, "message", None)
        error_message = (
            message_raw
            if isinstance(message_raw, str) and message_raw
            else "No error message provided"
        )

        status = response.status_code
        if status == HTTPStatus.UNAUTHORIZED:
            raise AuthenticationError(error_message, status, parsed)
        if status == HTTPStatus.UNPROCESSABLE_ENTITY:
            detailed = parsed if isinstance(parsed, ValidationErrorResponse) else None
            raise ValidationError(error_message, status, detailed)
        if status == HTTPStatus.TOO_MANY_REQUESTS:
            raise RateLimitError(error_message, status, parsed)
        if 500 <= status < 600:
            raise ServerError(error_message, status, parsed)
        raise APIError(error_message, status, parsed)

    return response.parsed

unwrap_as(response, expected_type, *, raise_on_error=True) 

unwrap_as(
    response: Response[T],
    expected_type: type[ExpectedT],
    *,
    raise_on_error: bool = True
) -> ExpectedT

unwrap_as(
    response: Response[T],
    expected_type: type[ExpectedT],
    *,
    raise_on_error: bool = False
) -> ExpectedT | None

Unwrap and assert the parsed body is of the expected type.

Source code in statuspro_public_api_client/utils.py 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
def unwrap_as[T, ExpectedT](
    response: Response[T],
    expected_type: type[ExpectedT],
    *,
    raise_on_error: bool = True,
) -> ExpectedT | None:
    """Unwrap and assert the parsed body is of the expected type."""
    result = unwrap(response, raise_on_error=raise_on_error)
    if result is None:
        if raise_on_error:
            raise TypeError(
                f"Expected {expected_type.__name__}, got None from response"
            )
        return None
    if not isinstance(result, expected_type):
        raise TypeError(
            f"Expected {expected_type.__name__}, got {type(result).__name__}"
        )
    return result

unwrap_data(response, *, raise_on_error=True, default=None) 

unwrap_data(
    response: Response[T],
    *,
    raise_on_error: bool = True,
    default: None = None
) -> Any

unwrap_data(
    response: Response[T],
    *,
    raise_on_error: bool = False,
    default: None = None
) -> Any | None

unwrap_data(
    response: Response[T],
    *,
    raise_on_error: bool = False,
    default: list[DataT]
) -> Any

Unwrap a list response and extract the data array.

For StatusPro's wrapped list endpoints like GET /orders which return {"data": [...], "meta": {...}}. For raw-array endpoints like GET /statuses, use unwrap() directly — the parsed response is the list itself.

Source code in statuspro_public_api_client/utils.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
def unwrap_data[T, DataT](
    response: Response[T],
    *,
    raise_on_error: bool = True,
    default: list[DataT] | None = None,
) -> Any | None:
    """Unwrap a list response and extract the ``data`` array.

    For StatusPro's wrapped list endpoints like ``GET /orders`` which return
    ``{"data": [...], "meta": {...}}``. For raw-array endpoints like
    ``GET /statuses``, use ``unwrap()`` directly — the parsed response is the
    list itself.
    """
    try:
        parsed = unwrap(response, raise_on_error=raise_on_error)
    except APIError:
        if raise_on_error:
            raise
        return default

    if parsed is None:
        return default

    # If the parsed result is already a list, return it directly
    if isinstance(parsed, list):
        return parsed

    data = getattr(parsed, "data", None)
    if isinstance(data, Unset):
        return default if default is not None else []
    if data is not None:
        return data

    if default is not None:
        return default
    return [parsed]