Client API Reference

StockTrimClient

StockTrimClient

StockTrimClient(
    api_auth_id: str | None = None,
    api_auth_signature: str | None = None,
    base_url: str | None = None,
    timeout: float = 30.0,
    max_retries: int = 5,
    logger: Logger | None = None,
    **httpx_kwargs: Any,
)

Bases: AuthenticatedClient

The pythonic StockTrim API client with automatic resilience.

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

Features: - Automatic retries on server errors (5xx) for idempotent methods only - Custom header authentication (api-auth-id, api-auth-signature) - Rich error logging and observability - Minimal configuration - just works out of the box

Simplifications vs other APIs: - No rate limiting handling - StockTrim doesn't rate limit - No automatic pagination - StockTrim API doesn't paginate - Only retries idempotent methods (GET, HEAD, OPTIONS, TRACE) on 5xx errors

Usage

Basic usage with environment variables

async with StockTrimClient() as client: from stocktrim_public_api_client.api.products import get_api_products

response = await get_api_products.asyncio_detailed(
    client=client  # Pass client directly - no .client needed!
)

With explicit credentials

async with StockTrimClient( api_auth_id="your-id", api_auth_signature="your-signature" ) as client: # All API calls through client get automatic resilience response = await some_api_method.asyncio_detailed(client=client)

Initialize the StockTrim API client with automatic resilience features.

Parameters:

Name	Type	Description	Default
api_auth_id	str | None	StockTrim API authentication ID. If None, will try to load from STOCKTRIM_API_AUTH_ID env var.	None
api_auth_signature	str | None	StockTrim API authentication signature. If None, will try to load from STOCKTRIM_API_AUTH_SIGNATURE env var.	None
base_url	str | None	Base URL for the StockTrim API. Defaults to https://api.stocktrim.com	None
timeout	float	Request timeout in seconds. Defaults to 30.0.	30.0
max_retries	int	Maximum number of retry attempts for failed requests. Defaults to 5.	5
logger	Logger | None	Logger instance for capturing client operations. If None, creates a default logger.	None
**httpx_kwargs	Any	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:

Type	Description
ValueError	If no API credentials are provided and environment variables are not set.

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 StockTrimClient() as client: ... # All API calls through client get automatic resilience ... response = await some_api_method.asyncio_detailed(client=client)

Source code in stocktrim_public_api_client/stocktrim_client.py

def __init__(
    self,
    api_auth_id: str | None = None,
    api_auth_signature: str | None = None,
    base_url: str | None = None,
    timeout: float = 30.0,
    max_retries: int = 5,
    logger: logging.Logger | None = None,
    **httpx_kwargs: Any,
):
    """
    Initialize the StockTrim API client with automatic resilience features.

    Args:
        api_auth_id: StockTrim API authentication ID. If None, will try to load from
            STOCKTRIM_API_AUTH_ID env var.
        api_auth_signature: StockTrim API authentication signature. If None, will try to
            load from STOCKTRIM_API_AUTH_SIGNATURE env var.
        base_url: Base URL for the StockTrim API. Defaults to https://api.stocktrim.com
        timeout: Request timeout in seconds. Defaults to 30.0.
        max_retries: Maximum number of retry attempts for failed requests. Defaults to 5.
        logger: Logger instance for capturing client operations. If None, creates a default 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 credentials are provided and environment variables are not set.

    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 StockTrimClient() as client:
        ...     # All API calls through client get automatic resilience
        ...     response = await some_api_method.asyncio_detailed(client=client)
    """
    load_dotenv()

    # Setup credentials
    api_auth_id = api_auth_id or os.getenv("STOCKTRIM_API_AUTH_ID")
    api_auth_signature = api_auth_signature or os.getenv(
        "STOCKTRIM_API_AUTH_SIGNATURE"
    )
    base_url = (
        base_url or os.getenv("STOCKTRIM_BASE_URL") or "https://api.stocktrim.com"
    )

    if not api_auth_id or not api_auth_signature:
        raise ValueError(
            "API credentials required (STOCKTRIM_API_AUTH_ID and "
            "STOCKTRIM_API_AUTH_SIGNATURE env vars or api_auth_id and "
            "api_auth_signature params)"
        )

    self.logger = logger or logging.getLogger(__name__)
    self.max_retries = max_retries

    # 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._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

    # Create resilient transport with all the layers
    # Note: This transport only adds the api-auth-signature header.
    # The api-auth-id header is added by AuthenticatedClient's native mechanism.
    transport, error_logging_transport = create_resilient_transport(
        api_auth_signature=api_auth_signature,
        max_retries=max_retries,
        logger=self.logger,
        **httpx_kwargs,  # Pass through http2, limits, verify, etc.
    )

    # Store reference to error logging transport for helper methods
    # Public API for helper methods to use enhanced error logging
    self.error_logging_transport = error_logging_transport

    # Initialize parent with resilient transport
    # Use AuthenticatedClient's native customization to add the api-auth-id header:
    # - token: the API auth ID value
    # - auth_header_name: "api-auth-id" (StockTrim's custom header name)
    # - prefix: "" (no prefix like "Bearer")
    super().__init__(
        base_url=base_url,
        token=api_auth_id,  # Use the auth ID as the token value
        auth_header_name="api-auth-id",  # StockTrim's custom header name
        prefix="",  # No prefix (disables "Bearer ")
        timeout=httpx.Timeout(timeout),
        httpx_args={
            "transport": transport,
            "event_hooks": event_hooks,
        },
    )

base_url property

base_url: str

Get the base URL for the API.

bill_of_materials property

bill_of_materials: BillOfMaterials

Access the BillOfMaterials helper for BOM management.

customers property

customers: Customers

Access the Customers helper for customer management.

forecasting property

forecasting: Forecasting

Access the Forecasting helper for forecast management and processing status.

inventory property

inventory: Inventory

Access the Inventory helper for inventory management.

locations property

locations: Locations

Access the Locations helper for location management.

order_plan property

order_plan: OrderPlan

Access the OrderPlan helper for forecast and demand planning operations.

products property

products: Products

Access the Products helper for product catalog operations.

purchase_orders property

purchase_orders: PurchaseOrders

Access the PurchaseOrders helper for purchase order management.

purchase_orders_v2 property

purchase_orders_v2: PurchaseOrdersV2

Access the PurchaseOrdersV2 helper for V2 purchase order operations (recommended over V1).

sales_orders property

sales_orders: SalesOrders

Access the SalesOrders helper for sales order management.

suppliers property

suppliers: Suppliers

Access the Suppliers helper for supplier management.

__aenter__ async

__aenter__() -> StockTrimClient

Enter async context manager, returning self for proper type checking.

Source code in stocktrim_public_api_client/stocktrim_client.py

async def __aenter__(self) -> "StockTrimClient":
    """Enter async context manager, returning self for proper type checking."""
    await super().__aenter__()
    return self

__aexit__ async

__aexit__(*args: Any) -> None

Exit async context manager.

Source code in stocktrim_public_api_client/stocktrim_client.py

async def __aexit__(self, *args: Any) -> None:
    """Exit async context manager."""
    await super().__aexit__(*args)

__repr__

__repr__() -> str

String representation of the client.

Source code in stocktrim_public_api_client/stocktrim_client.py

def __repr__(self) -> str:
    """String representation of the client."""
    return (
        f"StockTrimClient(base_url='{self._base_url}', "
        f"max_retries={self.max_retries})"
    )

IdempotentOnlyRetry

IdempotentOnlyRetry

IdempotentOnlyRetry(*args: Any, **kwargs: Any)

Bases: Retry

Custom Retry class that only retries idempotent methods (GET, HEAD, OPTIONS, TRACE) on server errors (5xx status codes).

StockTrim doesn't have rate limiting (429), so we only need to handle 5xx errors and we only retry idempotent methods to avoid duplicate operations.

Initialize and track the current request method.

Source code in stocktrim_public_api_client/stocktrim_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

increment

increment() -> IdempotentOnlyRetry

Return a new retry instance with the attempt count incremented.

Source code in stocktrim_public_api_client/stocktrim_client.py

def increment(self) -> "IdempotentOnlyRetry":
    """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(IdempotentOnlyRetry, super().increment())
    # Preserve the current method across retry attempts
    new_retry._current_method = self._current_method
    return new_retry

is_retryable_method

is_retryable_method(method: str) -> bool

Allow all methods to pass through the initial check.

Store the method for later use in is_retryable_status_code.

Source code in stocktrim_public_api_client/stocktrim_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

is_retryable_status_code(status_code: int) -> bool

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

For 5xx errors, only allow idempotent methods.

Source code in stocktrim_public_api_client/stocktrim_client.py

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

    For 5xx errors, 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

    # Server errors (5xx) - only retry idempotent methods
    return self._current_method in self.IDEMPOTENT_METHODS

ErrorLoggingTransport

ErrorLoggingTransport

ErrorLoggingTransport(
    wrapped_transport: AsyncHTTPTransport | None = None,
    logger: Logger | None = None,
    **kwargs: Any,
)

Bases: AsyncHTTPTransport

Transport layer that adds comprehensive logging for all HTTP requests and responses.

This transport wraps another AsyncHTTPTransport and intercepts responses to log: - DEBUG: Request details (sanitized headers), response bodies for 2xx responses - INFO: Successful 2xx responses with timing - WARNING: Null responses that may cause TypeErrors - ERROR: 4xx client errors and 5xx server errors with response details

Initialize the error logging transport.

Parameters:

Name	Type	Description	Default
wrapped_transport	AsyncHTTPTransport | None	The transport to wrap. If None, creates a new AsyncHTTPTransport.	None
logger	Logger | None	Logger instance for capturing error details. If None, creates a default logger.	None
**kwargs	Any	Additional arguments passed to AsyncHTTPTransport if wrapped_transport is None.	{}

Source code in stocktrim_public_api_client/stocktrim_client.py

def __init__(
    self,
    wrapped_transport: AsyncHTTPTransport | None = None,
    logger: logging.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 or logging.getLogger(__name__)

handle_async_request async

handle_async_request(request: Request) -> httpx.Response

Handle request and log based on response status code.

Source code in stocktrim_public_api_client/stocktrim_client.py

async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
    """Handle request and log based on response status code."""
    # Log request details at DEBUG level
    await self._log_request(request)

    # Track timing
    start_time = time.time()
    response = await self._wrapped_transport.handle_async_request(request)
    duration_ms = (time.time() - start_time) * 1000

    # Log based on status code
    if 200 <= response.status_code < 300:
        await self._log_success_response(response, request, duration_ms)
    elif 400 <= response.status_code < 500:
        await self._log_client_error(response, request, duration_ms)
    elif 500 <= response.status_code < 600:
        await self._log_server_error(response, request, duration_ms)
    else:
        # Unexpected status codes (1xx, 3xx redirects shouldn't reach here)
        self.logger.warning(
            f"{request.method} {request.url} -> {response.status_code} "
            f"({duration_ms:.0f}ms)"
        )

    return response

log_parsing_error

log_parsing_error(
    error: TypeError
    | ValueError
    | AttributeError
    | Exception,
    response: Response,
    request: Request,
) -> None

Log detailed information when an error occurs during response parsing.

This method intelligently inspects the error and response to provide actionable debugging information with fix suggestions: - For TypeErrors: Shows null fields and provides 3 fix options with doc links - For ValueErrors: Shows the error and response excerpt - For any parsing error: Logs the raw response for inspection

Parameters:

Name	Type	Description	Default
error	TypeError | ValueError | AttributeError | Exception	The exception that occurred during parsing	required
response	Response	The HTTP response that was being parsed	required
request	Request	The original HTTP request	required

Example output for TypeError with null fields

ERROR: TypeError during parsing for GET /api/V2/PurchaseOrders ERROR: TypeError: object of type 'NoneType' has no len() ERROR: Found 3 null field(s) in response: ERROR: - orderDate ERROR: - fullyReceivedDate ERROR: - supplier.supplierName ERROR: ERROR: Possible fixes: ERROR: 1. Add fields to NULLABLE_FIELDS in scripts/regenerate_client.py and regenerate ERROR: 2. Update OpenAPI spec to mark these fields as 'nullable: true' ERROR: 3. Handle null values defensively in helper methods ERROR: ERROR: See: docs/contributing/api-feedback.md#nullable-date-fields-not-marked-in-spec

Example output for ValueError

ERROR: ValueError during parsing for POST /api/Products ERROR: ValueError: invalid literal for int() with base 10: 'abc' ERROR: Response excerpt: {"id": "abc", "name": "Product 1"}

Source code in stocktrim_public_api_client/stocktrim_client.py

def log_parsing_error(
    self,
    error: TypeError | ValueError | AttributeError | Exception,
    response: httpx.Response,
    request: httpx.Request,
) -> None:
    """
    Log detailed information when an error occurs during response parsing.

    This method intelligently inspects the error and response to provide
    actionable debugging information with fix suggestions:
    - For TypeErrors: Shows null fields and provides 3 fix options with doc links
    - For ValueErrors: Shows the error and response excerpt
    - For any parsing error: Logs the raw response for inspection

    Args:
        error: The exception that occurred during parsing
        response: The HTTP response that was being parsed
        request: The original HTTP request

    Example output for TypeError with null fields:
        ERROR: TypeError during parsing for GET /api/V2/PurchaseOrders
        ERROR: TypeError: object of type 'NoneType' has no len()
        ERROR: Found 3 null field(s) in response:
        ERROR:   - orderDate
        ERROR:   - fullyReceivedDate
        ERROR:   - supplier.supplierName
        ERROR:
        ERROR: Possible fixes:
        ERROR:   1. Add fields to NULLABLE_FIELDS in scripts/regenerate_client.py and regenerate
        ERROR:   2. Update OpenAPI spec to mark these fields as 'nullable: true'
        ERROR:   3. Handle null values defensively in helper methods
        ERROR:
        ERROR: See: docs/contributing/api-feedback.md#nullable-date-fields-not-marked-in-spec

    Example output for ValueError:
        ERROR: ValueError during parsing for POST /api/Products
        ERROR: ValueError: invalid literal for int() with base 10: 'abc'
        ERROR: Response excerpt: {"id": "abc", "name": "Product 1"}
    """
    method = request.method
    url = str(request.url)
    error_type = type(error).__name__

    self.logger.error(f"{error_type} during parsing for {method} {url}")
    self.logger.error(f"{error_type}: {error}")

    # Try to parse response and provide context
    try:
        response_data = response.json()

        # For TypeErrors, check for null fields (common cause)
        if isinstance(error, TypeError):
            null_fields = _find_null_fields(response_data)

            if null_fields:
                self.logger.error(
                    f"Found {len(null_fields)} null field(s) in response:"
                )
                for field_path in null_fields[: self.MAX_NULL_FIELDS_TO_LOG]:
                    self.logger.error(f"  - {field_path}")
                if len(null_fields) > self.MAX_NULL_FIELDS_TO_LOG:
                    self.logger.error(
                        f"  ... and {len(null_fields) - self.MAX_NULL_FIELDS_TO_LOG} more null fields"
                    )

                # Provide actionable fix suggestions
                self.logger.error("")  # Blank line for readability
                self.logger.error("Possible fixes:")
                self.logger.error(
                    "  1. Add fields to NULLABLE_FIELDS in scripts/regenerate_client.py and regenerate"
                )
                self.logger.error(
                    "  2. Update OpenAPI spec to mark these fields as 'nullable: true'"
                )
                self.logger.error(
                    "  3. Handle null values defensively in helper methods"
                )
                self.logger.error("")  # Blank line for readability
                self.logger.error(
                    "See: docs/contributing/api-feedback.md#nullable-arrays-vs-optional-fields"
                )
            else:
                # TypeError but no null fields - show response excerpt
                response_str = str(response_data)
                excerpt = response_str[: self.RESPONSE_BODY_MAX_LENGTH]
                ellipsis = (
                    "..."
                    if len(response_str) > self.RESPONSE_BODY_MAX_LENGTH
                    else ""
                )
                self.logger.error(
                    f"No null fields found. Response excerpt: {excerpt}{ellipsis}"
                )
        else:
            # For other errors, show response excerpt
            response_str = str(response_data)
            excerpt = response_str[: self.RESPONSE_BODY_MAX_LENGTH]
            ellipsis = (
                "..." if len(response_str) > self.RESPONSE_BODY_MAX_LENGTH else ""
            )
            self.logger.error(f"Response excerpt: {excerpt}{ellipsis}")

    except (json.JSONDecodeError, TypeError, ValueError) as e:
        self.logger.error(
            f"Could not parse response as JSON: {type(e).__name__}: {e}"
        )
        text_excerpt = response.text[: self.RESPONSE_BODY_MAX_LENGTH]
        ellipsis = (
            "..." if len(response.text) > self.RESPONSE_BODY_MAX_LENGTH else ""
        )
        self.logger.error(f"Response text: {text_excerpt}{ellipsis}")