guide.md

KatanaClient Guide

The KatanaClient is the modern, pythonic client for the Katana Manufacturing ERP API. It provides automatic resilience (retries, rate limiting, error handling) using httpx's native transport layer - no decorators or wrapper methods needed.

🎯 Key Features

• 🛡️ Automatic Resilience: Transport-layer retries and rate limiting
• 🚀 Zero Configuration: Works out of the box with environment variables
• 📦 Complete Type Safety: Full type hints and IDE support
• 🔄 Smart Pagination: Built-in pagination with safety limits
• 🔍 Rich Observability: Structured logging and metrics
• ⚡ Pythonic Design: Uses httpx's native extension points

🚀 Quick Start

Installation & Setup

# Install the client
pip install -e .

# Create .env file with credentials
echo "KATANA_API_KEY=your-api-key-here" > .env
echo "KATANA_BASE_URL=https://api.katanamrp.com/v1" >> .env

Basic Usage

import asyncio

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 def main():
    # Automatic configuration from .env file
    async with KatanaClient() as client:
        # Direct API usage - automatic resilience built-in
        response = await get_all_products.asyncio_detailed(client=client, limit=50)

        products = unwrap_data(response, default=[])
        print(f"Retrieved {len(products)} products")


asyncio.run(main())

Synchronous endpoints

Generated endpoints also expose sync and sync_detailed. Use a regular context manager to close their separate synchronous HTTP client:

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

with KatanaClient(api_key="your-api-key") as client:
    response = get_all_products.sync_detailed(client=client, limit=50, page=1)

Synchronous calls perform a single request. Automatic retries, adaptive rate limiting, auto-pagination, and the event_hooks passed to KatanaClient apply only to asynchronous calls. Synchronous callers must handle subsequent pages and retry policy themselves. Authentication, timeout, TLS, and connection settings still apply. To customize synchronous transport or hooks, construct an httpx.Client and pass it to client.set_httpx_client(...); that supplied client owns its authentication and other settings. Default sync and async transports have independent lifetimes. An explicitly supplied transport is reused for sync calls if it supports them; an async-only override raises a clear error instead of falling back to real network requests.

📥 Response Handling

Use the helper utilities in katana_public_api_client.utils for consistent response handling. They replace manual status-code checks, give you typed errors, and handle the {"data": [...]} wrapping that Katana applies to every list endpoint.

Helpers at a glance

from katana_public_api_client.utils import unwrap, unwrap_as, unwrap_data, is_success
from katana_public_api_client.domain.converters import unwrap_unset

# Single-object responses (200 OK with parsed model — includes POST creates)
order = unwrap_as(response, ManufacturingOrder)  # type-safe, raises on error

# List responses (200 OK with `data` array)
items = unwrap_data(response, default=[])  # extracts the .data field

# Success-only responses (204 No Content — typically DELETE)
if is_success(response):
    ...

# attrs model fields that may be UNSET
status = unwrap_unset(order.status, None)  # returns None if UNSET

Katana POST creates return 200, not 201. Treat them like any other 200-with-parsed-body response — use unwrap_as(response, Type). See spec-authoring.md for the full convention.

When to use each

Scenario	Pattern	Example
Single object (200)	unwrap_as(response, Type)	Get / update / create ops
List endpoint (200)	unwrap_data(response, default=[])	List operations
Delete (204)	is_success(response)	DELETE
attrs UNSET field	unwrap_unset(field, default)	Optional API fields

Anti-patterns

# ❌ DON'T: manual status code checks
if response.status_code == 200:
    result = response.parsed
# ✅ DO: use helpers
result = unwrap_as(response, ExpectedType)

# ❌ DON'T: isinstance with UNSET
if not isinstance(value, type(UNSET)):
    use(value)
# ✅ DO: use unwrap_unset
use(unwrap_unset(value, default))

# ❌ DON'T: hasattr for attrs-defined fields
if hasattr(order, "status"):
    status = order.status
# ✅ DO: use unwrap_unset (attrs fields always exist, may be UNSET)
status = unwrap_unset(order.status, None)

# ❌ DON'T: wrap API methods to add retries / rate limiting
# ✅ DO: nothing — resilience is at the transport layer; every endpoint inherits it.

# ❌ DON'T: build attrs request bodies with conditional UNSET
optional_field = value if value is not None else UNSET
# ✅ DO: use to_unset
from katana_public_api_client.domain.converters import to_unset

optional_field = to_unset(value)

Exception hierarchy

unwrap() and unwrap_as() raise typed exceptions on non-2xx responses:

• AuthenticationError — 401 Unauthorized
• ValidationError — 422 Unprocessable Entity
• RateLimitError — 429 Too Many Requests (retries exhausted)
• ServerError — 5xx server errors
• APIError — other errors (400, 403, 404, …)

is_success(response) returns True for any 2xx status without raising.

🛡️ Automatic Resilience

Every API call through KatanaClient automatically includes:

Smart Retries

• Network Errors: Automatic retry with exponential backoff (1s, 2s, 4s, 8s, 16s)
• Rate Limits (429): ALL methods (including POST/PATCH) retry with Retry-After header support
• Server Errors (502/503/504): Only idempotent methods (GET, PUT, DELETE, etc.) are retried
• Client Errors (4xx except 429): No retries

async with KatanaClient(max_retries=5) as client:
    # This call will automatically retry on failures
    # POST/PATCH requests will retry on 429 but not on 5xx errors
    # GET/PUT/DELETE requests will retry on both 429 and 5xx errors
    response = await get_all_products.asyncio_detailed(client=client, limit=100)
    # No decorators or wrapper methods needed!

Rate Limit Handling

The client throttles requests in two layers, both automatic:

1. Proactive throttling via RateLimitTransport — every outgoing request passes through a pyrate-limiter token bucket sized to Katana's documented 60 req/min budget. Concurrent and paginated workloads pace cleanly without thrashing 429s.
2. Reactive Retry-After handling in RetryTransport — if a 429 still slips through (e.g. another client is sharing the API key), the retry layer honors Retry-After (numeric seconds or HTTP-date) before the next attempt.

The proactive layer also reads X-Ratelimit-Remaining and X-Ratelimit-Reset on every response and adapts:

• If the server reports fewer remaining tokens than the local estimate (e.g., another client consumed budget), the local bucket is drained to match.
• When X-Ratelimit-Remaining hits 0, future requests are gated by an asyncio.Event until X-Ratelimit-Reset elapses.

# Default behavior — 60/min budget, fully automatic.
async with KatanaClient() as client:
    # 100 calls fire as fast as the rate limit allows; no manual pacing
    # needed. We don't pass an explicit ``page`` parameter — that would
    # disable auto-pagination — so each call here is a full-result fetch
    # demonstrating throttling.
    for _ in range(100):
        response = await get_all_products.asyncio_detailed(
            client=client,
            limit=50,
        )

Tuning the budget

If your account has a different rate limit (some Katana plans differ), pass requests_per_minute=:

async with KatanaClient(requests_per_minute=120) as client:
    ...  # paces at 120/min instead of 60/min

Opting out

If you want to manage throttling yourself (or in a test that needs raw throughput), pass requests_per_minute=None to skip the rate-limit layer entirely:

# Reactive 429-retry only; no proactive throttling.
async with KatanaClient(requests_per_minute=None) as client:
    ...

Error Recovery

import logging

# Configure logging to see resilience in action
logging.basicConfig(level=logging.INFO)

async with KatanaClient() as client:
    # Automatic error recovery with detailed logging
    response = await get_all_products.asyncio_detailed(client=client, limit=100)
    # Logs will show retry attempts and recovery

🔄 Smart Pagination

Auto-pagination is ON by default for all GET requests. All pages are automatically collected into a single response.

Automatic Pagination (Default)

async with KatanaClient() as client:
    # Get ALL products across all pages automatically
    all_products = await get_all_products.asyncio_detailed(
        client=client,
        is_sellable=True,  # API filter parameters
        limit=250,  # Page size (all pages still collected)
    )
    print(f"Total products: {len(all_products.parsed.data)}")

Single Page Request (Explicit Page)

To get a specific page instead of all pages, add an explicit page parameter. Note: ANY explicit page value (including page=1) disables auto-pagination:

async with KatanaClient() as client:
    # Get ONLY page 2 (auto-pagination disabled when page is explicit)
    page2_products = await get_all_products.asyncio_detailed(
        client=client,
        page=2,  # Explicit page disables auto-pagination
        limit=50,
    )
    # Returns just the 50 items on page 2

    # page=1 ALSO disables auto-pagination (returns only first page)
    first_page = await get_all_products.asyncio_detailed(
        client=client,
        page=1,  # Get ONLY page 1, not all pages
        limit=50,
    )

Limiting Total Items

To limit the total number of items collected (not just page size), use the max_items extension via the httpx client:

async with KatanaClient() as client:
    httpx_client = client.get_async_httpx_client()
    response = await httpx_client.get(
        "/products",
        params={"limit": 50},  # 50 items per page
        extensions={"max_items": 200},  # Stop after 200 items total
    )

The transport intelligently adjusts the limit on the final request to fetch only what's needed, avoiding over-fetching.

Pagination Behavior Summary

Parameter	Scope	Effect
limit=50	URL param	Page size (50 items per request)
page=2	URL param	Get specific page only (disables auto-pagination)
max_pages=5	Client	Max pages to fetch
max_items=200	Extension	Max total items to collect

⚙️ Configuration

Authentication Methods

The client supports multiple authentication methods (in priority order):

1. Direct parameter: Pass api_key to KatanaClient()
2. Environment variable: Set KATANA_API_KEY
3. .env file: Create a .env file with your credentials
4. ~/.netrc file: Use standard Unix credential file

Using .env file (Recommended)

# Create .env file
KATANA_API_KEY=your-api-key-here
KATANA_BASE_URL=https://api.katanamrp.com/v1  # Optional

# Automatically loads from .env
async with KatanaClient() as client:
    # Uses credentials from .env
    pass

Using ~/.netrc file

For centralized credential management across multiple tools:

# Add to ~/.netrc
machine api.katanamrp.com
password your-api-key-here

# Set proper permissions (required)
chmod 600 ~/.netrc

# Automatically loads from ~/.netrc
async with KatanaClient() as client:
    # Uses credentials from ~/.netrc
    pass

Note: The password field in netrc stores your API key (bearer token), not an actual password. The login field is optional and ignored.

Using environment variable

export KATANA_API_KEY=your-api-key-here

Using direct parameter

async with KatanaClient(api_key="your-api-key-here") as client:
    # Explicit API key
    pass

Custom Configuration

import logging

# Custom configuration with stdlib logger
async with KatanaClient(
    api_key="custom-key",
    base_url="https://custom.katana.com/v1",
    timeout=60.0,  # Request timeout
    max_retries=5,  # Maximum retry attempts
    logger=logging.getLogger("custom"),  # stdlib logger
) as client:
    # Your API calls here
    pass

The logger parameter accepts any object whose debug, info, warning, and error methods accept (msg, *args, **kwargs) — the standard logging.Logger call convention. Both logging.Logger and structlog's BoundLogger satisfy this without adapters:

import structlog

# Use structlog directly — no adapter needed
logger = structlog.get_logger("katana")
async with KatanaClient(logger=logger) as client:
    pass

Advanced httpx Configuration

import httpx

# Pass through httpx configuration
async with KatanaClient(
    # Standard KatanaClient options
    api_key="your-key",
    max_retries=3,
    # httpx client options
    verify=False,  # SSL verification
    proxies="http://proxy:8080",
    headers={"Custom": "Header"},
    event_hooks={"request": [custom_request_hook], "response": [custom_response_hook]},
) as client:
    # Client has both resilience AND custom httpx config
    pass

🔍 Observability

Logging

import logging

# Configure logging to see what's happening
logging.basicConfig(
    level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)

async with KatanaClient() as client:
    # All resilience actions are logged
    response = await get_all_products.asyncio_detailed(client=client, limit=100)

Example log output:

2025-01-15 10:30:15 - katana_client - WARNING - Rate limited on attempt 1, waiting 60.0s
2025-01-15 10:31:16 - katana_client - INFO - Request succeeded after 2 attempts
2025-01-15 10:31:16 - katana_client - DEBUG - Response: 200 GET https://api.katanamrp.com/v1/products (1.24s)

Custom Event Hooks

async def custom_response_hook(response):
    """Custom hook to track API usage."""
    print(f"API call: {response.request.method} {response.request.url}")
    print(f"Status: {response.status_code}")


async with KatanaClient(event_hooks={"response": [custom_response_hook]}) as client:
    # Your custom hooks are called alongside built-in ones
    response = await get_all_products.asyncio_detailed(client=client, limit=10)

🧪 Testing

Mocking for Tests

import pytest
from unittest.mock import AsyncMock, patch
from katana_public_api_client import KatanaClient


@pytest.mark.asyncio
async def test_api_integration():
    """Test API integration with mocked responses."""
    with patch.dict("os.environ", {"KATANA_API_KEY": "test-key"}):
        async with KatanaClient() as client:
            # Mock the underlying httpx client
            with patch.object(client, "get_async_httpx_client") as mock_httpx:
                mock_response = AsyncMock()
                mock_response.status_code = 200
                mock_response.json.return_value = {"data": [{"id": 1}]}

                mock_httpx.return_value.request = AsyncMock(return_value=mock_response)

                # Test your API logic here
                from katana_public_api_client.api.product import get_all_products

                response = await get_all_products.asyncio_detailed(
                    client=client, limit=10
                )

                assert response.status_code == 200

Integration Tests

import os

import pytest

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


@pytest.mark.integration
@pytest.mark.asyncio
async def test_real_api():
    """Test against real Katana API (requires KATANA_API_KEY)."""
    api_key = os.getenv("KATANA_API_KEY")
    if not api_key:
        pytest.skip("KATANA_API_KEY not set")

    async with KatanaClient() as client:
        response = await get_all_products.asyncio_detailed(client=client, limit=1)

        assert response.status_code == 200
        assert hasattr(response.parsed, "data")

🔧 Advanced Patterns

Custom Transport

from katana_public_api_client.katana_client import ResilientAsyncTransport

# Create custom transport with different settings
custom_transport = ResilientAsyncTransport(
    max_retries=10,
    max_pages=50,  # Limit automatic pagination
    logger=custom_logger,
)

async with KatanaClient(transport=custom_transport) as client:
    # Uses your custom retry logic
    pass

Batch Operations

import asyncio
from katana_public_api_client.api.product import get_product


async def process_products_in_batches(product_ids, batch_size=10):
    """Process products in batches with automatic resilience."""
    async with KatanaClient() as client:
        results = []
        for i in range(0, len(product_ids), batch_size):
            batch = product_ids[i : i + batch_size]

            # Each call automatically has resilience
            batch_results = await asyncio.gather(
                *[
                    get_product.asyncio_detailed(client=client, id=product_id)
                    for product_id in batch
                ]
            )

            results.extend(batch_results)

            # Be nice to the API
            await asyncio.sleep(0.1)

        return results

Error Handling

import httpx
from katana_public_api_client.utils import unwrap_data, APIError

async with KatanaClient() as client:
    try:
        response = await get_all_products.asyncio_detailed(client=client, limit=50)
        products = unwrap_data(response, default=[])
        print(f"Success: {len(products)} products")

    except httpx.TimeoutException:
        print("Request timed out after retries")
    except httpx.ConnectError:
        print("Connection failed after retries")
    except APIError as e:
        print(f"Unexpected API response: {e.status_code}")

📚 Best Practices

1. Use Context Managers

# ✅ Good: Properly manages connections
async with KatanaClient() as client:
    response = await get_all_products.asyncio_detailed(client=client, limit=50)

# ❌ Bad: Doesn't close connections
client = KatanaClient()
response = await get_all_products.asyncio_detailed(client=client, limit=50)

2. Configure Appropriate Timeouts

# ✅ Good: Reasonable timeout for your use case
async with KatanaClient(timeout=30.0) as client:
    # Timeout is appropriate for expected response time
    pass

# ❌ Bad: Too short timeout causes unnecessary retries
async with KatanaClient(timeout=1.0) as client:
    # Will likely timeout and retry frequently
    pass

3. Let Auto-Pagination Handle Large Datasets

# ✅ Good: Auto-pagination is ON by default with safety limits
all_products = await get_all_products.asyncio_detailed(
    client=client,
    limit=250,  # Sets page size; all pages collected automatically
)

# ✅ Good: Use explicit page when you need just one page
page2 = await get_all_products.asyncio_detailed(
    client=client,
    page=2,  # Explicit page = single page only
    limit=100,
)

# ❌ Bad: Manual pagination loop without safety limits
page = 1
while True:  # Could run forever!
    response = await get_all_products.asyncio_detailed(
        client=client, page=page, limit=100
    )
    # ... handle response
    page += 1

4. Handle Different Response Types

from katana_public_api_client.utils import unwrap_data, AuthenticationError, APIError

async with KatanaClient() as client:
    response = await get_all_products.asyncio_detailed(client=client, limit=50)

    # ✅ Good: use unwrap helpers + typed exceptions
    try:
        products = unwrap_data(response, default=[])
        print(f"Retrieved {len(products)} products")
    except AuthenticationError:
        print("Authentication failed")
    except APIError as e:
        print(f"Unexpected status: {e.status_code}")

5. Use Direct API Calls

from katana_public_api_client.api.product import get_all_products

# ✅ Good: Direct API calls with automatic resilience
async with KatanaClient() as client:
    response = await get_all_products.asyncio_detailed(client=client, is_sellable=True)
    products = unwrap_data(response, default=[])

🚀 Performance Tips

1. Reuse Client Instances

from katana_public_api_client.api.product import get_all_products
from katana_public_api_client.api.sales_order import get_all_sales_orders
from katana_public_api_client.api.inventory import get_all_inventory_points

# ✅ Good: One client for multiple operations
async with KatanaClient() as client:
    products = await get_all_products.asyncio_detailed(client=client)
    orders = await get_all_sales_orders.asyncio_detailed(client=client)
    inventory = await get_all_inventory_points.asyncio_detailed(client=client)

# ❌ Bad: New client for each operation
for operation in [get_all_products, get_all_sales_orders]:
    async with KatanaClient() as client:
        await operation.asyncio_detailed(client=client)

2. Optimize Page Sizes

# ✅ Good: Reasonable page size
products = await get_all_products.asyncio_detailed(
    client=client,
    limit=250,  # Good balance of efficiency and memory - automatically paginated
)

# ❌ Bad: Too small (many requests)
products = await get_all_products.asyncio_detailed(
    client=client,
    limit=10,  # Will make many small requests
)

# ❌ Bad: Too large (may hit API limits)
products = await get_all_products.asyncio_detailed(
    client=client,
    limit=10000,  # May exceed API limits
)

3. Use Concurrent Requests Wisely

import asyncio
from katana_public_api_client.api.product import get_product


# ✅ Good: Limited concurrency respects rate limits
async def get_multiple_products(product_ids):
    async with KatanaClient() as client:
        # Process in small batches
        results = []
        for i in range(0, len(product_ids), 5):  # 5 concurrent requests
            batch = product_ids[i : i + 5]
            batch_results = await asyncio.gather(
                *[get_product.asyncio_detailed(client=client, id=pid) for pid in batch]
            )
            results.extend(batch_results)

            # Small delay between batches
            if i + 5 < len(product_ids):
                await asyncio.sleep(0.2)

        return results

📖 API Reference

KatanaClient

class KatanaClient(AuthenticatedClient):
    """The pythonic Katana API client with automatic resilience and pagination."""

    def __init__(
        self,
        api_key: Optional[str] = None,
        base_url: Optional[str] = None,
        timeout: float = 30.0,
        max_retries: int = 5,
        max_pages: int = 100,
        logger: Optional[Logger] = None,
        **httpx_kwargs: Any,
    ): ...

    async def __aenter__(self) -> "KatanaClient": ...
    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: ...

ResilientAsyncTransport

Factory function that creates a layered transport with automatic resilience:

def ResilientAsyncTransport(
    max_retries: int = 5,
    max_pages: int = 100,
    logger: Optional[Logger] = None,
    **kwargs: Any,
) -> RetryTransport:
    """
    Creates a chained transport with:
    1. AsyncHTTPTransport (base HTTP transport)
    2. ErrorLoggingTransport (logs detailed 4xx errors)
    3. PaginationTransport (auto-collects paginated responses)
    4. RetryTransport (handles retries with Retry-After header support)
    """
    ...

---

Next Steps: Check out the API Reference for detailed endpoint documentation, or see Testing Guide for testing patterns.