Skip to content

katana_public_api_client.helpers.variants

katana_public_api_client.helpers.variants

Variant catalog operations.

Classes

VariantCache(ttl_seconds=300)

Cache for variant data with multiple access patterns.

Provides: - List of all variants (for iteration/filtering) - Dict by variant ID (O(1) lookup by ID) - Dict by SKU (O(1) lookup by SKU) - TTL-based invalidation

Note: Cache stores Pydantic KatanaVariant models (not attrs models).

Parameters:

  • ttl_seconds (int, default: 300 ) –

    Time-to-live in seconds. Default 5 minutes.

Source code in katana_public_api_client/helpers/variants.py 
45
46
47
48
49
50
51
52
53
54
55
def __init__(self, ttl_seconds: int = 300):
    """Initialize cache with TTL.

    Args:
        ttl_seconds: Time-to-live in seconds. Default 5 minutes.
    """
    self.ttl_seconds = ttl_seconds
    self.variants: List[KatanaVariant] = []
    self.by_id: dict[int, KatanaVariant] = {}
    self.by_sku: dict[str, KatanaVariant] = {}
    self.cached_at: float = 0
Functions
clear()

Clear all cached data.

Source code in katana_public_api_client/helpers/variants.py 
87
88
89
90
91
92
def clear(self) -> None:
    """Clear all cached data."""
    self.variants = []
    self.by_id = {}
    self.by_sku = {}
    self.cached_at = 0
is_valid()

Check if cache is still valid.

Source code in katana_public_api_client/helpers/variants.py 
57
58
59
60
61
62
def is_valid(self) -> bool:
    """Check if cache is still valid."""
    if not self.variants:
        return False
    age = time.monotonic() - self.cached_at
    return age < self.ttl_seconds
update(variants)

Update cache with new variant list.

Parameters:

  • variants (list[KatanaVariant]) –

    List of domain variants to cache

Source code in katana_public_api_client/helpers/variants.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def update(self, variants: List[KatanaVariant]) -> None:
    """Update cache with new variant list.

    Args:
        variants: List of domain variants to cache
    """
    self.variants = variants
    self.cached_at = time.monotonic()

    # Build lookup dictionaries
    self.by_id = {v.id: v for v in variants}

    # Build SKU lookup with duplicate detection
    self.by_sku = {}
    for v in variants:
        if v.sku:
            if v.sku in self.by_sku:
                logger.warning(
                    f"Duplicate SKU detected: {v.sku} "
                    f"(variant IDs: {self.by_sku[v.sku].id} and {v.id})"
                )
            self.by_sku[v.sku] = v

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)

Functions