From 65d31a3f2f013141251e6fc06795cb359941a94a Mon Sep 17 00:00:00 2001 From: claudi Date: Tue, 7 Apr 2026 11:11:19 +0200 Subject: [PATCH] Add bulk inventory and offer management methods to InventoryClient; enhance tests for new endpoints --- ebay_client/inventory/client.py | 144 ++++++++++++++++++++++++++ tests/test_public_wrappers.py | 177 ++++++++++++++++++++++++++++++++ 2 files changed, 321 insertions(+) diff --git a/ebay_client/inventory/client.py b/ebay_client/inventory/client.py index 7a51b93..b3dfc8f 100644 --- a/ebay_client/inventory/client.py +++ b/ebay_client/inventory/client.py @@ -3,6 +3,19 @@ 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, @@ -13,6 +26,7 @@ from ebay_client.generated.inventory.models import ( InventoryLocationResponse, InventoryItemWithSkuLocaleGroupid, InventoryItems, + LocationMapping, LocationResponse, OfferKeysWithId, OfferResponse, @@ -63,6 +77,21 @@ class InventoryClient: 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", @@ -79,6 +108,57 @@ class InventoryClient: 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, @@ -133,6 +213,21 @@ class InventoryClient: 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, @@ -183,6 +278,16 @@ class InventoryClient: 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, @@ -284,3 +389,42 @@ class InventoryClient: 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], + ) diff --git a/tests/test_public_wrappers.py b/tests/test_public_wrappers.py index e8897b4..e12b54f 100644 --- a/tests/test_public_wrappers.py +++ b/tests/test_public_wrappers.py @@ -46,20 +46,40 @@ from ebay_client.generated.fulfillment.models import ( from ebay_client.generated.inventory.models import ( Address, BaseResponse, + BulkEbayOfferDetailsWithKeys, + BulkGetInventoryItem, + BulkGetInventoryItemResponse, + BulkInventoryItem, + BulkInventoryItemResponse, + BulkMigrateListing, + BulkMigrateListingResponse, + BulkOffer, + BulkOfferResponse, + BulkPriceQuantity, + BulkPriceQuantityResponse, + BulkPublishResponse, + Compatibility, EbayOfferDetailsWithId, EbayOfferDetailsWithKeys, FeesSummaryResponse, + GetInventoryItem, InventoryItem, InventoryItemGroup, + InventoryItemWithSkuLocale, InventoryItemWithSkuLocaleGroupid, InventoryLocation, InventoryLocationFull, InventoryLocationResponse, + LocationAvailabilityDetails, LocationDetails, + LocationMapping, LocationResponse, + MigrateListing, + NameValueList, OfferKeyWithId, OfferKeysWithId, OfferResponse, + OfferPriceQuantity, PublishByInventoryItemGroupRequest, PublishResponse, WithdrawByInventoryItemGroupRequest, @@ -519,6 +539,163 @@ def test_inventory_wrapper_supports_location_management(httpx_mock: HTTPXMock) - assert update_body["name"] == "Main warehouse" +def test_inventory_wrapper_supports_bulk_and_auxiliary_endpoints(httpx_mock: HTTPXMock) -> None: + httpx_mock.add_response( + method="POST", + url="https://api.ebay.com/sell/inventory/v1/bulk_create_or_replace_inventory_item", + json={"responses": [{"sku": "SKU-2", "statusCode": 200}]}, + ) + httpx_mock.add_response( + method="POST", + url="https://api.ebay.com/sell/inventory/v1/bulk_get_inventory_item", + json={"responses": [{"sku": "SKU-2"}]}, + ) + httpx_mock.add_response( + method="POST", + url="https://api.ebay.com/sell/inventory/v1/bulk_update_price_quantity", + json={"responses": [{"offerId": "OFFER-2", "sku": "SKU-2", "statusCode": 200}]}, + ) + httpx_mock.add_response( + method="GET", + url="https://api.ebay.com/sell/inventory/v1/inventory_item/SKU-2/product_compatibility", + json={"sku": "SKU-2", "compatibleProducts": [{"compatibilityProperties": [{"name": "make", "value": "Toyota"}]}]}, + ) + httpx_mock.add_response( + method="PUT", + url="https://api.ebay.com/sell/inventory/v1/inventory_item/SKU-2/product_compatibility", + json={"warnings": []}, + ) + httpx_mock.add_response( + method="DELETE", + url="https://api.ebay.com/sell/inventory/v1/inventory_item/SKU-2/product_compatibility", + status_code=204, + ) + httpx_mock.add_response( + method="POST", + url="https://api.ebay.com/sell/inventory/v1/bulk_migrate_listing", + json={"responses": [{"listingId": "ITEM-1", "statusCode": 200}]}, + ) + httpx_mock.add_response( + method="GET", + url="https://api.ebay.com/sell/inventory/v1/listing/ITEM-1/sku/SKU-2/locations", + json={"locations": [{"merchantLocationKey": "FC-1"}]}, + ) + httpx_mock.add_response( + method="PUT", + url="https://api.ebay.com/sell/inventory/v1/listing/ITEM-1/sku/SKU-2/locations", + status_code=204, + ) + httpx_mock.add_response( + method="DELETE", + url="https://api.ebay.com/sell/inventory/v1/listing/ITEM-1/sku/SKU-2/locations", + status_code=204, + ) + httpx_mock.add_response( + method="POST", + url="https://api.ebay.com/sell/inventory/v1/bulk_create_offer", + json={"responses": [{"offerId": "OFFER-2", "sku": "SKU-2", "marketplaceId": "EBAY_US"}]}, + ) + httpx_mock.add_response( + method="POST", + url="https://api.ebay.com/sell/inventory/v1/bulk_publish_offer", + json={"responses": [{"offerId": "OFFER-2", "listingId": "ITEM-2", "statusCode": 200}]}, + ) + + client = InventoryClient(build_transport()) + bulk_item_response = client.bulk_create_or_replace_inventory_item( + BulkInventoryItem( + requests=[InventoryItemWithSkuLocale(sku="SKU-2", locale="en_US", condition="NEW")] + ), + content_language="en-US", + ) + bulk_get_response = client.bulk_get_inventory_item( + BulkGetInventoryItem(requests=[GetInventoryItem(sku="SKU-2")]) + ) + bulk_price_response = client.bulk_update_price_quantity( + BulkPriceQuantity( + requests=[ + { + "sku": "SKU-2", + "shipToLocationAvailability": {"quantity": 5}, + "offers": [{"offerId": "OFFER-2", "availableQuantity": 2}], + } + ] + ) + ) + compatibility = client.get_product_compatibility("SKU-2") + compatibility_result = client.create_or_replace_product_compatibility( + "SKU-2", + Compatibility(compatibleProducts=[{"compatibilityProperties": [{"name": "make", "value": "Toyota"}]}]), + content_language="en-US", + ) + client.delete_product_compatibility("SKU-2") + migrate_response = client.bulk_migrate_listing(BulkMigrateListing(requests=[MigrateListing(listingId="ITEM-1")])) + location_mapping = client.get_sku_location_mapping("ITEM-1", "SKU-2") + client.create_or_replace_sku_location_mapping( + "ITEM-1", + "SKU-2", + LocationMapping(locations=[LocationAvailabilityDetails(merchantLocationKey="FC-1")]), + ) + client.delete_sku_location_mapping("ITEM-1", "SKU-2") + bulk_offer_response = client.bulk_create_offer( + BulkEbayOfferDetailsWithKeys( + requests=[EbayOfferDetailsWithKeys(sku="SKU-2", marketplaceId="EBAY_US", format="FIXED_PRICE")] + ), + content_language="en-US", + ) + bulk_publish_response = client.bulk_publish_offer( + BulkOffer(requests=[OfferKeyWithId(offerId="OFFER-2")]) + ) + + assert isinstance(bulk_item_response, BulkInventoryItemResponse) + assert bulk_item_response.responses and bulk_item_response.responses[0].sku == "SKU-2" + assert isinstance(bulk_get_response, BulkGetInventoryItemResponse) + assert bulk_get_response.responses and bulk_get_response.responses[0].sku == "SKU-2" + assert isinstance(bulk_price_response, BulkPriceQuantityResponse) + assert bulk_price_response.responses and bulk_price_response.responses[0].offerId == "OFFER-2" + assert isinstance(compatibility, Compatibility) + assert compatibility.sku == "SKU-2" + assert isinstance(compatibility_result, BaseResponse) + assert isinstance(migrate_response, BulkMigrateListingResponse) + assert migrate_response.responses and migrate_response.responses[0].listingId == "ITEM-1" + assert isinstance(location_mapping, LocationMapping) + assert location_mapping.locations and location_mapping.locations[0].merchantLocationKey == "FC-1" + assert isinstance(bulk_offer_response, BulkOfferResponse) + assert bulk_offer_response.responses and bulk_offer_response.responses[0].offerId == "OFFER-2" + assert isinstance(bulk_publish_response, BulkPublishResponse) + assert bulk_publish_response.responses and bulk_publish_response.responses[0].listingId == "ITEM-2" + + bulk_item_request = httpx_mock.get_requests()[0] + assert bulk_item_request.headers["Content-Language"] == "en-US" + bulk_item_body = json.loads(bulk_item_request.content.decode("utf-8")) + assert bulk_item_body["requests"][0]["sku"] == "SKU-2" + + bulk_get_request = httpx_mock.get_requests()[1] + bulk_get_body = json.loads(bulk_get_request.content.decode("utf-8")) + assert bulk_get_body["requests"][0]["sku"] == "SKU-2" + + bulk_price_request = httpx_mock.get_requests()[2] + bulk_price_body = json.loads(bulk_price_request.content.decode("utf-8")) + assert bulk_price_body["requests"][0]["sku"] == "SKU-2" + assert bulk_price_body["requests"][0]["offers"][0]["offerId"] == "OFFER-2" + + compatibility_request = httpx_mock.get_requests()[4] + assert compatibility_request.headers["Content-Language"] == "en-US" + + location_mapping_request = httpx_mock.get_requests()[8] + location_mapping_body = json.loads(location_mapping_request.content.decode("utf-8")) + assert location_mapping_body["locations"][0]["merchantLocationKey"] == "FC-1" + + bulk_offer_request = httpx_mock.get_requests()[10] + assert bulk_offer_request.headers["Content-Language"] == "en-US" + bulk_offer_body = json.loads(bulk_offer_request.content.decode("utf-8")) + assert bulk_offer_body["requests"][0]["sku"] == "SKU-2" + + bulk_publish_request = httpx_mock.get_requests()[11] + bulk_publish_body = json.loads(bulk_publish_request.content.decode("utf-8")) + assert bulk_publish_body["requests"] == [{"offerId": "OFFER-2"}] + + def test_fulfillment_wrapper_returns_order_model(httpx_mock: HTTPXMock) -> None: httpx_mock.add_response( method="GET",