ebay_client/ebay_client/inventory/client.py

from __future__ import annotations

from ebay_client.core.http.transport import ApiTransport
from ebay_client.generated.inventory.models import (
    BaseResponse,
    EbayOfferDetailsWithId,
    EbayOfferDetailsWithKeys,
    FeesSummaryResponse,
    InventoryItem,
    InventoryItemGroup,
    InventoryLocation,
    InventoryLocationFull,
    InventoryLocationResponse,
    InventoryItemWithSkuLocaleGroupid,
    InventoryItems,
    LocationResponse,
    OfferKeysWithId,
    OfferResponse,
    OfferResponseWithListingId,
    Offers,
    PublishByInventoryItemGroupRequest,
    PublishResponse,
    WithdrawByInventoryItemGroupRequest,
    WithdrawResponse,
)

INVENTORY_SCOPE = "https://api.ebay.com/oauth/api_scope/sell.inventory"
INVENTORY_READ_SCOPE = "https://api.ebay.com/oauth/api_scope/sell.inventory.readonly"
INVENTORY_READ_SCOPE_OPTIONS = [[INVENTORY_READ_SCOPE], [INVENTORY_SCOPE]]


class InventoryClient:
    def __init__(self, transport: ApiTransport) -> None:
        self.transport = transport

    def _write_headers(self, *, content_language: str | None = None) -> dict[str, str]:
        headers = {"Content-Type": "application/json"}
        if content_language is not None:
            headers["Content-Language"] = content_language
        return headers

    def get_inventory_item(self, sku: str) -> InventoryItemWithSkuLocaleGroupid:
        return self.transport.request_model(
            InventoryItemWithSkuLocaleGroupid,
            "GET",
            f"/sell/inventory/v1/inventory_item/{sku}",
            scope_options=INVENTORY_READ_SCOPE_OPTIONS,
        )

    def create_or_replace_inventory_item(
        self,
        sku: str,
        payload: InventoryItem,
        *,
        content_language: str,
    ) -> BaseResponse:
        return self.transport.request_model(
            BaseResponse,
            "PUT",
            f"/sell/inventory/v1/inventory_item/{sku}",
            scopes=[INVENTORY_SCOPE],
            headers=self._write_headers(content_language=content_language),
            json_body=payload.model_dump(by_alias=True, exclude_none=True),
        )

    def delete_inventory_item(self, sku: str) -> None:
        self.transport.request_json(
            "DELETE",
            f"/sell/inventory/v1/inventory_item/{sku}",
            scopes=[INVENTORY_SCOPE],
        )

    def get_inventory_items(self, *, limit: int | None = None, offset: int | None = None) -> InventoryItems:
        return self.transport.request_model(
            InventoryItems,
            "GET",
            "/sell/inventory/v1/inventory_item",
            scope_options=INVENTORY_READ_SCOPE_OPTIONS,
            params={"limit": limit, "offset": offset},
        )

    def get_inventory_item_group(self, inventory_item_group_key: str) -> InventoryItemGroup:
        return self.transport.request_model(
            InventoryItemGroup,
            "GET",
            f"/sell/inventory/v1/inventory_item_group/{inventory_item_group_key}",
            scope_options=INVENTORY_READ_SCOPE_OPTIONS,
        )

    def create_or_replace_inventory_item_group(
        self,
        inventory_item_group_key: str,
        payload: InventoryItemGroup,
        *,
        content_language: str,
    ) -> BaseResponse:
        return self.transport.request_model(
            BaseResponse,
            "PUT",
            f"/sell/inventory/v1/inventory_item_group/{inventory_item_group_key}",
            scopes=[INVENTORY_SCOPE],
            headers=self._write_headers(content_language=content_language),
            json_body=payload.model_dump(by_alias=True, exclude_none=True),
        )

    def delete_inventory_item_group(self, inventory_item_group_key: str) -> None:
        self.transport.request_json(
            "DELETE",
            f"/sell/inventory/v1/inventory_item_group/{inventory_item_group_key}",
            scopes=[INVENTORY_SCOPE],
        )

    def get_offer(self, offer_id: str) -> OfferResponseWithListingId:
        return self.transport.request_model(
            OfferResponseWithListingId,
            "GET",
            f"/sell/inventory/v1/offer/{offer_id}",
            scope_options=INVENTORY_READ_SCOPE_OPTIONS,
        )

    def create_offer(
        self,
        payload: EbayOfferDetailsWithKeys,
        *,
        content_language: str,
    ) -> OfferResponse:
        return self.transport.request_model(
            OfferResponse,
            "POST",
            "/sell/inventory/v1/offer",
            scopes=[INVENTORY_SCOPE],
            headers=self._write_headers(content_language=content_language),
            json_body=payload.model_dump(by_alias=True, exclude_none=True),
        )

    def update_offer(
        self,
        offer_id: str,
        payload: EbayOfferDetailsWithId,
        *,
        content_language: str,
    ) -> OfferResponse:
        return self.transport.request_model(
            OfferResponse,
            "PUT",
            f"/sell/inventory/v1/offer/{offer_id}",
            scopes=[INVENTORY_SCOPE],
            headers=self._write_headers(content_language=content_language),
            json_body=payload.model_dump(by_alias=True, exclude_none=True),
        )

    def delete_offer(self, offer_id: str) -> None:
        self.transport.request_json(
            "DELETE",
            f"/sell/inventory/v1/offer/{offer_id}",
            scopes=[INVENTORY_SCOPE],
        )

    def get_offers(self, *, limit: int | None = None, offset: int | None = None, sku: str | None = None) -> Offers:
        return self.transport.request_model(
            Offers,
            "GET",
            "/sell/inventory/v1/offer",
            scope_options=INVENTORY_READ_SCOPE_OPTIONS,
            params={"limit": limit, "offset": offset, "sku": sku},
        )

    def get_listing_fees(self, payload: OfferKeysWithId | None = None) -> FeesSummaryResponse:
        return self.transport.request_model(
            FeesSummaryResponse,
            "POST",
            "/sell/inventory/v1/offer/get_listing_fees",
            scope_options=INVENTORY_READ_SCOPE_OPTIONS,
            headers={"Content-Type": "application/json"},
            json_body=None if payload is None else payload.model_dump(by_alias=True, exclude_none=True),
        )

    def publish_offer(self, offer_id: str) -> PublishResponse:
        return self.transport.request_model(
            PublishResponse,
            "POST",
            f"/sell/inventory/v1/offer/{offer_id}/publish",
            scopes=[INVENTORY_SCOPE],
        )

    def publish_offer_by_inventory_item_group(
        self,
        payload: PublishByInventoryItemGroupRequest,
    ) -> PublishResponse:
        return self.transport.request_model(
            PublishResponse,
            "POST",
            "/sell/inventory/v1/offer/publish_by_inventory_item_group",
            scopes=[INVENTORY_SCOPE],
            headers={"Content-Type": "application/json"},
            json_body=payload.model_dump(by_alias=True, exclude_none=True),
        )

    def withdraw_offer(self, offer_id: str) -> WithdrawResponse:
        return self.transport.request_model(
            WithdrawResponse,
            "POST",
            f"/sell/inventory/v1/offer/{offer_id}/withdraw",
            scopes=[INVENTORY_SCOPE],
        )

    def withdraw_offer_by_inventory_item_group(
        self,
        payload: WithdrawByInventoryItemGroupRequest,
    ) -> None:
        self.transport.request_json(
            "POST",
            "/sell/inventory/v1/offer/withdraw_by_inventory_item_group",
            scopes=[INVENTORY_SCOPE],
            headers={"Content-Type": "application/json"},
            json_body=payload.model_dump(by_alias=True, exclude_none=True),
        )

    def get_inventory_location(self, merchant_location_key: str) -> InventoryLocationResponse:
        return self.transport.request_model(
            InventoryLocationResponse,
            "GET",
            f"/sell/inventory/v1/location/{merchant_location_key}",
            scope_options=INVENTORY_READ_SCOPE_OPTIONS,
        )

    def create_inventory_location(
        self,
        merchant_location_key: str,
        payload: InventoryLocationFull,
    ) -> None:
        self.transport.request_json(
            "POST",
            f"/sell/inventory/v1/location/{merchant_location_key}",
            scopes=[INVENTORY_SCOPE],
            headers={"Content-Type": "application/json"},
            json_body=payload.model_dump(by_alias=True, exclude_none=True),
        )

    def delete_inventory_location(self, merchant_location_key: str) -> None:
        self.transport.request_json(
            "DELETE",
            f"/sell/inventory/v1/location/{merchant_location_key}",
            scopes=[INVENTORY_SCOPE],
        )

    def disable_inventory_location(self, merchant_location_key: str) -> None:
        self.transport.request_json(
            "POST",
            f"/sell/inventory/v1/location/{merchant_location_key}/disable",
            scopes=[INVENTORY_SCOPE],
        )

    def enable_inventory_location(self, merchant_location_key: str) -> None:
        self.transport.request_json(
            "POST",
            f"/sell/inventory/v1/location/{merchant_location_key}/enable",
            scopes=[INVENTORY_SCOPE],
        )

    def get_inventory_locations(
        self,
        *,
        limit: int | None = None,
        offset: int | None = None,
    ) -> LocationResponse:
        return self.transport.request_model(
            LocationResponse,
            "GET",
            "/sell/inventory/v1/location",
            scope_options=INVENTORY_READ_SCOPE_OPTIONS,
            params={"limit": limit, "offset": offset},
        )

    def update_inventory_location(
        self,
        merchant_location_key: str,
        payload: InventoryLocation,
    ) -> None:
        self.transport.request_json(
            "POST",
            f"/sell/inventory/v1/location/{merchant_location_key}/update_location_details",
            scopes=[INVENTORY_SCOPE],
            headers={"Content-Type": "application/json"},
            json_body=payload.model_dump(by_alias=True, exclude_none=True),
        )