carts.py•5.37 kB
"""Cart and cart entry 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 CartsMixin:
async def get_carts(
self: SupportsOccHttp,
base_site_id: str,
user_id: str = "anonymous",
*,
current_page: int | None = None,
page_size: int | None = None,
sort: str | None = None,
saved_carts_only: bool | None = None,
fields: str | None = None,
) -> JsonObject:
params = clean_params(
{
"fields": coalesce_fields(fields, settings.occ_default_fields),
"currentPage": current_page,
"pageSize": page_size,
"sort": sort,
"savedCartsOnly": saved_carts_only,
}
)
path = f"/{base_site_id}/users/{user_id}/carts"
return await self.get_json(path, params=params)
async def create_cart(
self: SupportsOccHttp,
base_site_id: str,
user_id: str = "anonymous",
*,
fields: str | None = None,
old_cart_id: str | None = None,
to_merge_cart_guid: str | None = None,
to_merge_cart_id: str | None = None,
payload: dict[str, Any] | None = None,
) -> JsonObject:
params = clean_params(
{
"fields": coalesce_fields(fields, settings.occ_default_fields),
"oldCartId": old_cart_id,
"toMergeCartGuid": to_merge_cart_guid,
"toMergeCartId": to_merge_cart_id,
}
)
path = f"/{base_site_id}/users/{user_id}/carts"
return await self.post_json(path, params=params, json_data=payload or {})
async def get_cart(
self: SupportsOccHttp,
base_site_id: str,
user_id: str,
cart_id: str,
*,
fields: str | None = None,
) -> JsonObject:
params = {"fields": coalesce_fields(fields, settings.occ_default_fields)}
path = f"/{base_site_id}/users/{user_id}/carts/{cart_id}"
return await self.get_json(path, params=params)
async def delete_cart(
self: SupportsOccHttp,
base_site_id: str,
user_id: str,
cart_id: str,
) -> dict[str, Any]:
path = f"/{base_site_id}/users/{user_id}/carts/{cart_id}"
return await self.delete(path, params={}, expect_body=False)
async def get_cart_entries(
self: SupportsOccHttp,
base_site_id: str,
user_id: str,
cart_id: str,
*,
fields: str | None = None,
) -> JsonObject:
params = {"fields": coalesce_fields(fields, settings.occ_default_fields)}
path = f"/{base_site_id}/users/{user_id}/carts/{cart_id}/entries"
return await self.get_json(path, params=params)
async def add_cart_entry(
self: SupportsOccHttp,
base_site_id: str,
user_id: str,
cart_id: str,
*,
fields: str | None = None,
payload: dict[str, Any],
) -> JsonObject:
params = {"fields": coalesce_fields(fields, settings.occ_default_fields)}
path = f"/{base_site_id}/users/{user_id}/carts/{cart_id}/entries"
return await self.post_json(path, params=params, json_data=payload or {})
async def get_cart_entry(
self: SupportsOccHttp,
base_site_id: str,
user_id: str,
cart_id: str,
entry_number: str,
*,
fields: str | None = None,
) -> JsonObject:
params = {"fields": coalesce_fields(fields, settings.occ_default_fields)}
path = (
f"/{base_site_id}/users/{user_id}/carts/{cart_id}/entries/{entry_number}"
)
return await self.get_json(path, params=params)
async def update_cart_entry(
self: SupportsOccHttp,
base_site_id: str,
user_id: str,
cart_id: str,
entry_number: str,
*,
fields: str | None = None,
payload: dict[str, Any],
) -> JsonObject:
params = {"fields": coalesce_fields(fields, settings.occ_default_fields)}
path = (
f"/{base_site_id}/users/{user_id}/carts/{cart_id}/entries/{entry_number}"
)
return await self.put_json(path, params=params, json_data=payload or {})
async def delete_cart_entry(
self: SupportsOccHttp,
base_site_id: str,
user_id: str,
cart_id: str,
entry_number: str,
) -> dict[str, Any]:
path = (
f"/{base_site_id}/users/{user_id}/carts/{cart_id}/entries/{entry_number}"
)
return await self.delete(path, params={}, expect_body=False)
async def patch_cart_entry(
self: SupportsOccHttp,
base_site_id: str,
user_id: str,
cart_id: str,
entry_number: str,
*,
fields: str | None = None,
payload: dict[str, Any],
) -> JsonObject:
params = {"fields": coalesce_fields(fields, settings.occ_default_fields)}
path = (
f"/{base_site_id}/users/{user_id}/carts/{cart_id}/entries/{entry_number}"
)
return await self.patch_json(path, params=params, json_data=payload or {})