statuspro_public_api_client.statuspro_client¶

`statuspro_public_api_client.statuspro_client` ¶

StatusProClient - The pythonic StatusPro API client with automatic resilience.

This client uses httpx's native transport layer to provide automatic retries, rate limiting, error handling, and pagination for all API calls without any decorators or wrapper methods needed.

Attributes¶

Classes¶

`ErrorLoggingTransport(wrapped_transport=None, logger=None, **kwargs)` ¶

Bases: AsyncHTTPTransport

Transport layer that adds detailed error logging for 4xx client errors.

This transport wraps another AsyncHTTPTransport and intercepts responses to log detailed error information using the generated error models.

Parameters:

wrapped_transport (AsyncHTTPTransport | None, default: None ) –

The transport to wrap. If None, creates a new AsyncHTTPTransport.
logger (Logger | None, default: None ) –

Logger instance for capturing error details. If None, creates a default logger.
**kwargs (Any, default: {} ) –

Additional arguments passed to AsyncHTTPTransport if wrapped_transport is None.

Source code in statuspro_public_api_client/statuspro_client.py

def __init__(
    self,
    wrapped_transport: AsyncHTTPTransport | None = None,
    logger: Logger | None = None,
    **kwargs: Any,
):
    """
    Initialize the error logging transport.

    Args:
        wrapped_transport: The transport to wrap. If None, creates a new AsyncHTTPTransport.
        logger: Logger instance for capturing error details. If None, creates a default logger.
        **kwargs: Additional arguments passed to AsyncHTTPTransport if wrapped_transport is None.
    """
    super().__init__()
    if wrapped_transport is None:
        wrapped_transport = AsyncHTTPTransport(**kwargs)
    self._wrapped_transport = wrapped_transport
    self.logger: Logger = logger or logging.getLogger(__name__)

Functions¶

`handle_async_request(request)` `async` ¶

Handle request and log detailed error information for 4xx responses.

Source code in statuspro_public_api_client/statuspro_client.py

async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
    """Handle request and log detailed error information for 4xx responses."""
    response = await self._wrapped_transport.handle_async_request(request)

    # Log detailed information for 400-level client errors
    if 400 <= response.status_code < 500:
        await self._log_client_error(response, request)

    return response

`PaginationTransport(wrapped_transport=None, max_pages=100, logger=None, **kwargs)` ¶

Bases: AsyncHTTPTransport

Transport layer that adds automatic pagination for GET requests.

Auto-pagination behavior (for StatusPro's page/per_page scheme): - ON by default for GET requests with NO page parameter in URL - Uses 100 items per page (StatusPro's max) when no per_page is specified - If caller specifies per_page, that value is respected - ANY explicit page parameter disables auto-pagination - Disabled when request has extensions={"auto_pagination": False} - Only applies to GET requests - Only applies to wrapped list responses ({"data": [...], "meta": {...}}); raw-array responses like /statuses are passed through unchanged.

Controlling pagination limits: - max_pages (constructor): Maximum number of pages to fetch - max_items (extension): Maximum total items to collect, e.g., extensions={"max_items": 200}

Parameters:

wrapped_transport (AsyncHTTPTransport | None, default: None ) –

The transport to wrap. If None, creates a new AsyncHTTPTransport.
max_pages (int, default: 100 ) –

Maximum number of pages to collect during auto-pagination. Defaults to 100.
logger (Logger | None, default: None ) –

Logger instance for capturing pagination operations. If None, creates a default logger.
**kwargs (Any, default: {} ) –

Additional arguments passed to AsyncHTTPTransport if wrapped_transport is None.

Source code in statuspro_public_api_client/statuspro_client.py

def __init__(
    self,
    wrapped_transport: AsyncHTTPTransport | None = None,
    max_pages: int = 100,
    logger: Logger | None = None,
    **kwargs: Any,
):
    """
    Initialize the pagination transport.

    Args:
        wrapped_transport: The transport to wrap. If None, creates a new AsyncHTTPTransport.
        max_pages: Maximum number of pages to collect during auto-pagination. Defaults to 100.
        logger: Logger instance for capturing pagination operations. If None, creates a default logger.
        **kwargs: Additional arguments passed to AsyncHTTPTransport if wrapped_transport is None.
    """
    # If no wrapped transport provided, create a base one
    if wrapped_transport is None:
        wrapped_transport = AsyncHTTPTransport(**kwargs)
        super().__init__()
    else:
        super().__init__()

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

Functions¶

`handle_async_request(request)` `async` ¶

Handle request with automatic pagination for GET requests.

Auto-pagination is ON by default for GET requests. It is disabled when: - extensions={"auto_pagination": False} is set, OR - ANY explicit page parameter is in the URL (e.g., ?page=1 or ?page=2)

To get auto-pagination, simply don't pass a page parameter. The transport will automatically use 100 items per page (StatusPro's max) unless you specify a limit, in which case your limit will be respected.

Source code in statuspro_public_api_client/statuspro_client.py

async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
    """Handle request with automatic pagination for GET requests.

    Auto-pagination is ON by default for GET requests. It is disabled when:
    - `extensions={"auto_pagination": False}` is set, OR
    - ANY explicit `page` parameter is in the URL (e.g., `?page=1` or `?page=2`)

    To get auto-pagination, simply don't pass a page parameter. The transport
    will automatically use 100 items per page (StatusPro's max) unless you specify
    a limit, in which case your limit will be respected.
    """
    # Check if auto-pagination is explicitly disabled via request extensions
    auto_pagination = request.extensions.get("auto_pagination", True)

    # ANY page param in URL disables auto-pagination - caller wants specific page
    has_explicit_page = "page" in request.url.params

    # Only paginate GET requests when auto_pagination is enabled and no explicit page
    should_paginate = (
        request.method == "GET" and auto_pagination and not has_explicit_page
    )

    if should_paginate:
        return await self._handle_paginated_request(request)
    else:
        # For non-paginated requests, just pass through to wrapped transport
        return await self._wrapped_transport.handle_async_request(request)

`RateLimitAwareRetry(*args, **kwargs)` ¶

Bases: Retry

Custom Retry class that allows non-idempotent methods (POST, PATCH) to be retried ONLY when receiving a 429 (Too Many Requests) status code.

For all other retryable status codes (502, 503, 504), only idempotent methods (HEAD, GET, PUT, DELETE, OPTIONS, TRACE) will be retried.

This ensures we don't accidentally retry non-idempotent operations after server errors, but we DO retry them when we're being rate-limited.

Source code in statuspro_public_api_client/statuspro_client.py

def __init__(self, *args: Any, **kwargs: Any):
    """Initialize and track the current request method."""
    super().__init__(*args, **kwargs)
    self._current_method: str | None = None

Functions¶

`increment()` ¶

Return a new retry instance with the attempt count incremented.

Source code in statuspro_public_api_client/statuspro_client.py

def increment(self) -> "RateLimitAwareRetry":
    """Return a new retry instance with the attempt count incremented."""
    # Call parent's increment which creates a new instance of our class
    new_retry = cast(RateLimitAwareRetry, super().increment())
    # Preserve the current method across retry attempts
    new_retry._current_method = self._current_method
    return new_retry

`is_retryable_method(method)` ¶

Allow all methods to pass through the initial check.

Store the method for later use in is_retryable_status_code.

Source code in statuspro_public_api_client/statuspro_client.py

def is_retryable_method(self, method: str) -> bool:
    """
    Allow all methods to pass through the initial check.

    Store the method for later use in is_retryable_status_code.
    """
    self._current_method = method.upper()
    # Accept all methods - we'll filter in is_retryable_status_code
    return self._current_method in self.allowed_methods

`is_retryable_status_code(status_code)` ¶

Check if a status code is retryable for the current method.

For 429 (rate limiting), allow all methods. For other errors (502, 503, 504), only allow idempotent methods.

Source code in statuspro_public_api_client/statuspro_client.py

def is_retryable_status_code(self, status_code: int) -> bool:
    """
    Check if a status code is retryable for the current method.

    For 429 (rate limiting), allow all methods.
    For other errors (502, 503, 504), only allow idempotent methods.
    """
    # First check if the status code is in the allowed list at all
    if status_code not in self.status_forcelist:
        return False

    # If we don't know the method, fall back to default behavior
    if self._current_method is None:
        return True

    # Rate limiting (429) - retry all methods
    if status_code == HTTPStatus.TOO_MANY_REQUESTS:
        return True

    # Other retryable errors - only retry idempotent methods
    return self._current_method in self.IDEMPOTENT_METHODS

`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

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¶

`ResilientAsyncTransport(max_retries=5, max_pages=100, logger=None, **kwargs)` ¶

Factory function that creates a chained transport with error logging, pagination, and retry capabilities.

This function chains multiple transport layers: 1. AsyncHTTPTransport (base HTTP transport) 2. ErrorLoggingTransport (logs detailed 4xx errors) 3. PaginationTransport (auto-collects paginated responses) 4. RetryTransport (handles retries with Retry-After header support)

Parameters:

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 ) –

Logger instance for capturing operations. If None, creates a default logger.
**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

Returns:

RetryTransport –

A RetryTransport instance wrapping all the layered transports.

Note

When using a custom transport, parameters like http2, limits, and verify must be passed to this factory function (which passes them to the base AsyncHTTPTransport), not to the httpx.Client/AsyncClient constructor.

Example

transport = ResilientAsyncTransport(max_retries=3, max_pages=50)
async with httpx.AsyncClient(transport=transport) as client:
    response = await client.get("https://api.example.com/items")

Source code in statuspro_public_api_client/statuspro_client.py

def ResilientAsyncTransport(
    max_retries: int = 5,
    max_pages: int = 100,
    logger: Logger | None = None,
    **kwargs: Any,
) -> RetryTransport:
    """
    Factory function that creates a chained transport with error logging,
    pagination, and retry capabilities.

    This function chains multiple transport layers:
    1. AsyncHTTPTransport (base HTTP transport)
    2. ErrorLoggingTransport (logs detailed 4xx errors)
    3. PaginationTransport (auto-collects paginated responses)
    4. RetryTransport (handles retries with Retry-After header support)

    Args:
        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: Logger instance for capturing operations. If None, creates a default logger.
        **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

    Returns:
        A RetryTransport instance wrapping all the layered transports.

    Note:
        When using a custom transport, parameters like http2, limits, and verify
        must be passed to this factory function (which passes them to the base
        AsyncHTTPTransport), not to the httpx.Client/AsyncClient constructor.

    Example:
        ```python
        transport = ResilientAsyncTransport(max_retries=3, max_pages=50)
        async with httpx.AsyncClient(transport=transport) as client:
            response = await client.get("https://api.example.com/items")
        ```
    """
    resolved_logger: Logger = (
        logger if logger is not None else logging.getLogger(__name__)
    )

    # Build the transport chain from inside out:
    # 1. Base AsyncHTTPTransport
    base_transport = AsyncHTTPTransport(**kwargs)

    # 2. Wrap with error logging
    error_logging_transport = ErrorLoggingTransport(
        wrapped_transport=base_transport,
        logger=resolved_logger,
    )

    # 3. Wrap with pagination
    pagination_transport = PaginationTransport(
        wrapped_transport=error_logging_transport,
        max_pages=max_pages,
        logger=resolved_logger,
    )

    # Finally wrap with retry logic (outermost layer)
    # Use RateLimitAwareRetry which:
    # - Retries ALL methods (including POST/PATCH) for 429 rate limiting
    # - Retries ONLY idempotent methods for server errors (502, 503, 504)
    retry = RateLimitAwareRetry(
        total=max_retries,
        backoff_factor=1.0,  # Exponential backoff: 1, 2, 4, 8, 16 seconds
        respect_retry_after_header=True,  # Honor server's Retry-After header
        status_forcelist=[
            429,
            502,
            503,
            504,
        ],  # Status codes that should trigger retries
        allowed_methods=[
            "HEAD",
            "GET",
            "PUT",
            "DELETE",
            "OPTIONS",
            "TRACE",
            "POST",
            "PATCH",
        ],
    )
    retry_transport = RetryTransport(
        transport=pagination_transport,
        retry=retry,
    )

    return retry_transport

statuspro_public_api_client.statuspro_client¶