Skip to content

katana_public_api_client.helpers

katana_public_api_client.helpers

Domain classes for the Katana API client.

These domain classes provide ergonomic, domain-specific methods that reduce boilerplate and serve as the foundation for MCP tools.

Example

async with KatanaClient() as client: ... # Product catalog operations ... products = await client.products.list(is_sellable=True) ... product = await client.products.get(123) ... results = await client.products.search("widget") ... ... # Inventory and stock operations ... stock = await client.inventory.check_stock("WIDGET-001") ... low_stock = await client.inventory.list_low_stock(threshold=10)

Classes

Base(client)

Base class for all domain classes.

Provides common functionality and access to the KatanaClient instance.

Parameters:

  • client (KatanaClient) –

    The KatanaClient instance to use for API calls.

Parameters:

  • client (KatanaClient) –

    The KatanaClient instance to use for API calls.

Source code in katana_public_api_client/helpers/base.py 
20
21
22
23
24
25
26
def __init__(self, client: KatanaClient) -> None:
    """Initialize with a client instance.

    Args:
        client: The KatanaClient instance to use for API calls.
    """
    self._client = client
Functions

Inventory(client)

Bases: Base

Inventory and stock operations.

Provides methods for checking stock levels, movements, adjustments, and transfers. For product catalog CRUD, use client.products instead.

Example

async with KatanaClient() as client: ... # Check stock levels (MCP tool support) ... stock = await client.inventory.check_stock("WIDGET-001") ... low_stock = await client.inventory.list_low_stock(threshold=10) ... ... # Stock movements and adjustments ... movements = await client.inventory.get_movements() ... await client.inventory.create_adjustment( ... {"product_id": 123, "quantity": 10} ... )

Source code in katana_public_api_client/helpers/base.py 
20
21
22
23
24
25
26
def __init__(self, client: KatanaClient) -> None:
    """Initialize with a client instance.

    Args:
        client: The KatanaClient instance to use for API calls.
    """
    self._client = client
Functions
check_stock(sku) async

Check stock levels for a specific SKU.

Used by: MCP tool check_inventory

Parameters:

  • sku (str) –

    The SKU to check stock for.

Returns:

  • Product | None –

    Product model with stock information, or None if SKU not found.

Example

product = await client.inventory.check_stock("WIDGET-001") if product: ... stock = product.stock_information ... print(f"Available: {stock.available}, In Stock: {stock.in_stock}")

Source code in katana_public_api_client/helpers/inventory.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
async def check_stock(self, sku: str) -> Product | None:
    """Check stock levels for a specific SKU.

    Used by: MCP tool check_inventory

    Args:
        sku: The SKU to check stock for.

    Returns:
        Product model with stock information, or None if SKU not found.

    Example:
        >>> product = await client.inventory.check_stock("WIDGET-001")
        >>> if product:
        ...     stock = product.stock_information
        ...     print(f"Available: {stock.available}, In Stock: {stock.in_stock}")
    """
    # Note: The API doesn't support direct SKU filtering yet
    # We need to fetch products and filter client-side
    # TODO: When API adds SKU parameter, use that instead
    response = await get_all_products.asyncio_detailed(
        client=self._client,
        limit=100,
    )
    products = unwrap_data(response)

    # Find product by SKU - check both product.sku and variant.sku
    for product in products:
        # Check if product has sku attribute directly
        if hasattr(product, "sku") and product.sku == sku:
            return product

        # Check variants for matching SKU
        if hasattr(product, "variants"):
            for variant in product.variants or []:
                if hasattr(variant, "sku") and variant.sku == sku:
                    return product

    return None
get_inventory_points(**filters) async

Get inventory points for all products.

This will use the inventory API to get current stock levels, reorder points, and safety stock levels.

Parameters:

  • **filters (Any, default: {} ) –

    Filtering parameters.

Returns:

Note

To be implemented using katana_public_api_client.api.inventory

Source code in katana_public_api_client/helpers/inventory.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
async def get_inventory_points(self, **filters: Any) -> list[dict[str, Any]]:
    """Get inventory points for all products.

    This will use the inventory API to get current stock levels,
    reorder points, and safety stock levels.

    Args:
        **filters: Filtering parameters.

    Returns:
        List of inventory point data.

    Note:
        To be implemented using katana_public_api_client.api.inventory
    """
    # TODO: Implement using get_all_inventory_point API
    raise NotImplementedError("Coming soon - will use inventory API")
get_negative_stock() async

Get products with negative stock.

Returns:

Note

To be implemented using katana_public_api_client.api.inventory

Source code in katana_public_api_client/helpers/inventory.py 
142
143
144
145
146
147
148
149
150
151
152
async def get_negative_stock(self) -> list[dict[str, Any]]:
    """Get products with negative stock.

    Returns:
        List of products with negative inventory.

    Note:
        To be implemented using katana_public_api_client.api.inventory
    """
    # TODO: Implement using get_all_negative_stock API
    raise NotImplementedError("Coming soon - will use inventory API")
list_low_stock(threshold=None) async

Find products below their reorder point.

Used by: MCP tool list_low_stock_items

Parameters:

  • threshold (int | None, default: None ) –

    Optional stock threshold. Products with stock below this will be returned. If None, uses each product's reorder point.

Returns:

  • list[Product] –

    List of Product models that are below stock threshold.

Example

low_stock = await client.inventory.list_low_stock(threshold=10) for product in low_stock: ... stock = product.stock_information ... print(f"{product.sku}: {stock.in_stock} units")

Source code in katana_public_api_client/helpers/inventory.py 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
async def list_low_stock(self, threshold: int | None = None) -> list[Product]:
    """Find products below their reorder point.

    Used by: MCP tool list_low_stock_items

    Args:
        threshold: Optional stock threshold. Products with stock below this will be returned.
                  If None, uses each product's reorder point.

    Returns:
        List of Product models that are below stock threshold.

    Example:
        >>> low_stock = await client.inventory.list_low_stock(threshold=10)
        >>> for product in low_stock:
        ...     stock = product.stock_information
        ...     print(f"{product.sku}: {stock.in_stock} units")
    """
    # Note: Stock information is included in product response by default
    response = await get_all_products.asyncio_detailed(
        client=self._client,
        limit=100,  # KatanaClient handles pagination automatically
    )
    products = unwrap_data(response)

    low_stock_items = []
    for product in products:
        stock_info = getattr(product, "stock_information", None)
        if not stock_info:
            continue

        in_stock = getattr(stock_info, "in_stock", 0) or 0
        reorder_point = getattr(stock_info, "reorder_point", 0)

        # Determine if this is low stock
        is_low = False
        if threshold is not None:
            is_low = in_stock < threshold
        elif reorder_point > 0:
            is_low = in_stock < reorder_point

        if is_low:
            low_stock_items.append(product)

    return low_stock_items
set_reorder_point(product_id, quantity) async

Set reorder point for a product.

Parameters:

  • product_id (int) –

    The product ID.

  • quantity (int) –

    The reorder point quantity.

Note

To be implemented using katana_public_api_client.api.inventory

Source code in katana_public_api_client/helpers/inventory.py 
154
155
156
157
158
159
160
161
162
163
164
165
async def set_reorder_point(self, product_id: int, quantity: int) -> None:
    """Set reorder point for a product.

    Args:
        product_id: The product ID.
        quantity: The reorder point quantity.

    Note:
        To be implemented using katana_public_api_client.api.inventory
    """
    # TODO: Implement using create_inventory_reorder_point API
    raise NotImplementedError("Coming soon - will use inventory API")
set_safety_stock(product_id, quantity) async

Set safety stock level for a product.

Parameters:

  • product_id (int) –

    The product ID.

  • quantity (int) –

    The safety stock quantity.

Note

To be implemented using katana_public_api_client.api.inventory

Source code in katana_public_api_client/helpers/inventory.py 
167
168
169
170
171
172
173
174
175
176
177
178
async def set_safety_stock(self, product_id: int, quantity: int) -> None:
    """Set safety stock level for a product.

    Args:
        product_id: The product ID.
        quantity: The safety stock quantity.

    Note:
        To be implemented using katana_public_api_client.api.inventory
    """
    # TODO: Implement using create_inventory_safety_stock_level API
    raise NotImplementedError("Coming soon - will use inventory API")

Materials(client)

Bases: Base

Material catalog management.

Provides CRUD operations for materials in the Katana catalog.

Example

async with KatanaClient() as client: ... # CRUD operations ... materials = await client.materials.list() ... material = await client.materials.get(123) ... new_material = await client.materials.create({"name": "Steel"})

Source code in katana_public_api_client/helpers/base.py 
20
21
22
23
24
25
26
def __init__(self, client: KatanaClient) -> None:
    """Initialize with a client instance.

    Args:
        client: The KatanaClient instance to use for API calls.
    """
    self._client = client
Functions
create(material_data) async

Create a new material.

Parameters:

Returns:

Example

from katana_public_api_client.models import CreateMaterialRequest new_material = await client.materials.create( ... CreateMaterialRequest(name="Steel") ... )

Source code in katana_public_api_client/helpers/materials.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
async def create(self, material_data: CreateMaterialRequest) -> KatanaMaterial:
    """Create a new material.

    Args:
        material_data: CreateMaterialRequest model with material details.

    Returns:
        Created KatanaMaterial domain model object.

    Example:
        >>> from katana_public_api_client.models import CreateMaterialRequest
        >>> new_material = await client.materials.create(
        ...     CreateMaterialRequest(name="Steel")
        ... )
    """
    response = await create_material.asyncio_detailed(
        client=self._client,
        body=material_data,
    )
    # unwrap() raises on errors, so cast is safe
    attrs_material = cast(Material, unwrap(response))
    return material_to_katana(attrs_material)
delete(material_id) async

Delete a material.

Parameters:

  • material_id (int) –

    The material ID to delete.

Example

await client.materials.delete(123)

Source code in katana_public_api_client/helpers/materials.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
async def delete(self, material_id: int) -> None:
    """Delete a material.

    Args:
        material_id: The material ID to delete.

    Example:
        >>> await client.materials.delete(123)
    """
    await delete_material.asyncio_detailed(
        client=self._client,
        id=material_id,
    )
get(material_id) async

Get a specific material by ID.

Parameters:

  • material_id (int) –

    The material ID.

Returns:

Example

material = await client.materials.get(123)

Source code in katana_public_api_client/helpers/materials.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
async def get(self, material_id: int) -> KatanaMaterial:
    """Get a specific material by ID.

    Args:
        material_id: The material ID.

    Returns:
        KatanaMaterial domain model object.

    Example:
        >>> material = await client.materials.get(123)
    """
    response = await get_material.asyncio_detailed(
        client=self._client,
        id=material_id,
    )
    # unwrap() raises on errors, so cast is safe
    attrs_material = cast(Material, unwrap(response))
    return material_to_katana(attrs_material)
list(**filters) async

List all materials with optional filters.

Parameters:

  • **filters (Any, default: {} ) –

    Filtering parameters.

Returns:

Example

materials = await client.materials.list(limit=100)

Source code in katana_public_api_client/helpers/materials.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
async def list(self, **filters: Any) -> list[KatanaMaterial]:
    """List all materials with optional filters.

    Args:
        **filters: Filtering parameters.

    Returns:
        List of KatanaMaterial domain model objects.

    Example:
        >>> materials = await client.materials.list(limit=100)
    """
    response = await get_all_materials.asyncio_detailed(
        client=self._client,
        **filters,
    )
    attrs_materials = unwrap_data(response)
    return materials_to_katana(attrs_materials)
update(material_id, material_data) async

Update an existing material.

Parameters:

  • material_id (int) –

    The material ID to update.

  • material_data (UpdateMaterialRequest) –

    UpdateMaterialRequest model with fields to update.

Returns:

Example

from katana_public_api_client.models import UpdateMaterialRequest updated = await client.materials.update( ... 123, UpdateMaterialRequest(name="Aluminum") ... )

Source code in katana_public_api_client/helpers/materials.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
async def update(
    self, material_id: int, material_data: UpdateMaterialRequest
) -> KatanaMaterial:
    """Update an existing material.

    Args:
        material_id: The material ID to update.
        material_data: UpdateMaterialRequest model with fields to update.

    Returns:
        Updated KatanaMaterial domain model object.

    Example:
        >>> from katana_public_api_client.models import UpdateMaterialRequest
        >>> updated = await client.materials.update(
        ...     123, UpdateMaterialRequest(name="Aluminum")
        ... )
    """
    response = await update_material.asyncio_detailed(
        client=self._client,
        id=material_id,
        body=material_data,
    )
    # unwrap() raises on errors, so cast is safe
    attrs_material = cast(Material, unwrap(response))
    return material_to_katana(attrs_material)

Products(client)

Bases: Base

Product catalog management.

Provides CRUD operations and search for products in the Katana catalog.

Example

async with KatanaClient() as client: ... # Search products ... products = await client.products.search("widget") ... ... # CRUD operations ... products = await client.products.list(is_sellable=True) ... product = await client.products.get(123) ... new_product = await client.products.create({"name": "Widget"})

Source code in katana_public_api_client/helpers/base.py 
20
21
22
23
24
25
26
def __init__(self, client: KatanaClient) -> None:
    """Initialize with a client instance.

    Args:
        client: The KatanaClient instance to use for API calls.
    """
    self._client = client
Functions
create(product_data) async

Create a new product.

Parameters:

Returns:

Example

from katana_public_api_client.models import CreateProductRequest new_product = await client.products.create( ... CreateProductRequest( ... name="New Widget", ... sku="WIDGET-NEW", ... is_sellable=True, ... variants=[], ... ) ... )

Source code in katana_public_api_client/helpers/products.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
async def create(self, product_data: CreateProductRequest) -> KatanaProduct:
    """Create a new product.

    Args:
        product_data: CreateProductRequest model with product details.

    Returns:
        Created KatanaProduct domain model object.

    Example:
        >>> from katana_public_api_client.models import CreateProductRequest
        >>> new_product = await client.products.create(
        ...     CreateProductRequest(
        ...         name="New Widget",
        ...         sku="WIDGET-NEW",
        ...         is_sellable=True,
        ...         variants=[],
        ...     )
        ... )
    """
    response = await create_product.asyncio_detailed(
        client=self._client,
        body=product_data,
    )
    # unwrap() raises on errors, so cast is safe
    attrs_product = cast(Product, unwrap(response))
    return product_to_katana(attrs_product)
delete(product_id) async

Delete a product.

Parameters:

  • product_id (int) –

    The product ID to delete.

Example

await client.products.delete(123)

Source code in katana_public_api_client/helpers/products.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
async def delete(self, product_id: int) -> None:
    """Delete a product.

    Args:
        product_id: The product ID to delete.

    Example:
        >>> await client.products.delete(123)
    """
    await delete_product.asyncio_detailed(
        client=self._client,
        id=product_id,
    )
get(product_id) async

Get a specific product by ID.

Parameters:

  • product_id (int) –

    The product ID.

Returns:

Example

product = await client.products.get(123)

Source code in katana_public_api_client/helpers/products.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
async def get(self, product_id: int) -> KatanaProduct:
    """Get a specific product by ID.

    Args:
        product_id: The product ID.

    Returns:
        KatanaProduct domain model object.

    Example:
        >>> product = await client.products.get(123)
    """
    response = await get_product.asyncio_detailed(
        client=self._client,
        id=product_id,
    )
    # unwrap() raises on errors, so cast is safe
    attrs_product = cast(Product, unwrap(response))
    return product_to_katana(attrs_product)
list(**filters) async

List all products with optional filters.

Parameters:

  • **filters (Any, default: {} ) –

    Filtering parameters (e.g., is_sellable, is_producible, include_deleted).

Returns:

Example

products = await client.products.list(is_sellable=True, limit=100)

Source code in katana_public_api_client/helpers/products.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
async def list(self, **filters: Any) -> list[KatanaProduct]:
    """List all products with optional filters.

    Args:
        **filters: Filtering parameters (e.g., is_sellable, is_producible, include_deleted).

    Returns:
        List of KatanaProduct domain model objects.

    Example:
        >>> products = await client.products.list(is_sellable=True, limit=100)
    """
    response = await get_all_products.asyncio_detailed(
        client=self._client,
        **filters,
    )
    attrs_products = unwrap_data(response)
    return products_to_katana(attrs_products)
search(query, limit=50) async

Search products by name and category (case-insensitive substring search).

Used by: MCP tool search_products

Note: The Katana API 'name' parameter only does exact matches, so we fetch all products and perform client-side substring searching against product names and categories.

Parameters:

  • query (str) –

    Search query to match against product names (case-insensitive).

  • limit (int, default: 50 ) –

    Maximum number of results to return.

Returns:

  • list[KatanaProduct] –

    List of matching KatanaProduct domain model objects, sorted by relevance.

Example

products = await client.products.search("fox", limit=10) for product in products: ... print(f"{product.id}: {product.name}")

Source code in katana_public_api_client/helpers/products.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
async def search(self, query: str, limit: int = 50) -> builtins.list[KatanaProduct]:
    """Search products by name and category (case-insensitive substring search).

    Used by: MCP tool search_products

    Note: The Katana API 'name' parameter only does exact matches, so we
    fetch all products and perform client-side substring searching against
    product names and categories.

    Args:
        query: Search query to match against product names (case-insensitive).
        limit: Maximum number of results to return.

    Returns:
        List of matching KatanaProduct domain model objects, sorted by relevance.

    Example:
        >>> products = await client.products.search("fox", limit=10)
        >>> for product in products:
        ...     print(f"{product.id}: {product.name}")
    """
    # Fetch all products (the API doesn't support partial/fuzzy search)
    response = await get_all_products.asyncio_detailed(
        client=self._client,
        limit=1000,  # Fetch up to 1000 products for searching
    )
    attrs_products = unwrap_data(response)

    # Convert to domain models
    domain_products = products_to_katana(attrs_products)

    # Use domain model's matches_search method
    matches = [p for p in domain_products if p.matches_search(query)]

    return matches[:limit]
update(product_id, product_data) async

Update an existing product.

Parameters:

  • product_id (int) –

    The product ID to update.

  • product_data (UpdateProductRequest) –

    UpdateProductRequest model with fields to update.

Returns:

Example

from katana_public_api_client.models import UpdateProductRequest updated = await client.products.update( ... 123, UpdateProductRequest(name="Updated Name") ... )

Source code in katana_public_api_client/helpers/products.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
async def update(
    self, product_id: int, product_data: UpdateProductRequest
) -> KatanaProduct:
    """Update an existing product.

    Args:
        product_id: The product ID to update.
        product_data: UpdateProductRequest model with fields to update.

    Returns:
        Updated KatanaProduct domain model object.

    Example:
        >>> from katana_public_api_client.models import UpdateProductRequest
        >>> updated = await client.products.update(
        ...     123, UpdateProductRequest(name="Updated Name")
        ... )
    """
    response = await update_product.asyncio_detailed(
        client=self._client,
        id=product_id,
        body=product_data,
    )
    # unwrap() raises on errors, so cast is safe
    attrs_product = cast(Product, unwrap(response))
    return product_to_katana(attrs_product)

Services(client)

Bases: Base

Service catalog management.

Provides CRUD operations for services in the Katana catalog.

Example

async with KatanaClient() as client: ... # CRUD operations ... services = await client.services.list() ... service = await client.services.get(123) ... new_service = await client.services.create({"name": "Assembly"})

Source code in katana_public_api_client/helpers/base.py 
20
21
22
23
24
25
26
def __init__(self, client: KatanaClient) -> None:
    """Initialize with a client instance.

    Args:
        client: The KatanaClient instance to use for API calls.
    """
    self._client = client
Functions
create(service_data) async

Create a new service.

Parameters:

Returns:

Example

from katana_public_api_client.models import CreateServiceRequest new_service = await client.services.create( ... CreateServiceRequest(name="Assembly") ... )

Source code in katana_public_api_client/helpers/services.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
async def create(self, service_data: CreateServiceRequest) -> KatanaService:
    """Create a new service.

    Args:
        service_data: CreateServiceRequest model with service details.

    Returns:
        Created KatanaService domain model object.

    Example:
        >>> from katana_public_api_client.models import CreateServiceRequest
        >>> new_service = await client.services.create(
        ...     CreateServiceRequest(name="Assembly")
        ... )
    """
    response = await create_service.asyncio_detailed(
        client=self._client,
        body=service_data,
    )
    # unwrap() raises on errors, so cast is safe
    attrs_service = cast(Service, unwrap(response))
    return service_to_katana(attrs_service)
delete(service_id) async

Delete a service.

Parameters:

  • service_id (int) –

    The service ID to delete.

Example

await client.services.delete(123)

Source code in katana_public_api_client/helpers/services.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
async def delete(self, service_id: int) -> None:
    """Delete a service.

    Args:
        service_id: The service ID to delete.

    Example:
        >>> await client.services.delete(123)
    """
    await delete_service.asyncio_detailed(
        client=self._client,
        id=service_id,
    )
get(service_id) async

Get a specific service by ID.

Parameters:

  • service_id (int) –

    The service ID.

Returns:

Example

service = await client.services.get(123)

Source code in katana_public_api_client/helpers/services.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
async def get(self, service_id: int) -> KatanaService:
    """Get a specific service by ID.

    Args:
        service_id: The service ID.

    Returns:
        KatanaService domain model object.

    Example:
        >>> service = await client.services.get(123)
    """
    response = await get_service.asyncio_detailed(
        client=self._client,
        id=service_id,
    )
    # unwrap() raises on errors, so cast is safe
    attrs_service = cast(Service, unwrap(response))
    return service_to_katana(attrs_service)
list(**filters) async

List all services with optional filters.

Parameters:

  • **filters (Any, default: {} ) –

    Filtering parameters.

Returns:

Example

services = await client.services.list(limit=100)

Source code in katana_public_api_client/helpers/services.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
async def list(self, **filters: Any) -> list[KatanaService]:
    """List all services with optional filters.

    Args:
        **filters: Filtering parameters.

    Returns:
        List of KatanaService domain model objects.

    Example:
        >>> services = await client.services.list(limit=100)
    """
    response = await get_all_services.asyncio_detailed(
        client=self._client,
        **filters,
    )
    attrs_services = unwrap_data(response)
    return services_to_katana(attrs_services)
update(service_id, service_data) async

Update an existing service.

Parameters:

  • service_id (int) –

    The service ID to update.

  • service_data (UpdateServiceRequest) –

    UpdateServiceRequest model with fields to update.

Returns:

Example

from katana_public_api_client.models import UpdateServiceRequest updated = await client.services.update( ... 123, UpdateServiceRequest(name="QA Testing") ... )

Source code in katana_public_api_client/helpers/services.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
async def update(
    self, service_id: int, service_data: UpdateServiceRequest
) -> KatanaService:
    """Update an existing service.

    Args:
        service_id: The service ID to update.
        service_data: UpdateServiceRequest model with fields to update.

    Returns:
        Updated KatanaService domain model object.

    Example:
        >>> from katana_public_api_client.models import UpdateServiceRequest
        >>> updated = await client.services.update(
        ...     123, UpdateServiceRequest(name="QA Testing")
        ... )
    """
    response = await update_service.asyncio_detailed(
        client=self._client,
        id=service_id,
        body=service_data,
    )
    # unwrap() raises on errors, so cast is safe
    attrs_service = cast(Service, unwrap(response))
    return service_to_katana(attrs_service)

Variants(*args, **kwargs)

Bases: Base

Variant catalog management.

Provides CRUD operations for product variants in the Katana catalog. Includes caching for improved search performance.

Example

async with KatanaClient() as client: ... # CRUD operations ... variants = await client.variants.list() ... variant = await client.variants.get(123) ... new_variant = await client.variants.create({"name": "Large"}) ... ... # Fast repeated searches (uses cache) ... results1 = await client.variants.search("fox") ... results2 = await client.variants.search( ... "fork" ... ) # Instant - uses cached data

Source code in katana_public_api_client/helpers/variants.py 
115
116
117
118
def __init__(self, *args: Any, **kwargs: Any):
    """Initialize with variant cache."""
    super().__init__(*args, **kwargs)
    self._cache = VariantCache(ttl_seconds=300)  # 5 minute cache
Functions
create(variant_data) async

Create a new variant.

Note: Clears the variant cache after creation.

Parameters:

Returns:

Example

from katana_public_api_client.models import CreateVariantRequest new_variant = await client.variants.create( ... CreateVariantRequest(name="Large", product_id=123) ... )

Source code in katana_public_api_client/helpers/variants.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
async def create(self, variant_data: CreateVariantRequest) -> KatanaVariant:
    """Create a new variant.

    Note: Clears the variant cache after creation.

    Args:
        variant_data: CreateVariantRequest model with variant details.

    Returns:
        Created Variant object.

    Example:
        >>> from katana_public_api_client.models import CreateVariantRequest
        >>> new_variant = await client.variants.create(
        ...     CreateVariantRequest(name="Large", product_id=123)
        ... )
    """
    response = await create_variant.asyncio_detailed(
        client=self._client,
        body=variant_data,
    )
    # Clear cache since data changed
    self._cache.clear()
    # unwrap() raises on errors, so cast is safe
    attrs_variant = cast(Variant, unwrap(response))
    return variant_to_katana(attrs_variant)
delete(variant_id) async

Delete a variant.

Note: Clears the variant cache after deletion.

Parameters:

  • variant_id (int) –

    The variant ID to delete.

Example

await client.variants.delete(123)

Source code in katana_public_api_client/helpers/variants.py 
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
async def delete(self, variant_id: int) -> None:
    """Delete a variant.

    Note: Clears the variant cache after deletion.

    Args:
        variant_id: The variant ID to delete.

    Example:
        >>> await client.variants.delete(123)
    """
    await delete_variant.asyncio_detailed(
        client=self._client,
        id=variant_id,
    )
    # Clear cache since data changed
    self._cache.clear()
get(variant_id) async

Get a specific variant by ID.

Parameters:

  • variant_id (int) –

    The variant ID.

Returns:

Example

variant = await client.variants.get(123) print(variant.get_display_name()) print(f"Profit margin: {variant.profit_margin}%")

Source code in katana_public_api_client/helpers/variants.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
async def get(self, variant_id: int) -> KatanaVariant:
    """Get a specific variant by ID.

    Args:
        variant_id: The variant ID.

    Returns:
        KatanaVariant object.

    Example:
        >>> variant = await client.variants.get(123)
        >>> print(variant.get_display_name())
        >>> print(f"Profit margin: {variant.profit_margin}%")
    """
    response = await get_variant.asyncio_detailed(
        client=self._client,
        id=variant_id,
    )
    # unwrap() raises on errors, so cast is safe
    attrs_variant = cast(Variant, unwrap(response))
    return variant_to_katana(attrs_variant)
list(**filters) async

List all variants with optional filters.

Parameters:

  • **filters (Any, default: {} ) –

    Filtering parameters.

Returns:

Example

variants = await client.variants.list(limit=100) for v in variants: ... print(f"{v.get_display_name()}: {v.profit_margin}%")

Source code in katana_public_api_client/helpers/variants.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
async def list(self, **filters: Any) -> List[KatanaVariant]:
    """List all variants with optional filters.

    Args:
        **filters: Filtering parameters.

    Returns:
        List of KatanaVariant objects.

    Example:
        >>> variants = await client.variants.list(limit=100)
        >>> for v in variants:
        ...     print(f"{v.get_display_name()}: {v.profit_margin}%")
    """
    response = await get_all_variants.asyncio_detailed(
        client=self._client,
        **filters,
    )
    attrs_variants = unwrap_data(response)
    return variants_to_katana(attrs_variants)
search(query, limit=50) async

Search variants by SKU or parent product/material name with relevance ranking.

Used by: MCP tool search_products

Features: - Fetches all variants with parent product/material info (cached for 5 min) - Multi-token matching (all tokens must match) - Relevance-based ranking (exact matches first) - Case-insensitive substring matching

Parameters:

  • query (str) –

    Search query (e.g., "fox fork 160")

  • limit (int, default: 50 ) –

    Maximum number of results to return

Returns:

  • list[KatanaVariant] –

    List of matching Variant objects, sorted by relevance

Example
First search: fetches from API (~1-2s)

variants = await client.variants.search("fox fork", limit=10)

Subsequent searches: instant (<10ms, uses cache)

variants = await client.variants.search("fox 160", limit=10)

for variant in variants: ... print(f"{variant.sku}: {variant.product_or_material_name}")

Source code in katana_public_api_client/helpers/variants.py 
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
async def search(self, query: str, limit: int = 50) -> List[KatanaVariant]:
    """Search variants by SKU or parent product/material name with relevance ranking.

    Used by: MCP tool search_products

    Features:
    - Fetches all variants with parent product/material info (cached for 5 min)
    - Multi-token matching (all tokens must match)
    - Relevance-based ranking (exact matches first)
    - Case-insensitive substring matching

    Args:
        query: Search query (e.g., "fox fork 160")
        limit: Maximum number of results to return

    Returns:
        List of matching Variant objects, sorted by relevance

    Example:
        >>> # First search: fetches from API (~1-2s)
        >>> variants = await client.variants.search("fox fork", limit=10)
        >>>
        >>> # Subsequent searches: instant (<10ms, uses cache)
        >>> variants = await client.variants.search("fox 160", limit=10)
        >>>
        >>> for variant in variants:
        ...     print(f"{variant.sku}: {variant.product_or_material_name}")
    """
    # Tokenize query
    query_tokens = query.lower().split()
    if not query_tokens:
        return []

    # Fetch all variants (uses cache if valid)
    all_variants = await self._fetch_all_variants()

    # Score and filter variants
    scored_matches: list[tuple[KatanaVariant, int]] = []

    for variant in all_variants:
        score = self._calculate_relevance(variant, query_tokens)
        if score > 0:
            scored_matches.append((variant, score))

    # Sort by relevance (highest first)
    scored_matches.sort(key=lambda x: x[1], reverse=True)

    # Return top N variants
    return [variant for variant, _score in scored_matches[:limit]]
update(variant_id, variant_data) async

Update an existing variant.

Note: Clears the variant cache after update.

Parameters:

  • variant_id (int) –

    The variant ID to update.

  • variant_data (UpdateVariantRequest) –

    UpdateVariantRequest model with fields to update.

Returns:

Example

from katana_public_api_client.models import UpdateVariantRequest updated = await client.variants.update( ... 123, UpdateVariantRequest(name="XL") ... )

Source code in katana_public_api_client/helpers/variants.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
async def update(
    self, variant_id: int, variant_data: UpdateVariantRequest
) -> KatanaVariant:
    """Update an existing variant.

    Note: Clears the variant cache after update.

    Args:
        variant_id: The variant ID to update.
        variant_data: UpdateVariantRequest model with fields to update.

    Returns:
        Updated Variant object.

    Example:
        >>> from katana_public_api_client.models import UpdateVariantRequest
        >>> updated = await client.variants.update(
        ...     123, UpdateVariantRequest(name="XL")
        ... )
    """
    response = await update_variant.asyncio_detailed(
        client=self._client,
        id=variant_id,
        body=variant_data,
    )
    # Clear cache since data changed
    self._cache.clear()
    # unwrap() raises on errors, so cast is safe
    attrs_variant = cast(Variant, unwrap(response))
    return variant_to_katana(attrs_variant)