HIM-public/ebay_client
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add bulk inventory and offer management methods to InventoryClient; enhance tests for new endpoints

Browse source
This commit is contained in:
claudi 2026-04-07 11:11:19 +02:00
parent f30b31ec00
commit 65d31a3f2f
2 changed files with 321 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

144
ebay_client/inventory/client.py
View file

@ -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],
)

177
tests/test_public_wrappers.py
View file

@ -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",