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 = (
        error_response.details if error_response and error_response.details else []
    )

Functions

__str__()

Format validation error with code-specific details.

Source code in katana_public_api_client/utils.py

def __str__(self) -> str:
    """Format validation error with code-specific details."""
    msg = super().__str__()

    # Add code-specific details if present using type-safe isinstance checks
    if self.validation_errors:
        error_details = []
        for detail in self.validation_errors:
            field = detail.path.lstrip("/")

            # Use isinstance for type-safe error handling
            if isinstance(detail, EnumValidationError):
                error_details.append(
                    f"  Field '{field}' must be one of: {detail.allowed_values}"
                )

            elif isinstance(detail, MinValidationError):
                error_details.append(
                    f"  Field '{field}' must be >= {detail.minimum}"
                )

            elif isinstance(detail, MaxValidationError):
                error_details.append(
                    f"  Field '{field}' must be <= {detail.maximum}"
                )

            elif isinstance(detail, InvalidTypeValidationError):
                error_details.append(
                    f"  Field '{field}' must be of type: {detail.expected_type}"
                )

            elif isinstance(detail, TooSmallValidationError):
                if not isinstance(detail.min_length, Unset):
                    error_details.append(
                        f"  Field '{field}' must have minimum length: {detail.min_length}"
                    )
                elif not isinstance(detail.min_items, Unset):
                    error_details.append(
                        f"  Field '{field}' must have minimum items: {detail.min_items}"
                    )

            elif isinstance(detail, TooBigValidationError):
                if not isinstance(detail.max_length, Unset):
                    error_details.append(
                        f"  Field '{field}' must have maximum length: {detail.max_length}"
                    )
                elif not isinstance(detail.max_items, Unset):
                    error_details.append(
                        f"  Field '{field}' must have maximum items: {detail.max_items}"
                    )

            elif isinstance(detail, RequiredValidationError):
                error_details.append(
                    f"  Missing required field: '{detail.missing_property}'"
                )

            elif isinstance(detail, PatternValidationError):
                error_details.append(
                    f"  Field '{field}' must match pattern: {detail.pattern}"
                )

            elif isinstance(detail, UnrecognizedKeysValidationError):
                error_details.append(f"  Unrecognized fields: {detail.keys}")
                if not isinstance(detail.valid_keys, Unset):
                    error_details.append(f"  Valid fields: {detail.valid_keys}")

        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

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

    # Type narrowing: at this point parsed is ErrorResponse | DetailedErrorResponse
    parsed_error = response.parsed

    error_message = (
        parsed_error.message  # type: ignore[attr-defined]
        if not isinstance(parsed_error.message, Unset)  # type: ignore[attr-defined]
        else None
    )

    # Check nested error format
    if hasattr(parsed_error, "additional_properties"):
        nested = parsed_error.additional_properties  # type: ignore[attr-defined]
        if isinstance(nested, dict) and "error" in nested:
            nested_error = nested["error"]  # type: ignore[index]
            if isinstance(nested_error, dict):
                error_message = str(nested_error.get("message", error_message))  # type: ignore[arg-type]

    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: "Mayhem 140 / Liquid Black / Large / 5 Star"

When the variant has been fetched with extend=product_or_material, the API returns variants with a nested product_or_material object (Product or Material). This function extracts the base product/material name and appends config attribute values separated by " / ".

Parameters:

• variant (Variant) –

  Variant object (ideally with product_or_material populated)

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_variant
from katana_public_api_client.utils import get_variant_display_name

async with KatanaClient() as client:
    response = await get_variant.asyncio_detailed(client=client, id=123)
    variant = unwrap(response)
    display_name = get_variant_display_name(variant)
    print(display_name)  # "Mayhem 140 / Liquid Black / Large / 5 Star"

Source code in katana_public_api_client/utils.py

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

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

    Example: "Mayhem 140 / Liquid Black / Large / 5 Star"

    When the variant has been fetched with extend=product_or_material, the API
    returns variants with a nested product_or_material object (Product or Material).
    This function extracts the base product/material name and appends config attribute
    values separated by " / ".

    Args:
        variant: Variant object (ideally with product_or_material populated)

    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_variant
        from katana_public_api_client.utils import get_variant_display_name

        async with KatanaClient() as client:
            response = await get_variant.asyncio_detailed(client=client, id=123)
            variant = unwrap(response)
            display_name = get_variant_display_name(variant)
            print(display_name)  # "Mayhem 140 / Liquid Black / Large / 5 Star"
        ```
    """
    # Get base product/material name
    base_name = ""
    if hasattr(variant, "product_or_material") and variant.product_or_material:
        product_or_material = variant.product_or_material
        if hasattr(product_or_material, "name"):
            base_name = product_or_material.name or ""

    if not base_name:
        return ""

    # Append config attribute values (just values, not "name: value")
    parts = [base_name]
    if hasattr(variant, "config_attributes") and variant.config_attributes:
        for attr in variant.config_attributes:
            if hasattr(attr, "config_value") and attr.config_value:
                parts.append(str(attr.config_value))  # type: ignore[arg-type]

    # Join with forward slashes (Katana UI format)
    return " / ".join(parts)  # type: ignore[arg-type]

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(
    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: bool = True
) -> T

unwrap(
    response: Response[T], *, raise_on_error: bool = 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:
        if raise_on_error:
            raise APIError(
                f"No parsed response data for status {response.status_code}",
                response.status_code,
            )
        return None

    # Check if it's an error response
    if isinstance(response.parsed, ErrorResponse | DetailedErrorResponse):
        if not raise_on_error:
            return None

        # Type narrowing: at this point parsed is ErrorResponse | DetailedErrorResponse
        parsed_error = response.parsed

        error_name = (
            parsed_error.name  # type: ignore[attr-defined]
            if not isinstance(parsed_error.name, Unset)  # type: ignore[attr-defined]
            else "Unknown"
        )
        error_message = (
            parsed_error.message  # type: ignore[attr-defined]
            if not isinstance(parsed_error.message, Unset)  # type: ignore[attr-defined]
            else "No error message provided"
        )

        # Handle nested error format
        if hasattr(parsed_error, "additional_properties"):
            nested = parsed_error.additional_properties  # type: ignore[attr-defined]
            if isinstance(nested, dict) and "error" in nested:
                nested_error = nested["error"]  # type: ignore[index]
                if isinstance(nested_error, dict):
                    error_name = str(nested_error.get("name", error_name))  # type: ignore[arg-type]
                    error_message = str(nested_error.get("message", error_message))  # type: ignore[arg-type]

        message = f"{error_name}: {error_message}"

        if response.status_code == HTTPStatus.UNAUTHORIZED:
            raise AuthenticationError(message, response.status_code, parsed_error)  # type: ignore[arg-type]
        elif response.status_code == HTTPStatus.UNPROCESSABLE_ENTITY:
            # ValidationError expects DetailedErrorResponse, but parsed_error could be ErrorResponse
            detailed_error = (
                parsed_error
                if isinstance(parsed_error, DetailedErrorResponse)
                else None
            )
            raise ValidationError(
                message,
                response.status_code,
                detailed_error,
            )
        elif response.status_code == HTTPStatus.TOO_MANY_REQUESTS:
            raise RateLimitError(message, response.status_code, parsed_error)  # type: ignore[arg-type]
        elif 500 <= response.status_code < 600:
            raise ServerError(message, response.status_code, parsed_error)  # type: ignore[arg-type]
        else:
            raise APIError(message, response.status_code, parsed_error)  # type: ignore[arg-type]

    return response.parsed

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 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]