products.py•2.39 kB
"""Product endpoints for the OCC client."""
from __future__ import annotations
from typing import Any
from app.models import JsonObject
from app.settings import settings
from app.utils import clean_params, coalesce_fields
from .base import SupportsOccHttp
class ProductsMixin:
async def get_product(
self: SupportsOccHttp,
base_site_id: str,
product_code: str,
*,
fields: str | None = None,
) -> JsonObject:
"""Retrieve full product details for the given code."""
params = {"fields": coalesce_fields(fields, settings.occ_default_fields)}
path = f"/{base_site_id}/products/{product_code}"
return await self.get_json(path, params=params)
async def get_product_stock(
self: SupportsOccHttp,
base_site_id: str,
product_code: str,
*,
location: str | None = None,
latitude: float | None = None,
longitude: float | None = None,
) -> JsonObject:
"""Retrieve detailed stock information for a product."""
params = clean_params(
{
"location": location,
"latitude": latitude,
"longitude": longitude,
}
)
path = f"/{base_site_id}/products/{product_code}/stock"
return await self.get_json(path, params=params)
async def head_product_stock(
self: SupportsOccHttp,
base_site_id: str,
product_code: str,
*,
location: str | None = None,
latitude: float | None = None,
longitude: float | None = None,
) -> dict[str, Any]:
"""Fetch stock metadata (via HEAD) for a product."""
params = clean_params(
{
"location": location,
"latitude": latitude,
"longitude": longitude,
}
)
path = f"/{base_site_id}/products/{product_code}/stock"
metadata = await self.head_metadata(path, params=params)
headers = {k.lower(): v for k, v in metadata.get("headers", {}).items()}
total_count = headers.get("x-total-count")
try:
total_stock_levels = int(total_count) if total_count is not None else None
except ValueError:
total_stock_levels = None
metadata["total_stock_levels"] = total_stock_levels
return metadata