430 lines
15 KiB
Python
430 lines
15 KiB
Python
from __future__ import annotations
|
|
|
|
from ebay_client.core.http.transport import ApiTransport
|
|
from ebay_client.generated.inventory.models import (
|
|
BaseResponse,
|
|
BulkEbayOfferDetailsWithKeys,
|
|
BulkGetInventoryItem,
|
|
BulkGetInventoryItemResponse,
|
|
BulkInventoryItem,
|
|
BulkInventoryItemResponse,
|
|
BulkMigrateListing,
|
|
BulkMigrateListingResponse,
|
|
BulkOffer,
|
|
BulkOfferResponse,
|
|
BulkPriceQuantity,
|
|
BulkPriceQuantityResponse,
|
|
BulkPublishResponse,
|
|
Compatibility,
|
|
EbayOfferDetailsWithId,
|
|
EbayOfferDetailsWithKeys,
|
|
FeesSummaryResponse,
|
|
InventoryItem,
|
|
InventoryItemGroup,
|
|
InventoryLocation,
|
|
InventoryLocationFull,
|
|
InventoryLocationResponse,
|
|
InventoryItemWithSkuLocaleGroupid,
|
|
InventoryItems,
|
|
LocationMapping,
|
|
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 bulk_create_or_replace_inventory_item(
|
|
self,
|
|
payload: BulkInventoryItem,
|
|
*,
|
|
content_language: str,
|
|
) -> BulkInventoryItemResponse:
|
|
return self.transport.request_model(
|
|
BulkInventoryItemResponse,
|
|
"POST",
|
|
"/sell/inventory/v1/bulk_create_or_replace_inventory_item",
|
|
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 bulk_get_inventory_item(self, payload: BulkGetInventoryItem) -> BulkGetInventoryItemResponse:
|
|
return self.transport.request_model(
|
|
BulkGetInventoryItemResponse,
|
|
"POST",
|
|
"/sell/inventory/v1/bulk_get_inventory_item",
|
|
scope_options=INVENTORY_READ_SCOPE_OPTIONS,
|
|
headers={"Content-Type": "application/json"},
|
|
json_body=payload.model_dump(by_alias=True, exclude_none=True),
|
|
)
|
|
|
|
def bulk_update_price_quantity(self, payload: BulkPriceQuantity) -> BulkPriceQuantityResponse:
|
|
return self.transport.request_model(
|
|
BulkPriceQuantityResponse,
|
|
"POST",
|
|
"/sell/inventory/v1/bulk_update_price_quantity",
|
|
scopes=[INVENTORY_SCOPE],
|
|
headers={"Content-Type": "application/json"},
|
|
json_body=payload.model_dump(by_alias=True, exclude_none=True),
|
|
)
|
|
|
|
def get_product_compatibility(self, sku: str) -> Compatibility:
|
|
return self.transport.request_model(
|
|
Compatibility,
|
|
"GET",
|
|
f"/sell/inventory/v1/inventory_item/{sku}/product_compatibility",
|
|
scope_options=INVENTORY_READ_SCOPE_OPTIONS,
|
|
)
|
|
|
|
def create_or_replace_product_compatibility(
|
|
self,
|
|
sku: str,
|
|
payload: Compatibility,
|
|
*,
|
|
content_language: str,
|
|
) -> BaseResponse:
|
|
return self.transport.request_model(
|
|
BaseResponse,
|
|
"PUT",
|
|
f"/sell/inventory/v1/inventory_item/{sku}/product_compatibility",
|
|
scopes=[INVENTORY_SCOPE],
|
|
headers=self._write_headers(content_language=content_language),
|
|
json_body=payload.model_dump(by_alias=True, exclude_none=True),
|
|
)
|
|
|
|
def delete_product_compatibility(self, sku: str) -> None:
|
|
self.transport.request_json(
|
|
"DELETE",
|
|
f"/sell/inventory/v1/inventory_item/{sku}/product_compatibility",
|
|
scopes=[INVENTORY_SCOPE],
|
|
)
|
|
|
|
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 bulk_create_offer(
|
|
self,
|
|
payload: BulkEbayOfferDetailsWithKeys,
|
|
*,
|
|
content_language: str,
|
|
) -> BulkOfferResponse:
|
|
return self.transport.request_model(
|
|
BulkOfferResponse,
|
|
"POST",
|
|
"/sell/inventory/v1/bulk_create_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 bulk_publish_offer(self, payload: BulkOffer) -> BulkPublishResponse:
|
|
return self.transport.request_model(
|
|
BulkPublishResponse,
|
|
"POST",
|
|
"/sell/inventory/v1/bulk_publish_offer",
|
|
scopes=[INVENTORY_SCOPE],
|
|
headers={"Content-Type": "application/json"},
|
|
json_body=payload.model_dump(by_alias=True, exclude_none=True),
|
|
)
|
|
|
|
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),
|
|
)
|
|
|
|
def bulk_migrate_listing(self, payload: BulkMigrateListing) -> BulkMigrateListingResponse:
|
|
return self.transport.request_model(
|
|
BulkMigrateListingResponse,
|
|
"POST",
|
|
"/sell/inventory/v1/bulk_migrate_listing",
|
|
scopes=[INVENTORY_SCOPE],
|
|
headers={"Content-Type": "application/json"},
|
|
json_body=payload.model_dump(by_alias=True, exclude_none=True),
|
|
)
|
|
|
|
def get_sku_location_mapping(self, listing_id: str, sku: str) -> LocationMapping:
|
|
return self.transport.request_model(
|
|
LocationMapping,
|
|
"GET",
|
|
f"/sell/inventory/v1/listing/{listing_id}/sku/{sku}/locations",
|
|
scope_options=INVENTORY_READ_SCOPE_OPTIONS,
|
|
)
|
|
|
|
def create_or_replace_sku_location_mapping(
|
|
self,
|
|
listing_id: str,
|
|
sku: str,
|
|
payload: LocationMapping,
|
|
) -> None:
|
|
self.transport.request_json(
|
|
"PUT",
|
|
f"/sell/inventory/v1/listing/{listing_id}/sku/{sku}/locations",
|
|
scopes=[INVENTORY_SCOPE],
|
|
headers={"Content-Type": "application/json"},
|
|
json_body=payload.model_dump(by_alias=True, exclude_none=True),
|
|
)
|
|
|
|
def delete_sku_location_mapping(self, listing_id: str, sku: str) -> None:
|
|
self.transport.request_json(
|
|
"DELETE",
|
|
f"/sell/inventory/v1/listing/{listing_id}/sku/{sku}/locations",
|
|
scopes=[INVENTORY_SCOPE],
|
|
)
|