From 1f06ec9e448cb61459616aed60e68d0ddc2724ae Mon Sep 17 00:00:00 2001 From: claudi Date: Tue, 7 Apr 2026 10:59:48 +0200 Subject: [PATCH] Add inventory management methods and tests for offers and locations in InventoryClient --- ebay_client/inventory/client.py | 234 ++++++++++++++++++++++++++++++++ tests/test_public_wrappers.py | 228 ++++++++++++++++++++++++++++++- 2 files changed, 461 insertions(+), 1 deletion(-) diff --git a/ebay_client/inventory/client.py b/ebay_client/inventory/client.py index 4de7ad1..7a51b93 100644 --- a/ebay_client/inventory/client.py +++ b/ebay_client/inventory/client.py @@ -2,10 +2,26 @@ 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" @@ -17,6 +33,12 @@ 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, @@ -25,6 +47,29 @@ class InventoryClient: 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, @@ -34,6 +79,37 @@ class InventoryClient: 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, @@ -42,6 +118,44 @@ class InventoryClient: 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, @@ -50,3 +164,123 @@ class InventoryClient: 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), + ) diff --git a/tests/test_public_wrappers.py b/tests/test_public_wrappers.py index 6aec0a0..931d5c4 100644 --- a/tests/test_public_wrappers.py +++ b/tests/test_public_wrappers.py @@ -27,7 +27,28 @@ from ebay_client.generated.feed.models import ( UserScheduleResponse, ) from ebay_client.generated.fulfillment.models import Order -from ebay_client.generated.inventory.models import InventoryItemWithSkuLocaleGroupid +from ebay_client.generated.inventory.models import ( + Address, + BaseResponse, + EbayOfferDetailsWithId, + EbayOfferDetailsWithKeys, + FeesSummaryResponse, + InventoryItem, + InventoryItemGroup, + InventoryItemWithSkuLocaleGroupid, + InventoryLocation, + InventoryLocationFull, + InventoryLocationResponse, + LocationDetails, + LocationResponse, + OfferKeyWithId, + OfferKeysWithId, + OfferResponse, + PublishByInventoryItemGroupRequest, + PublishResponse, + WithdrawByInventoryItemGroupRequest, + WithdrawResponse, +) from ebay_client.generated.media.models import ( CreateDocumentFromUrlRequest, CreateDocumentRequest, @@ -277,6 +298,211 @@ def test_inventory_wrapper_returns_inventory_item_model(httpx_mock: HTTPXMock) - assert result.sku == "SKU-1" +def test_inventory_wrapper_supports_core_item_offer_and_group_workflows(httpx_mock: HTTPXMock) -> None: + httpx_mock.add_response( + method="PUT", + url="https://api.ebay.com/sell/inventory/v1/inventory_item/SKU-1", + json={"warnings": []}, + ) + httpx_mock.add_response( + method="DELETE", + url="https://api.ebay.com/sell/inventory/v1/inventory_item/SKU-1", + status_code=204, + ) + httpx_mock.add_response( + method="PUT", + url="https://api.ebay.com/sell/inventory/v1/inventory_item_group/GROUP-1", + json={"warnings": []}, + ) + httpx_mock.add_response( + method="GET", + url="https://api.ebay.com/sell/inventory/v1/inventory_item_group/GROUP-1", + json={"inventoryItemGroupKey": "GROUP-1", "variantSKUs": ["SKU-1"]}, + ) + httpx_mock.add_response( + method="POST", + url="https://api.ebay.com/sell/inventory/v1/offer", + json={"offerId": "OFFER-1"}, + status_code=201, + ) + httpx_mock.add_response( + method="PUT", + url="https://api.ebay.com/sell/inventory/v1/offer/OFFER-1", + json={"warnings": []}, + ) + httpx_mock.add_response( + method="POST", + url="https://api.ebay.com/sell/inventory/v1/offer/OFFER-1/publish", + json={"offerId": "OFFER-1", "listingId": "ITEM-1"}, + ) + httpx_mock.add_response( + method="POST", + url="https://api.ebay.com/sell/inventory/v1/offer/OFFER-1/withdraw", + json={"offerId": "OFFER-1", "listingId": "ITEM-1"}, + ) + httpx_mock.add_response( + method="POST", + url="https://api.ebay.com/sell/inventory/v1/offer/publish_by_inventory_item_group", + json={"listingId": "ITEM-2"}, + ) + httpx_mock.add_response( + method="POST", + url="https://api.ebay.com/sell/inventory/v1/offer/withdraw_by_inventory_item_group", + status_code=204, + ) + httpx_mock.add_response( + method="POST", + url="https://api.ebay.com/sell/inventory/v1/offer/get_listing_fees", + json={"feeSummaries": []}, + ) + httpx_mock.add_response( + method="DELETE", + url="https://api.ebay.com/sell/inventory/v1/offer/OFFER-1", + status_code=204, + ) + httpx_mock.add_response( + method="DELETE", + url="https://api.ebay.com/sell/inventory/v1/inventory_item_group/GROUP-1", + status_code=204, + ) + + client = InventoryClient(build_transport()) + item_result = client.create_or_replace_inventory_item( + "SKU-1", + InventoryItem(condition="NEW"), + content_language="en-US", + ) + client.delete_inventory_item("SKU-1") + group_result = client.create_or_replace_inventory_item_group( + "GROUP-1", + InventoryItemGroup(variantSKUs=["SKU-1"]), + content_language="en-US", + ) + group = client.get_inventory_item_group("GROUP-1") + offer = client.create_offer( + EbayOfferDetailsWithKeys(sku="SKU-1", marketplaceId="EBAY_US", format="FIXED_PRICE"), + content_language="en-US", + ) + updated_offer = client.update_offer( + "OFFER-1", + EbayOfferDetailsWithId(availableQuantity=2), + content_language="en-US", + ) + published_offer = client.publish_offer("OFFER-1") + withdrawn_offer = client.withdraw_offer("OFFER-1") + group_publish = client.publish_offer_by_inventory_item_group( + PublishByInventoryItemGroupRequest(inventoryItemGroupKey="GROUP-1", marketplaceId="EBAY_US") + ) + client.withdraw_offer_by_inventory_item_group( + WithdrawByInventoryItemGroupRequest(inventoryItemGroupKey="GROUP-1", marketplaceId="EBAY_US") + ) + fees = client.get_listing_fees(OfferKeysWithId(offers=[OfferKeyWithId(offerId="OFFER-1")])) + client.delete_offer("OFFER-1") + client.delete_inventory_item_group("GROUP-1") + + assert isinstance(item_result, BaseResponse) + assert isinstance(group_result, BaseResponse) + assert isinstance(group, InventoryItemGroup) + assert group.inventoryItemGroupKey == "GROUP-1" + assert isinstance(offer, OfferResponse) + assert offer.offerId == "OFFER-1" + assert isinstance(updated_offer, OfferResponse) + assert isinstance(published_offer, PublishResponse) + assert published_offer.listingId == "ITEM-1" + assert isinstance(withdrawn_offer, WithdrawResponse) + assert withdrawn_offer.listingId == "ITEM-1" + assert isinstance(group_publish, PublishResponse) + assert group_publish.listingId == "ITEM-2" + assert isinstance(fees, FeesSummaryResponse) + + item_request = httpx_mock.get_requests()[0] + assert item_request.headers["Content-Language"] == "en-US" + item_body = json.loads(item_request.content.decode("utf-8")) + assert item_body["condition"] == "NEW" + + offer_request = httpx_mock.get_requests()[4] + assert offer_request.headers["Content-Language"] == "en-US" + offer_body = json.loads(offer_request.content.decode("utf-8")) + assert offer_body["sku"] == "SKU-1" + assert offer_body["marketplaceId"] == "EBAY_US" + + update_request = httpx_mock.get_requests()[5] + update_body = json.loads(update_request.content.decode("utf-8")) + assert update_body["availableQuantity"] == 2 + + group_publish_request = httpx_mock.get_requests()[8] + assert group_publish_request.headers["Content-Type"] == "application/json" + group_publish_body = json.loads(group_publish_request.content.decode("utf-8")) + assert group_publish_body["inventoryItemGroupKey"] == "GROUP-1" + + fees_request = httpx_mock.get_requests()[10] + fees_body = json.loads(fees_request.content.decode("utf-8")) + assert fees_body["offers"] == [{"offerId": "OFFER-1"}] + + +def test_inventory_wrapper_supports_location_management(httpx_mock: HTTPXMock) -> None: + httpx_mock.add_response( + method="POST", + url="https://api.ebay.com/sell/inventory/v1/location/LOC-1", + status_code=204, + ) + httpx_mock.add_response( + method="GET", + url="https://api.ebay.com/sell/inventory/v1/location/LOC-1", + json={"merchantLocationKey": "LOC-1", "name": "Main warehouse"}, + ) + httpx_mock.add_response( + method="POST", + url="https://api.ebay.com/sell/inventory/v1/location/LOC-1/update_location_details", + status_code=204, + ) + httpx_mock.add_response( + method="POST", + url="https://api.ebay.com/sell/inventory/v1/location/LOC-1/disable", + json={}, + ) + httpx_mock.add_response( + method="POST", + url="https://api.ebay.com/sell/inventory/v1/location/LOC-1/enable", + json={}, + ) + httpx_mock.add_response( + method="GET", + url="https://api.ebay.com/sell/inventory/v1/location?limit=25&offset=0", + json={"locations": [{"merchantLocationKey": "LOC-1"}], "total": 1}, + ) + httpx_mock.add_response( + method="DELETE", + url="https://api.ebay.com/sell/inventory/v1/location/LOC-1", + status_code=204, + ) + + client = InventoryClient(build_transport()) + client.create_inventory_location( + "LOC-1", + InventoryLocationFull(location=LocationDetails(address=Address(country="US", postalCode="10001"))), + ) + location = client.get_inventory_location("LOC-1") + client.update_inventory_location("LOC-1", InventoryLocation(name="Main warehouse")) + client.disable_inventory_location("LOC-1") + client.enable_inventory_location("LOC-1") + locations = client.get_inventory_locations(limit=25, offset=0) + client.delete_inventory_location("LOC-1") + + assert isinstance(location, InventoryLocationResponse) + assert location.merchantLocationKey == "LOC-1" + assert isinstance(locations, LocationResponse) + assert locations.total == 1 + + create_request = httpx_mock.get_requests()[0] + create_body = json.loads(create_request.content.decode("utf-8")) + assert create_body["location"]["address"]["country"] == "US" + + update_request = httpx_mock.get_requests()[2] + update_body = json.loads(update_request.content.decode("utf-8")) + assert update_body["name"] == "Main warehouse" + + def test_fulfillment_wrapper_returns_order_model(httpx_mock: HTTPXMock) -> None: httpx_mock.add_response( method="GET",