katana_public_api_client.utils

katana_public_api_client.utils

Utility functions for working with Katana API responses.

This module provides convenient helpers for unwrapping API responses, handling errors, extracting data, and formatting display values.

Classes

APIError(message, status_code, error_response=None)

Bases: Exception

Base exception for API errors.

Parameters:

• message (str) –

  Human-readable error message

• status_code (int) –

  HTTP status code

• error_response (ErrorResponse | DetailedErrorResponse | None, default: None ) –

  The error response object from the API

Source code in katana_public_api_client/utils.py

def __init__(
    self,
    message: str,
    status_code: int,
    error_response: ErrorResponse | DetailedErrorResponse | None = None,
):
    """Initialize API error.

    Args:
        message: Human-readable error message
        status_code: HTTP status code
        error_response: The error response object from the API
    """
    super().__init__(message)
    self.status_code = status_code
    self.error_response = error_response

Functions

AuthenticationError(message, status_code, error_response=None)

Bases: APIError

Raised when authentication fails (401).

Source code in katana_public_api_client/utils.py

def __init__(
    self,
    message: str,
    status_code: int,
    error_response: ErrorResponse | DetailedErrorResponse | None = None,
):
    """Initialize API error.

    Args:
        message: Human-readable error message
        status_code: HTTP status code
        error_response: The error response object from the API
    """
    super().__init__(message)
    self.status_code = status_code
    self.error_response = error_response

RateLimitError(message, status_code, error_response=None)

Bases: APIError

Raised when rate limit is exceeded (429).

Source code in katana_public_api_client/utils.py

def __init__(
    self,
    message: str,
    status_code: int,
    error_response: ErrorResponse | DetailedErrorResponse | None = None,
):
    """Initialize API error.

    Args:
        message: Human-readable error message
        status_code: HTTP status code
        error_response: The error response object from the API
    """
    super().__init__(message)
    self.status_code = status_code
    self.error_response = error_response

ServerError(message, status_code, error_response=None)

Bases: APIError

Raised when server error occurs (5xx).

Source code in katana_public_api_client/utils.py

def __init__(
    self,
    message: str,
    status_code: int,
    error_response: ErrorResponse | DetailedErrorResponse | None = None,
):
    """Initialize API error.

    Args:
        message: Human-readable error message
        status_code: HTTP status code
        error_response: The error response object from the API
    """
    super().__init__(message)
    self.status_code = status_code
    self.error_response = error_response

ValidationError(message, status_code, error_response=None)

Bases: APIError

Raised when request validation fails (422).

Parameters:

• message (str) –

  Human-readable error message

• status_code (int) –

  HTTP status code (should be 422)

• error_response (DetailedErrorResponse | None, default: None ) –

  The detailed error response with validation details

Source code in katana_public_api_client/utils.py

def __init__(
    self,
    message: str,
    status_code: int,
    error_response: DetailedErrorResponse | None = None,
):
    """Initialize validation error.

    Args:
        message: Human-readable error message
        status_code: HTTP status code (should be 422)
        error_response: The detailed error response with validation details
    """
    super().__init__(message, status_code, error_response)
    self.validation_errors = unwrap_unset(
        getattr(error_response, "details", None), []
    )

Functions

__str__()

Format validation error with Ajv-keyword-specific details.

Each detail's typed subtype is dispatched into a keyword-specific renderer that consults info.* for keyword metadata (limit, pattern, allowed values, etc.). Unknown keywords route to GenericValidationError and render through a fallback that surfaces path/code/message plus any info captured in additional_properties.

Source code in katana_public_api_client/utils.py

def __str__(self) -> str:
    """Format validation error with Ajv-keyword-specific details.

    Each detail's typed subtype is dispatched into a keyword-specific
    renderer that consults ``info.*`` for keyword metadata (limit,
    pattern, allowed values, etc.). Unknown keywords route to
    ``GenericValidationError`` and render through a fallback that
    surfaces ``path``/``code``/``message`` plus any ``info`` captured
    in ``additional_properties``.
    """
    msg = super().__str__()

    if self.validation_errors:
        error_details = []
        for detail in self.validation_errors:
            error_details.append(_format_ajv_detail(detail))
        if error_details:
            msg += "\n" + "\n".join(error_details)

    return msg

Functions

get_error_message(response)

Extract error message from an error response.

Parameters:

• response (Response[T]) –

  The Response object (typically an error response)

Returns:

• str | None –

  Error message string, or None if no error message found

Example

response = await some_api_call.asyncio_detailed(client=client)
if is_error(response):
    error_msg = get_error_message(response)
    print(f"API Error: {error_msg}")

Source code in katana_public_api_client/utils.py

def get_error_message[T](response: Response[T]) -> str | None:
    """Extract error message from an error response.

    Args:
        response: The Response object (typically an error response)

    Returns:
        Error message string, or None if no error message found

    Example:
        ```python
        response = await some_api_call.asyncio_detailed(client=client)
        if is_error(response):
            error_msg = get_error_message(response)
            print(f"API Error: {error_msg}")
        ```
    """
    if response.parsed is None:
        return None

    # Assign to local for type narrowing
    parsed = response.parsed
    if not isinstance(parsed, ErrorResponse | DetailedErrorResponse):
        return None

    error_message = parsed.message if not isinstance(parsed.message, Unset) else None

    # Check nested error format
    nested = parsed.additional_properties
    if isinstance(nested, dict) and "error" in nested:
        nested_error = nested["error"]
        if isinstance(nested_error, dict):
            error_message = str(nested_error.get("message", error_message))

    return error_message

get_variant_display_name(variant)

Build the full variant display name matching Katana UI format.

Format: "{Product/Material Name} / {Config Value 1} / {Config Value 2} / ..."

Example: "Premium 140 / Glossy Black / Large / Type A"

Takes a VariantResponse (the discriminated-union variant schema with typed product_or_material: Material | Product | Unset). Variants returned without ?extend=product_or_material have product_or_material UNSET, in which case the display name falls back to empty string.

Parameters:

• variant (VariantResponse) –

  VariantResponse fetched with ?extend=product_or_material.

Returns:

• str –

  Formatted variant name with config values, or empty string if no name available.

Example

from katana_public_api_client import KatanaClient
from katana_public_api_client.api.variant import get_all_variants
from katana_public_api_client.models.get_all_variants_extend_item import (
    GetAllVariantsExtendItem,
)
from katana_public_api_client.utils import (
    get_variant_display_name,
    unwrap_data,
)

async with KatanaClient() as client:
    response = await get_all_variants.asyncio_detailed(
        client=client,
        extend=[GetAllVariantsExtendItem.PRODUCT_OR_MATERIAL],
    )
    for variant in unwrap_data(response):
        print(get_variant_display_name(variant))
        # e.g. "Premium 140 / Glossy Black / Large / Type A"

Source code in katana_public_api_client/utils.py

def get_variant_display_name(variant: "VariantResponse") -> str:
    """Build the full variant display name matching Katana UI format.

    Format: "{Product/Material Name} / {Config Value 1} / {Config Value 2} / ..."

    Example: "Premium 140 / Glossy Black / Large / Type A"

    Takes a `VariantResponse` (the discriminated-union variant schema with
    typed `product_or_material: Material | Product | Unset`). Variants returned
    *without* `?extend=product_or_material` have `product_or_material` UNSET, in
    which case the display name falls back to empty string.

    Args:
        variant: VariantResponse fetched with `?extend=product_or_material`.

    Returns:
        Formatted variant name with config values, or empty string if no name available.

    Example:
        ```python
        from katana_public_api_client import KatanaClient
        from katana_public_api_client.api.variant import get_all_variants
        from katana_public_api_client.models.get_all_variants_extend_item import (
            GetAllVariantsExtendItem,
        )
        from katana_public_api_client.utils import (
            get_variant_display_name,
            unwrap_data,
        )

        async with KatanaClient() as client:
            response = await get_all_variants.asyncio_detailed(
                client=client,
                extend=[GetAllVariantsExtendItem.PRODUCT_OR_MATERIAL],
            )
            for variant in unwrap_data(response):
                print(get_variant_display_name(variant))
                # e.g. "Premium 140 / Glossy Black / Large / Type A"
        ```
    """
    product_or_material = unwrap_unset(variant.product_or_material, None)
    base_name = ""
    if product_or_material is not None:
        base_name = unwrap_unset(product_or_material.name, "") or ""

    if not base_name:
        return ""

    parts: list[str] = [str(base_name)]
    config_attributes = unwrap_unset(variant.config_attributes, [])
    for attr in config_attributes or []:
        config_value = unwrap_unset(attr.config_value, None)
        if config_value:
            parts.append(str(config_value))

    return " / ".join(parts)

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

Handle a response with custom success and error handlers.

This function provides a convenient way to handle both success and error cases with custom callbacks.

Parameters:

• response (Response[T]) –

  The Response object from an API call

• on_success (Callable[[T], Any] | None, default: None ) –

  Callback function to call with parsed data on success

• on_error (Callable[[APIError], Any] | None, default: None ) –

  Callback function to call with APIError on error

• raise_on_error (bool, default: False ) –

  If True, raise the error even if on_error is provided

Returns:

• Any –

  Result of on_success callback, result of on_error callback, or None

Example

def handle_products(product_list):
    print(f"Got {len(product_list.data)} products")
    return product_list.data


def handle_error(error):
    print(f"Error: {error}")
    return []


response = await get_all_products.asyncio_detailed(client=client)
products = handle_response(
    response, on_success=handle_products, on_error=handle_error
)

Source code in katana_public_api_client/utils.py

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:
    """Handle a response with custom success and error handlers.

    This function provides a convenient way to handle both success and error
    cases with custom callbacks.

    Args:
        response: The Response object from an API call
        on_success: Callback function to call with parsed data on success
        on_error: Callback function to call with APIError on error
        raise_on_error: If True, raise the error even if on_error is provided

    Returns:
        Result of on_success callback, result of on_error callback, or None

    Example:
        ```python
        def handle_products(product_list):
            print(f"Got {len(product_list.data)} products")
            return product_list.data


        def handle_error(error):
            print(f"Error: {error}")
            return []


        response = await get_all_products.asyncio_detailed(client=client)
        products = handle_response(
            response, on_success=handle_products, on_error=handle_error
        )
        ```
    """
    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)

Check if a response was an error (4xx or 5xx status code).

Parameters:

• response (Response[Any]) –

  The Response object to check

Returns:

• bool –

  True if status code is 4xx or 5xx, False otherwise

Source code in katana_public_api_client/utils.py

def is_error(response: Response[Any]) -> bool:
    """Check if a response was an error (4xx or 5xx status code).

    Args:
        response: The Response object to check

    Returns:
        True if status code is 4xx or 5xx, False otherwise
    """
    return response.status_code >= 400

is_success(response)

Check if a response was successful (2xx status code).

Parameters:

• response (Response[Any]) –

  The Response object to check

Returns:

• bool –

  True if status code is 2xx, False otherwise

Example

response = await some_api_call.asyncio_detailed(client=client)
if is_success(response):
    data = unwrap_data(response)
else:
    print(f"Error: {response.status_code}")

Source code in katana_public_api_client/utils.py

def is_success(response: Response[Any]) -> bool:
    """Check if a response was successful (2xx status code).

    Args:
        response: The Response object to check

    Returns:
        True if status code is 2xx, False otherwise

    Example:
        ```python
        response = await some_api_call.asyncio_detailed(client=client)
        if is_success(response):
            data = unwrap_data(response)
        else:
            print(f"Error: {response.status_code}")
        ```
    """
    return 200 <= response.status_code < 300

unwrap(response, *, raise_on_error=True)

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

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

Unwrap a Response object and return the parsed data or raise an error.

This is the main utility function for handling API responses. It automatically raises appropriate exceptions for error responses and returns the parsed data for successful responses.

Parameters:

• response (Response[T]) –

  The Response object from an API call

• raise_on_error (bool, default: True ) –

  If True, raise exceptions on error status codes. If False, return None on errors.

Returns:

• T | None –

  The parsed response data

Raises:

• AuthenticationError –

  When status is 401

• ValidationError –

  When status is 422

• RateLimitError –

  When status is 429

• ServerError –

  When status is 5xx

• APIError –

  For other error status codes

Example

from katana_public_api_client import KatanaClient
from katana_public_api_client.api.product import get_all_products
from katana_public_api_client.utils import unwrap

async with KatanaClient() as client:
    response = await get_all_products.asyncio_detailed(client=client)
    product_list = unwrap(
        response
    )  # Raises on error, returns parsed data
    products = product_list.data  # List of Product objects

Source code in katana_public_api_client/utils.py

def unwrap[T](
    response: Response[T],
    *,
    raise_on_error: bool = True,
) -> T | None:
    """Unwrap a Response object and return the parsed data or raise an error.

    This is the main utility function for handling API responses. It automatically
    raises appropriate exceptions for error responses and returns the parsed data
    for successful responses.

    Args:
        response: The Response object from an API call
        raise_on_error: If True, raise exceptions on error status codes.
                        If False, return None on errors.

    Returns:
        The parsed response data

    Raises:
        AuthenticationError: When status is 401
        ValidationError: When status is 422
        RateLimitError: When status is 429
        ServerError: When status is 5xx
        APIError: For other error status codes

    Example:
        ```python
        from katana_public_api_client import KatanaClient
        from katana_public_api_client.api.product import get_all_products
        from katana_public_api_client.utils import unwrap

        async with KatanaClient() as client:
            response = await get_all_products.asyncio_detailed(client=client)
            product_list = unwrap(
                response
            )  # Raises on error, returns parsed data
            products = product_list.data  # List of Product objects
        ```
    """
    if response.parsed is None:
        # 2xx + empty body is a legitimate no-content success (Katana
        # uses 204 for ``POST /bom_rows`` and most DELETEs). The previous
        # behavior raised ``APIError`` here, which broke every per-row
        # apply against a no-body endpoint (#809). Callers that need a
        # body should keep using ``unwrap_as`` (raises on None).
        #
        # 2xx + *non-empty* body falls through to the error path on
        # purpose: ``parsed`` is None there because ``_parse_response``
        # didn't recognize the status (e.g. the server starts returning
        # 201 after an API change but the spec still declares 204). The
        # body is present and unparsed — raising surfaces the schema
        # drift instead of silently dropping the response.
        if is_success(response) and not response.content:
            return None
        if not raise_on_error:
            return None
        name, message, parsed_error = _try_parse_error_body(response.content)
        _raise_for_status(response.status_code, name, message, parsed_error)

    # Check if it's an error response — assign to local for type narrowing
    parsed = response.parsed
    if isinstance(parsed, ErrorResponse | DetailedErrorResponse):
        if not raise_on_error:
            return None
        name, message, parsed_error = _extract_error_fields(parsed)
        _raise_for_status(response.status_code, name, message, parsed_error)

    return response.parsed

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

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

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

Unwrap a Response and validate the parsed data is of the expected type.

This is a convenience function that combines unwrap() with type validation. It's useful when you expect a specific model type from an API response.

Parameters:

• response (Response[T]) –

  The Response object from an API call

• expected_type (type[ExpectedT]) –

  The expected type of the parsed response

• raise_on_error (bool, default: True ) –

  If True, raise exceptions on error status codes. If False, returns None on error instead of raising.

Returns:

• ExpectedT | None –

  The parsed response data, typed as ExpectedT (or ExpectedT | None if

• ExpectedT | None –

  raise_on_error=False)

Raises:

• Same exceptions as unwrap(), plus –
• TypeError –

  If the parsed response is not of the expected type

Example

from katana_public_api_client import KatanaClient
from katana_public_api_client.api.sales_order import get_sales_order
from katana_public_api_client.models import SalesOrder
from katana_public_api_client.utils import unwrap_as

async with KatanaClient() as client:
    response = await get_sales_order.asyncio_detailed(
        client=client, id=123
    )
    order = unwrap_as(response, SalesOrder)  # Type-safe SalesOrder
    print(order.order_no)

Source code in katana_public_api_client/utils.py

def unwrap_as[T, ExpectedT](
    response: Response[T],
    expected_type: type[ExpectedT],
    *,
    raise_on_error: bool = True,
) -> ExpectedT | None:
    """Unwrap a Response and validate the parsed data is of the expected type.

    This is a convenience function that combines unwrap() with type validation.
    It's useful when you expect a specific model type from an API response.

    Args:
        response: The Response object from an API call
        expected_type: The expected type of the parsed response
        raise_on_error: If True, raise exceptions on error status codes.
            If False, returns None on error instead of raising.

    Returns:
        The parsed response data, typed as ExpectedT (or ExpectedT | None if
        raise_on_error=False)

    Raises:
        Same exceptions as unwrap(), plus:
        TypeError: If the parsed response is not of the expected type

    Example:
        ```python
        from katana_public_api_client import KatanaClient
        from katana_public_api_client.api.sales_order import get_sales_order
        from katana_public_api_client.models import SalesOrder
        from katana_public_api_client.utils import unwrap_as

        async with KatanaClient() as client:
            response = await get_sales_order.asyncio_detailed(
                client=client, id=123
            )
            order = unwrap_as(response, SalesOrder)  # Type-safe SalesOrder
            print(order.order_no)
        ```
    """
    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: Literal[True] = True,
    default: None = None
) -> Any

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

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

Unwrap a Response and extract the data list from list responses.

This is a convenience function that unwraps the response and extracts the .data field from list response objects (like ProductListResponse, WebhookListResponse, etc.).

Parameters:

• response (Response[T]) –

  The Response object from an API call

• raise_on_error (bool, default: True ) –

  If True, raise exceptions on error status codes. If False, return default on errors.

• default (list[DataT] | None, default: None ) –

  Default value to return if data is not available

Returns:

• Any | None –

  List of data objects, or default if not available

Example

from katana_public_api_client import KatanaClient
from katana_public_api_client.api.product import get_all_products
from katana_public_api_client.utils import unwrap_data

async with KatanaClient() as client:
    response = await get_all_products.asyncio_detailed(client=client)
    products = unwrap_data(response)  # Directly get list of Products
    for product in products:
        print(product.name)

Source code in katana_public_api_client/utils.py

def unwrap_data[T, DataT](
    response: Response[T],
    *,
    raise_on_error: bool = True,
    default: list[DataT] | None = None,
) -> Any | None:
    """Unwrap a Response and extract the data list from list responses.

    This is a convenience function that unwraps the response and extracts
    the `.data` field from list response objects (like ProductListResponse,
    WebhookListResponse, etc.).

    Args:
        response: The Response object from an API call
        raise_on_error: If True, raise exceptions on error status codes.
                        If False, return default on errors.
        default: Default value to return if data is not available

    Returns:
        List of data objects, or default if not available

    Raises:
        Same exceptions as unwrap()

    Example:
        ```python
        from katana_public_api_client import KatanaClient
        from katana_public_api_client.api.product import get_all_products
        from katana_public_api_client.utils import unwrap_data

        async with KatanaClient() as client:
            response = await get_all_products.asyncio_detailed(client=client)
            products = unwrap_data(response)  # Directly get list of Products
            for product in products:
                print(product.name)
        ```
    """
    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

    # Extract data field if it exists
    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 there's no data field and no default, wrap the object in a list
    if default is not None:
        return default

    # If it's not a list response, return it as a single-item list
    return [parsed]