Skip to content

katana_public_api_client

katana_public_api_client

Katana Public API Client - Python client for Katana Manufacturing ERP.

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 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
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

AuthenticatedClient

A Client which has been authenticated for use on secured endpoints

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Manually set the underlying httpx.AsyncClient

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

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

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

Manually set the underlying httpx.Client

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

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

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

Get a new client matching this one with additional cookies

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

Get a new client matching this one with additional headers

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

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

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

AuthenticationError(message, status_code, error_response=None)

Bases: APIError

Raised when authentication fails (401).

Source code in katana_public_api_client/utils.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
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

Client

A class for keeping track of data related to the API

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Manually set the underlying httpx.AsyncClient

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

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

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

Manually set the underlying httpx.Client

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

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

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

Get a new client matching this one with additional cookies

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

Get a new client matching this one with additional headers

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

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

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

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

Bases: AuthenticatedClient

The pythonic Katana API client with automatic resilience and pagination.

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

Features: - Automatic retries on network errors and server errors (5xx) - Automatic rate limit handling with Retry-After header support - Auto-pagination ON by default for GET requests (collects all pages automatically) - Uses 250 items per page (Katana's max) for efficient pagination - Rich logging and observability - Minimal configuration - just works out of the box

Auto-pagination behavior: - ON by default for GET requests with NO page parameter - Uses 250 items per page when no limit specified by caller - If caller specifies a limit, that limit is used per page - ANY explicit page parameter disables auto-pagination (e.g., page=1) - Disabled per-request via extensions: extensions={"auto_pagination": False} - Control max pages via max_pages constructor parameter - Limit total items via extensions: extensions={"max_items": 200}

Usage

async with KatanaClient() as client: from katana_public_api_client.api.product import get_all_products

# Auto-pagination is ON - all pages collected automatically
# Uses 250 items per page for efficiency
response = await get_all_products.asyncio_detailed(
    client=client,  # Pass client directly - no .client needed!
)

# Use a custom limit per page (100 instead of 250)
response = await get_all_products.asyncio_detailed(
    client=client,
    limit=100,   # Use 100 per page
)

# Get a specific page only (ANY page param disables auto-pagination)
response = await get_all_products.asyncio_detailed(
    client=client,
    page=2,      # Get page 2 only
    limit=50
)

# Limit total items collected (via httpx client)
httpx_client = client.get_async_httpx_client()
response = await httpx_client.get(
    "/products",
    extensions={"max_items": 200}   # Stop after 200 items
)

# Control max pages globally
client_limited = KatanaClient(max_pages=5)  # Limit to 5 pages max

Parameters:

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

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

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

    Base URL for the Katana API. Defaults to https://api.katanamrp.com/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.

  • requests_per_minute (int | None, default: 60 ) –

    Steady-state request budget for the proactive rate limiter. Defaults to 60 (Katana's documented limit). Set to None to disable the rate limiter entirely (e.g. when callers want to manage throttling themselves, or for tests that need raw throughput). When the limiter is active, every actual HTTP request — including retries and per-page paginated fetches — consumes one token, and the transport adapts to the server's X-Ratelimit-Remaining / X-Ratelimit-Reset headers.

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

Source code in katana_public_api_client/katana_client.py 
1670
1671
1672
1673
1674
1675
1676
1677
1678
1679
1680
1681
1682
1683
1684
1685
1686
1687
1688
1689
1690
1691
1692
1693
1694
1695
1696
1697
1698
1699
1700
1701
1702
1703
1704
1705
1706
1707
1708
1709
1710
1711
1712
1713
1714
1715
1716
1717
1718
1719
1720
1721
1722
1723
1724
1725
1726
1727
1728
1729
1730
1731
1732
1733
1734
1735
1736
1737
1738
1739
1740
1741
1742
1743
1744
1745
1746
1747
1748
1749
1750
1751
1752
1753
1754
1755
1756
1757
1758
1759
1760
1761
1762
1763
1764
1765
1766
1767
1768
1769
1770
1771
1772
1773
1774
1775
1776
1777
1778
1779
1780
1781
1782
1783
1784
1785
1786
1787
1788
1789
1790
1791
1792
1793
1794
1795
1796
1797
1798
1799
1800
1801
1802
1803
1804
1805
1806
1807
1808
1809
1810
1811
1812
1813
1814
1815
1816
1817
1818
1819
1820
1821
1822
1823
1824
1825
1826
1827
1828
1829
1830
1831
1832
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,
    *,
    requests_per_minute: int | None = 60,
    **httpx_kwargs: Any,
):
    """
    Initialize the Katana API client with automatic resilience features.

    Args:
        api_key: Katana API key. If None, will try to load from KATANA_API_KEY env var,
            .env file, or ~/.netrc file (in that order).
        base_url: Base URL for the Katana API. Defaults to https://api.katanamrp.com/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.
        requests_per_minute: Steady-state request budget for the proactive
            rate limiter. Defaults to 60 (Katana's documented limit). Set to
            ``None`` to disable the rate limiter entirely (e.g. when callers
            want to manage throttling themselves, or for tests that need raw
            throughput). When the limiter is active, every actual HTTP
            request — including retries and per-page paginated fetches —
            consumes one token, and the transport adapts to the server's
            ``X-Ratelimit-Remaining`` / ``X-Ratelimit-Reset`` headers.
        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, KATANA_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 KatanaClient() 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("KATANA_BASE_URL") or "https://api.katanamrp.com/v1"
    )

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

    if not api_key:
        raise ValueError(
            "API key required via: api_key param, KATANA_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 class instances (lazy-loaded)
    self._products: Products | None = None
    self._materials: Materials | None = None
    self._variants: Variants | None = None
    self._services: Services | 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,
            requests_per_minute=requests_per_minute,
            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.

Example

async with KatanaClient() as client: ... products = await client.api.products.list(is_sellable=True) ... product = await client.api.products.get(123) ... await client.api.products.delete(123)

materials property

Access material catalog operations.

Returns:

  • Materials –

    Materials instance for material CRUD operations.

Example

async with KatanaClient() as client: ... materials = await client.materials.list() ... material = await client.materials.get(123)

products property

Access product catalog operations.

Returns:

  • Products –

    Products instance for product CRUD and search operations.

Example

async with KatanaClient() as client: ... # Product CRUD ... products = await client.products.list(is_sellable=True) ... product = await client.products.get(123) ... results = await client.products.search("widget")

services property

Access service catalog operations.

Returns:

  • Services –

    Services instance for service CRUD operations.

Example

async with KatanaClient() as client: ... services = await client.services.list() ... service = await client.services.get(123)

variants property

Access variant catalog operations.

Returns:

  • Variants –

    Variants instance for variant CRUD operations.

Example

async with KatanaClient() as client: ... variants = await client.variants.list() ... variant = await client.variants.get(123)

Functions

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 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
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 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
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 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
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 
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
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 
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
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 
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
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 
546
547
548
549
550
551
552
553
554
555
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 
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
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:

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 
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
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_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 
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
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]