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

Compare commits

base: HIM-public:937cd86c8b8e7d9e4117fe6165ab52e8f8e6329e
Branches Tags
HIM-public:main
..
compare: HIM-public:acdadad4b35481d490939eba76aba3589c8409ff
Branches Tags
HIM-public:main

No commits in common. "937cd86c8b8e7d9e4117fe6165ab52e8f8e6329e" and "acdadad4b35481d490939eba76aba3589c8409ff" have entirely different histories.
937cd86c8b ... acdadad4b3

23 changed files with 42 additions and 3739 deletions
Download patch file Download diff file Expand all files Collapse all files

323
.tmp_notification_models.py
View file

@ -1,323 +0,0 @@
# generated by datamodel-codegen:
# filename: commerce_notification_v1_oas3.yaml
# timestamp: 2026-04-07T06:52:28+00:00
from __future__ import annotations
from typing import Any, Dict, List, Optional
from pydantic import BaseModel, Field
class Config(BaseModel):
alertEmail: Optional[str] = Field(
None,
description='This field is used to add or modify an email address that will be used for Notification API alerts associated with the application. <b>getConfig</b> can be used to get the email address currently being used for alerts.',
)
class CreateSubscriptionFilterRequest(BaseModel):
filterSchema: Optional[Dict[str, Dict[str, Any]]] = Field(
None,
description='The content of a subscription filter as a valid <a href="https://json-schema.org " target="_blank">JSON Schema Core document</a> (version 2020-12 or later). The <strong>filterSchema</strong> provided must describe the subscription\'s notification payload such that it supplies valid criteria to filter the subscription\'s notifications.<br><br><span class="tablenote"><b>Note:</b> Not all topics can have filters applied to them. Use <a href="/api-docs/commerce/notification/resources/topic/methods/getTopic">getTopic</a> and <a href="/api-docs/commerce/notification/resources/topic/methods/getTopics">getTopics</a> requests to determine if a specific topic is filterable. Filterable topics have the boolean <b>filterable</b> returned as <code>true</code> in the response.</span><br><span class="tablenote"><b>Note:</b> If the JSON supplied as a subscription filter specifies a field that does not exist in the notifications for a topic, or if the topic is not filterable, the filter will be rejected and become <strong>DISABLED</strong>. If it is valid, however, the filter will move from <strong>PENDING</strong> status to <strong>ENABLED</strong> status.</span><br>Initially, when the <b>createSubscriptionFilter</b> request has been made, if the request has a valid JSON body a <b>201&nbsp;Created</b> is returned. After that, the validation of the <b>filterSchema</b> happens. See <a href="/api-docs/commerce/notification/overview.html#create-filter" target="_blank">Creating a subscription filter for a topic</a> for additional information.',
)
class DeliveryConfig(BaseModel):
endpoint: Optional[str] = Field(
None,
description='The endpoint for this destination.<br><br><span class="tablenote"><b>Note:</b> The provided endpoint URL should use the HTTPS protocol, and it should not contain an internal IP address or <code>localhost</code> in its path.</span>',
)
verificationToken: Optional[str] = Field(
None,
description='The verification token associated with this endpoint.<br><br><span class="tablenote"><b>Note:</b> The provided verification token must be between 32 and 80 characters. Allowed characters include alphanumeric characters, underscores (<code>_</code>), and hyphens (<code>-</code>); no other characters are allowed.</span>',
)
class Destination(BaseModel):
deliveryConfig: Optional[DeliveryConfig] = Field(
None, description='The configuration associated with this destination.'
)
destinationId: Optional[str] = Field(
None, description='The unique identifier for the destination.'
)
name: Optional[str] = Field(
None, description='The name associated with this destination.'
)
status: Optional[str] = Field(
None,
description='The status for this destination.<br><br><span class="tablenote"><b>Note:</b> The <b>MARKED_DOWN</b> value is set by eBay systems and cannot be used in a create or update call by applications.</span><br><br><b>Valid values:</b><ul><li><code>ENABLED</code></li><li><code>DISABLED</code></li><li><code>MARKED_DOWN</code></li></ul> For implementation help, refer to <a href=\'https://developer.ebay.com/api-docs/commerce/notification/types/api:DestinationStatusEnum\'>eBay API documentation</a>',
)
class DestinationRequest(BaseModel):
deliveryConfig: Optional[DeliveryConfig] = Field(
None,
description='This container is used to specify the destination endpoint and verification token associated with this endpoint.',
)
name: Optional[str] = Field(
None, description='The seller-specified name for the destination endpoint.'
)
status: Optional[str] = Field(
None,
description='This field sets the status for the destination endpoint as <code>ENABLED</code> or <code>DISABLED</code>.<br><br><span class="tablenote"><b>Note:</b> The <b>MARKED_DOWN</b> value is set by eBay systems and cannot be used in a create or update call by applications.</span> For implementation help, refer to <a href=\'https://developer.ebay.com/api-docs/commerce/notification/types/api:DestinationStatusEnum\'>eBay API documentation</a>',
)
class DestinationSearchResponse(BaseModel):
destinations: Optional[List[Destination]] = Field(
None, description='An array that contains the destination details.'
)
href: Optional[str] = Field(
None,
description='The path to the call URI that produced the current page of results.',
)
limit: Optional[int] = Field(
None,
description='The number of records to show in the current response.<br><br><b>Default:</b> 20',
)
next: Optional[str] = Field(
None,
description='The URL to access the next set of results. This field includes a <strong>continuation_token</strong>. No <b>prev</b> field is returned, but this value is persistent during the session so that you can use it to return to the next page.<br><br>This field is not returned if fewer records than specified by the <strong>limit</strong> field are returned.',
)
total: Optional[int] = Field(
None, description='The total number of matches for the search criteria.'
)
class ErrorParameter(BaseModel):
name: Optional[str] = Field(None, description='The object of the error.')
value: Optional[str] = Field(None, description='The value of the object.')
class PayloadDetail(BaseModel):
deliveryProtocol: Optional[str] = Field(
None,
description="The supported delivery protocols. For implementation help, refer to <a href='https://developer.ebay.com/api-docs/commerce/notification/types/api:ProtocolEnum'>eBay API documentation</a>",
)
deprecated: Optional[bool] = Field(None, description='A deprecation indicator.')
format: Optional[List[str]] = Field(
None,
description='The supported format. Presently, <code>JSON</code> is the only supported format.',
)
schemaVersion: Optional[str] = Field(
None, description='The supported schema version.'
)
class PublicKey(BaseModel):
algorithm: Optional[str] = Field(
None,
description='The algorithm associated with the public key that is returned, such as Elliptic Curve Digital Signature Algorithm (ECDSA).',
)
digest: Optional[str] = Field(
None,
description='The digest associated with the public key that is returned, such as Secure Hash Algorithm 1 (SHA1).',
)
key: Optional[str] = Field(
None,
description='The public key that is returned for the specified key ID.<br><br>This value is used to validate the eBay push notification message payload.',
)
class SubscriptionFilter(BaseModel):
creationDate: Optional[str] = Field(
None, description='The creation date for this subscription filter.'
)
filterId: Optional[str] = Field(
None, description='The unique identifier for this subscription filter.'
)
filterSchema: Optional[Dict[str, Dict[str, Any]]] = Field(
None,
description='The content of this subscription filter as a valid <a href="https://json-schema.org " target="_blank">JSON Schema Core document</a> (version 2020-12 or later). The <strong>filterSchema</strong> provided must describe the subscription\'s notification payload such that it supplies valid criteria to filter the subscription\'s notifications.',
)
filterStatus: Optional[str] = Field(
None,
description="The status of this subscription filter. For implementation help, refer to <a href='https://developer.ebay.com/api-docs/commerce/notification/types/api:SubscriptionFilterStatus'>eBay API documentation</a>",
)
subscriptionId: Optional[str] = Field(
None, description='The unique identifier for the subscription.'
)
class SubscriptionPayloadDetail(BaseModel):
deliveryProtocol: Optional[str] = Field(
None,
description='The supported delivery protocol of the notification topic.<br><br><span class="tablenote"><b>Note:</b> <code>HTTPS</code> is currently the only supported delivery protocol of all notification topics. </span> For implementation help, refer to <a href=\'https://developer.ebay.com/api-docs/commerce/notification/types/api:ProtocolEnum\'>eBay API documentation</a>',
)
format: Optional[str] = Field(
None,
description='The supported data format of the payload.<br><br><span class="tablenote"><b>Note:</b> JSON is currently the only supported format for all notification topics.</span> For implementation help, refer to <a href=\'https://developer.ebay.com/api-docs/commerce/notification/types/api:FormatTypeEnum\'>eBay API documentation</a>',
)
schemaVersion: Optional[str] = Field(
None,
description='The supported schema version for the notification topic. See the <b>supportedPayloads.schemaVersion</b> field for the topic in <b>getTopics</b> or <b>getTopic</b> response.',
)
class Topic(BaseModel):
authorizationScopes: Optional[List[str]] = Field(
None,
description='The authorization scopes required to subscribe to this topic.',
)
context: Optional[str] = Field(
None,
description="The business context associated with this topic. For implementation help, refer to <a href='https://developer.ebay.com/api-docs/commerce/notification/types/api:ContextEnum'>eBay API documentation</a>",
)
description: Optional[str] = Field(
None, description='The description of the topic.'
)
filterable: Optional[bool] = Field(
None, description='The indicator of whether this topic is filterable or not.'
)
scope: Optional[str] = Field(
None,
description="The scope of this topic. For implementation help, refer to <a href='https://developer.ebay.com/api-docs/commerce/notification/types/api:ScopeEnum'>eBay API documentation</a>",
)
status: Optional[str] = Field(
None,
description="The status of this topic. For implementation help, refer to <a href='https://developer.ebay.com/api-docs/commerce/notification/types/api:StatusEnum'>eBay API documentation</a>",
)
supportedPayloads: Optional[List[PayloadDetail]] = Field(
None, description='The supported payloads for this topic.'
)
topicId: Optional[str] = Field(
None, description='The unique identifier for the topic.'
)
class TopicSearchResponse(BaseModel):
href: Optional[str] = Field(
None,
description='The path to the call URI that produced the current page of results.',
)
limit: Optional[int] = Field(
None,
description='The value of the limit parameter submitted in the request, which is the maximum number of items to return per page, from the result set. A result set is the complete set of results returned by the method.<br><br><span class="tablenote"><b>Note:</b> Though this parameter is not required to be submitted in the request, the parameter defaults to <code>20</code> if omitted.</span>',
)
next: Optional[str] = Field(
None,
description='The URL to access the next set of results. This field includes a <strong>continuation_token</strong>. No <b>prev</b> field is returned, but this value is persistent during the session so that you can use it to return to the next page.<br><br>This field is not returned if fewer records than specified by the <strong>limit</strong> field are returned.',
)
topics: Optional[List[Topic]] = Field(
None, description='An array of topics that match the specified criteria.'
)
total: Optional[int] = Field(
None, description='The total number of matches for the search criteria.'
)
class UpdateSubscriptionRequest(BaseModel):
destinationId: Optional[str] = Field(
None,
description='The unique identifier of the destination endpoint that will receive notifications associated with this subscription. Use <b>getDestinations</b> to retrieve destination IDs.',
)
payload: Optional[SubscriptionPayloadDetail] = Field(
None, description='The payload associated with this subscription.'
)
status: Optional[str] = Field(
None,
description="Set the status of the subscription being updated to ENABLED or DISABLED. For implementation help, refer to <a href='https://developer.ebay.com/api-docs/commerce/notification/types/api:SubscriptionStatusEnum'>eBay API documentation</a>",
)
class CreateSubscriptionRequest(BaseModel):
destinationId: Optional[str] = Field(
None,
description='The unique identifier of the destination endpoint that will receive notifications associated with this subscription. Use the <b>getDestinations</b> method to retrieve destination IDs.',
)
payload: Optional[SubscriptionPayloadDetail] = Field(
None,
description='The payload associated with the notification topic. Use <b>getTopics</b> or <b>getTopic</b> to get the supported payload for the topic.',
)
status: Optional[str] = Field(
None,
description="Set the status of the subscription to <code>ENABLED</code> or <code>DISABLED</code>. For implementation help, refer to <a href='https://developer.ebay.com/api-docs/commerce/notification/types/api:SubscriptionStatusEnum'>eBay API documentation</a>",
)
topicId: Optional[str] = Field(
None,
description='The unique identifier of the notification topic to subscribe to. Use <b>getTopics</b> to get topic IDs.',
)
class Error(BaseModel):
category: Optional[str] = Field(None, description='Identifies the type of erro.')
domain: Optional[str] = Field(
None,
description='Name for the primary system where the error occurred. This is relevant for application errors.',
)
errorId: Optional[int] = Field(
None, description='A unique number to identify the error.'
)
inputRefIds: Optional[List[str]] = Field(
None,
description='An array of request elements most closely associated to the error.',
)
longMessage: Optional[str] = Field(
None, description='A more detailed explanation of the error.'
)
message: Optional[str] = Field(
None,
description="Information on how to correct the problem, in the end user's terms and language where applicable.",
)
outputRefIds: Optional[List[str]] = Field(
None,
description='An array of request elements most closely associated to the error.',
)
parameters: Optional[List[ErrorParameter]] = Field(
None,
description='An array of name/value pairs that describe details the error condition. These are useful when multiple errors are returned.',
)
subdomain: Optional[str] = Field(
None,
description='Further helps indicate which subsystem the error is coming from. System subcategories include: Initialization, Serialization, Security, Monitoring, Rate Limiting, etc.',
)
class Subscription(BaseModel):
creationDate: Optional[str] = Field(
None, description='The creation date for this subscription.'
)
destinationId: Optional[str] = Field(
None,
description='The unique identifier for the destination associated with this subscription.',
)
filterId: Optional[str] = Field(
None,
description='The unique identifier for the filter associated with this subscription.',
)
payload: Optional[SubscriptionPayloadDetail] = Field(
None, description='The payload associated with this subscription.'
)
status: Optional[str] = Field(
None,
description="The status of this subscription. For implementation help, refer to <a href='https://developer.ebay.com/api-docs/commerce/notification/types/api:SubscriptionStatusEnum'>eBay API documentation</a>",
)
subscriptionId: Optional[str] = Field(
None, description='The unique identifier for the subscription.'
)
topicId: Optional[str] = Field(
None,
description='The unique identifier for the topic associated with this subscription.',
)
class SubscriptionSearchResponse(BaseModel):
href: Optional[str] = Field(
None,
description='The path to the call URI that produced the current page of results.',
)
limit: Optional[int] = Field(
None,
description='The value of the limit parameter submitted in the request, which is the maximum number of items to return per page, from the result set. A result set is the complete set of results returned by the method.<br><br><span class="tablenote"><b>Note:</b> Though this parameter is not required to be submitted in the request, the parameter defaults to <code>20</code> if omitted.</span><br><br><b>Default:</b> 20',
)
next: Optional[str] = Field(
None,
description='The URL to access the next set of results. This field includes a <strong>continuation_token</strong>. No <b>prev</b> field is returned, but this value is persistent during the session so that you can use it to return to the next page.<br><br>This field is not returned if fewer records than specified by the <strong>limit</strong> field are returned.',
)
subscriptions: Optional[List[Subscription]] = Field(
None, description='The subscriptions that match the search criteria.'
)
total: Optional[int] = Field(
None, description='The total number of matches for the search criteria.'
)

31
README.md
View file

@ -7,26 +7,6 @@ This workspace contains a Python-first eBay REST client foundation with:
- public API wrappers per eBay REST domain
- an isolated Pydantic model generation script for each contract in this folder
Currently wired API domains include Notification, Inventory, Fulfillment, Account, Feed, and Media.
## Media Helpers
The Media wrapper includes workflow helpers on top of the raw endpoints:
- `extract_resource_id()` to pull a media resource ID from a `Location` header
- `wait_for_video()` to poll until a video reaches `LIVE` or a terminal failure state
- `wait_for_document()` to poll until a document reaches `ACCEPTED` or a terminal failure state
- `create_upload_and_wait_video()` to stage, upload, and poll a video in one call
- `create_upload_and_wait_document()` to stage, upload, and poll a document in one call
- `create_document_from_url_and_wait()` to create a document from a URL and poll until it is accepted
A concrete workflow example is available in `examples/media_workflows.py` for:
- uploading an image from a file
- creating an image from a URL
- staging, uploading, and polling a document
- staging, uploading, and polling a video
## Generate Low-Level Clients
The project uses a dedicated code generation environment because the main runtime is currently on Python 3.14 while the model generator still targets earlier Python versions.
@ -44,14 +24,3 @@ To generate only one API package:
```
This regenerates Pydantic v2 models into `ebay_client/generated/<api>/models.py`.
## Webhook Helpers
The Notification package also includes framework-agnostic webhook utilities for:
- responding to eBay challenge requests
- parsing and validating the `X-EBAY-SIGNATURE` header
- verifying signed notification payloads against the cached public key
- turning a verified notification body into a normalized `WebhookEventEnvelope`
A concrete FastAPI integration example is available in `examples/fastapi_notification_webhook.py`.

1350
commerce_media_v1_beta_oas3.yaml
View file

File diff suppressed because it is too large Load diff

11
ebay_client/account/client.py
View file

@ -11,7 +11,6 @@ from ebay_client.generated.account.models import (
ACCOUNT_SCOPE = "https://api.ebay.com/oauth/api_scope/sell.account"
ACCOUNT_READ_SCOPE = "https://api.ebay.com/oauth/api_scope/sell.account.readonly"
ACCOUNT_READ_SCOPE_OPTIONS = [[ACCOUNT_READ_SCOPE], [ACCOUNT_SCOPE]]
class AccountClient:
@ -23,7 +22,7 @@ class AccountClient:
FulfillmentPolicyResponse,
"GET",
"/sell/account/v1/fulfillment_policy",
scope_options=ACCOUNT_READ_SCOPE_OPTIONS,
scopes=[ACCOUNT_READ_SCOPE],
params={"marketplace_id": marketplace_id},
)
@ -32,7 +31,7 @@ class AccountClient:
PaymentPolicyResponse,
"GET",
"/sell/account/v1/payment_policy",
scope_options=ACCOUNT_READ_SCOPE_OPTIONS,
scopes=[ACCOUNT_READ_SCOPE],
params={"marketplace_id": marketplace_id},
)
@ -41,7 +40,7 @@ class AccountClient:
ReturnPolicyResponse,
"GET",
"/sell/account/v1/return_policy",
scope_options=ACCOUNT_READ_SCOPE_OPTIONS,
scopes=[ACCOUNT_READ_SCOPE],
params={"marketplace_id": marketplace_id},
)
@ -50,7 +49,7 @@ class AccountClient:
SellingPrivileges,
"GET",
"/sell/account/v1/privilege",
scope_options=ACCOUNT_READ_SCOPE_OPTIONS,
scopes=[ACCOUNT_SCOPE],
)
def get_opted_in_programs(self) -> Programs:
@ -58,5 +57,5 @@ class AccountClient:
Programs,
"GET",
"/sell/account/v1/program/get_opted_in_programs",
scope_options=ACCOUNT_READ_SCOPE_OPTIONS,
scopes=[ACCOUNT_READ_SCOPE],
)

2
ebay_client/client.py
View file

@ -7,7 +7,6 @@ from ebay_client.core.http.transport import ApiTransport
from ebay_client.feed.client import FeedClient
from ebay_client.fulfillment.client import FulfillmentClient
from ebay_client.inventory.client import InventoryClient
from ebay_client.media.client import MediaClient
from ebay_client.notification.client import NotificationClient
@ -32,4 +31,3 @@ class EbayClient:
self.fulfillment = FulfillmentClient(transport)
self.account = AccountClient(transport)
self.feed = FeedClient(transport)
self.media = MediaClient(transport)

49
ebay_client/core/auth/oauth.py
View file

@ -1,7 +1,7 @@
from __future__ import annotations
import base64
from typing import Iterable, Sequence
from typing import Iterable
from urllib.parse import urlencode
import httpx
@ -44,17 +44,10 @@ class EbayOAuthClient:
query["prompt"] = prompt
return f"{self.config.auth_base_url}?{urlencode(query)}"
def get_valid_token(
self,
*,
scopes: Iterable[str] | None = None,
scope_options: Sequence[Iterable[str]] | None = None,
) -> OAuthToken:
def get_valid_token(self, *, scopes: Iterable[str] | None = None) -> OAuthToken:
token = self.token_store.get_token()
if token is None or token.is_expired() or not self._has_required_scopes(token, scopes=scopes, scope_options=scope_options):
token = self.fetch_client_credentials_token(
scopes=self._choose_requested_scopes(scopes=scopes, scope_options=scope_options)
)
if token is None or token.is_expired() or not self._has_required_scopes(token, scopes):
token = self.fetch_client_credentials_token(scopes=scopes)
return token
def fetch_client_credentials_token(self, *, scopes: Iterable[str] | None = None) -> OAuthToken:
@ -118,34 +111,8 @@ class EbayOAuthClient:
return base64.b64encode(raw).decode("ascii")
@staticmethod
def _choose_requested_scopes(
*,
scopes: Iterable[str] | None = None,
scope_options: Sequence[Iterable[str]] | None = None,
) -> list[str] | None:
if scopes is not None:
requested = [scope for scope in scopes if scope]
return requested or None
if scope_options:
for option in scope_options:
requested = [scope for scope in option if scope]
if requested:
return requested
return None
@staticmethod
def _has_required_scopes(
token: OAuthToken,
*,
scopes: Iterable[str] | None = None,
scope_options: Sequence[Iterable[str]] | None = None,
) -> bool:
requested_sets: list[set[str]] = []
if scopes is not None:
requested_sets.append({scope for scope in scopes if scope})
if scope_options:
requested_sets.extend({scope for scope in option if scope} for option in scope_options)
if not requested_sets:
def _has_required_scopes(token: OAuthToken, scopes: Iterable[str] | None) -> bool:
requested = {scope for scope in (scopes or []) if scope}
if not requested:
return True
token_scopes = token.scopes()
return any(requested.issubset(token_scopes) for requested in requested_sets if requested)
return requested.issubset(token.scopes())

14
ebay_client/core/http/transport.py
View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import Any, Mapping, Sequence, TypeVar
from typing import Any, Mapping, TypeVar
import httpx
from pydantic import BaseModel
@ -31,31 +31,25 @@ class ApiTransport:
path: str,
*,
scopes: list[str] | None = None,
scope_options: Sequence[Sequence[str]] | None = None,
params: Mapping[str, Any] | None = None,
json_body: Any | None = None,
headers: Mapping[str, str] | None = None,
content: bytes | None = None,
files: Any | None = None,
) -> httpx.Response:
token = self.oauth_client.get_valid_token(scopes=scopes, scope_options=scope_options)
token = self.oauth_client.get_valid_token(scopes=scopes)
request_headers = dict(self.default_headers)
request_headers.update(headers or {})
request_headers["Authorization"] = f"Bearer {token.access_token}"
filtered_params = None
if params is not None:
filtered_params = {key: value for key, value in params.items() if value is not None}
url = f"{self.base_url}{path}"
try:
with httpx.Client(timeout=self.timeout_seconds) as client:
response = client.request(
method,
url,
params=filtered_params,
params=params,
json=json_body,
headers=request_headers,
content=content,
files=files,
)
except httpx.HTTPError as exc:
raise TransportError(f"HTTP request failed for {method} {path}") from exc
@ -72,8 +66,6 @@ class ApiTransport:
response = self.request(method, path, **kwargs)
if response.status_code == 204:
return None
if not response.content or not response.content.strip():
return None
try:
return response.json()
except ValueError as exc:

18
ebay_client/feed/client.py
View file

@ -10,16 +10,6 @@ from ebay_client.generated.feed.models import (
FEED_INVENTORY_SCOPE = "https://api.ebay.com/oauth/api_scope/sell.inventory"
FEED_FULFILLMENT_SCOPE = "https://api.ebay.com/oauth/api_scope/sell.fulfillment"
FEED_MARKETING_SCOPE = "https://api.ebay.com/oauth/api_scope/sell.marketing"
FEED_CATALOG_READ_SCOPE = "https://api.ebay.com/oauth/api_scope/commerce.catalog.readonly"
FEED_ANALYTICS_READ_SCOPE = "https://api.ebay.com/oauth/api_scope/sell.analytics.readonly"
FEED_READ_SCOPE_OPTIONS = [
[FEED_INVENTORY_SCOPE],
[FEED_FULFILLMENT_SCOPE],
[FEED_MARKETING_SCOPE],
[FEED_CATALOG_READ_SCOPE],
[FEED_ANALYTICS_READ_SCOPE],
]
class FeedClient:
@ -31,7 +21,7 @@ class FeedClient:
TaskCollection,
"GET",
"/sell/feed/v1/task",
scope_options=FEED_READ_SCOPE_OPTIONS,
scopes=[FEED_INVENTORY_SCOPE, FEED_FULFILLMENT_SCOPE],
params={"feed_type": feed_type},
)
@ -40,7 +30,7 @@ class FeedClient:
Task,
"GET",
f"/sell/feed/v1/task/{task_id}",
scope_options=FEED_READ_SCOPE_OPTIONS,
scopes=[FEED_INVENTORY_SCOPE, FEED_FULFILLMENT_SCOPE],
)
def get_schedule_templates(self) -> ScheduleTemplateCollection:
@ -48,7 +38,7 @@ class FeedClient:
ScheduleTemplateCollection,
"GET",
"/sell/feed/v1/schedule_template",
scope_options=FEED_READ_SCOPE_OPTIONS,
scopes=[FEED_INVENTORY_SCOPE, FEED_FULFILLMENT_SCOPE],
)
def get_schedules(self) -> UserScheduleCollection:
@ -56,5 +46,5 @@ class FeedClient:
UserScheduleCollection,
"GET",
"/sell/feed/v1/schedule",
scope_options=FEED_READ_SCOPE_OPTIONS,
scopes=[FEED_INVENTORY_SCOPE, FEED_FULFILLMENT_SCOPE],
)

10
ebay_client/fulfillment/client.py
View file

@ -8,9 +8,7 @@ from ebay_client.generated.fulfillment.models import (
ShippingFulfillmentPagedCollection,
)
FULFILLMENT_SCOPE = "https://api.ebay.com/oauth/api_scope/sell.fulfillment"
FULFILLMENT_READ_SCOPE = "https://api.ebay.com/oauth/api_scope/sell.fulfillment.readonly"
FULFILLMENT_READ_SCOPE_OPTIONS = [[FULFILLMENT_READ_SCOPE], [FULFILLMENT_SCOPE]]
class FulfillmentClient:
@ -22,7 +20,7 @@ class FulfillmentClient:
Order,
"GET",
f"/sell/fulfillment/v1/order/{order_id}",
scope_options=FULFILLMENT_READ_SCOPE_OPTIONS,
scopes=[FULFILLMENT_READ_SCOPE],
)
def get_orders(self, *, limit: int | None = None, offset: int | None = None) -> OrderSearchPagedCollection:
@ -30,7 +28,7 @@ class FulfillmentClient:
OrderSearchPagedCollection,
"GET",
"/sell/fulfillment/v1/order",
scope_options=FULFILLMENT_READ_SCOPE_OPTIONS,
scopes=[FULFILLMENT_READ_SCOPE],
params={"limit": limit, "offset": offset},
)
@ -39,7 +37,7 @@ class FulfillmentClient:
ShippingFulfillmentPagedCollection,
"GET",
f"/sell/fulfillment/v1/order/{order_id}/shipping_fulfillment",
scope_options=FULFILLMENT_READ_SCOPE_OPTIONS,
scopes=[FULFILLMENT_READ_SCOPE],
)
def get_shipping_fulfillment(self, order_id: str, fulfillment_id: str) -> ShippingFulfillment:
@ -47,5 +45,5 @@ class FulfillmentClient:
ShippingFulfillment,
"GET",
f"/sell/fulfillment/v1/order/{order_id}/shipping_fulfillment/{fulfillment_id}",
scope_options=FULFILLMENT_READ_SCOPE_OPTIONS,
scopes=[FULFILLMENT_READ_SCOPE],
)

3
ebay_client/generated/media/__init__.py
View file

@ -1,3 +0,0 @@
"""Generated Pydantic models from the OpenAPI contract."""
from .models import *

281
ebay_client/generated/media/models.py
View file

@ -1,281 +0,0 @@
# generated by datamodel-codegen:
# filename: commerce_media_v1_beta_oas3.yaml
# timestamp: 2026-04-07T08:09:11+00:00
from __future__ import annotations
from typing import List, Optional
from pydantic import BaseModel, Field
class CreateDocumentFromUrlRequest(BaseModel):
"""
This type contains the metadata used to create the document ID when creating a document using a URL.
"""
documentType: Optional[str] = Field(
None,
description="The type of the document being created. For example, a <code>USER_GUIDE_OR_MANUAL</code> or a <code>SAFETY_DATA_SHEET</code>. For implementation help, refer to <a href='https://developer.ebay.com/api-docs/commerce/media/types/api:DocumentTypeEnum'>eBay API documentation</a>",
)
documentUrl: Optional[str] = Field(
None,
description="The URL of the document being created.<br><br>The document referenced by the URL must be a .pdf, .png, .jpg, or .jpeg file, and must be no larger than 10 MB.",
)
languages: Optional[List[str]] = Field(
None, description="This array shows the language(s) used in the document."
)
class CreateDocumentRequest(BaseModel):
"""
This type contains the metadata used to create the document ID.
"""
documentType: Optional[str] = Field(
None,
description="The type of the document being uploaded. For example, a <code>USER_GUIDE_OR_MANUAL</code> or a <code>SAFETY_DATA_SHEET</code>. For implementation help, refer to <a href='https://developer.ebay.com/api-docs/commerce/media/types/api:DocumentTypeEnum'>eBay API documentation</a>",
)
languages: Optional[List[str]] = Field(
None, description="This array shows the language(s) used in the document."
)
class CreateDocumentResponse(BaseModel):
"""
This type provides information about the created document ID.
"""
documentId: Optional[str] = Field(
None,
description='The unique identifier of the document to be uploaded.<br><br>This value is returned in the response and <b>location</b> header of the <b>createDocument</b> and <b>createDocumentFromUrl</b> methods. This ID can be used with the <b>getDocument</b> and <b>uploadDocument</b> methods, and to add an uploaded document to a listing. See <a href="/api-docs/sell/static/inventory/managing-document-media.html#add-documents" target="_blank">Adding documents to listings</a> for more information. ',
)
documentStatus: Optional[str] = Field(
None,
description="The status of the document resource.<br><br>For example, the value <code>PENDING_UPLOAD</code> is the initial state when the reference to the document has been created using the <b>createDocument</b> method. When creating a document using the <b>createDocumentFromUrl</b> method, the initial state will be <code>SUBMITTED</code>. For implementation help, refer to <a href='https://developer.ebay.com/api-docs/commerce/media/types/api:DocumentStatusEnum'>eBay API documentation</a>",
)
documentType: Optional[str] = Field(
None,
description="The type of the document uploaded. For example, <code>USER_GUIDE_OR_MANUAL</code>. For implementation help, refer to <a href='https://developer.ebay.com/api-docs/commerce/media/types/api:DocumentTypeEnum'>eBay API documentation</a>",
)
languages: Optional[List[str]] = Field(
None, description="This array shows the language(s) used in the document."
)
class CreateImageFromUrlRequest(BaseModel):
"""
A type that provides the location of the image.
"""
imageUrl: Optional[str] = Field(
None,
description='The image URL of the self-hosted picture to upload to eBay Picture Services (EPS). In addition to the picture requirements in <a href="https://www.ebay.com/help/policies/listing-policies/picture-policy?id=4370" target="_blank">Picture policy</a>, the provided URL must be secured using HTTPS (HTTP is not permitted). For more information, see <a href="/api-docs/sell/static/inventory/managing-image-media.html#image-requirements" target="_blank">Image requirements</a>.',
)
class CreateVideoRequest(BaseModel):
"""
The request to create a video, which must contain the video's <b>title</b>, <b>size</b>, and <b>classification</b>. <b>Description</b> is an optional field when creating videos.
"""
classification: Optional[List[str]] = Field(
None,
description="The intended use for this video content. Currently, videos can only be added and associated with eBay listings, so the only supported value is <code>ITEM</code>.",
)
description: Optional[str] = Field(
None, description="The description of the video."
)
size: Optional[int] = Field(
None,
description="The size, in bytes, of the video content. <br><br><b>Max:</b> 157,286,400 bytes",
)
title: Optional[str] = Field(None, description="The title of the video.")
class DocumentMetadata(BaseModel):
"""
This type provides information about the <b>documentId</b>.
"""
fileName: Optional[str] = Field(
None,
description="The name of the file including its extension (for example, <code>drone_user_warranty.pdf</code>).",
)
fileSize: Optional[str] = Field(
None, description="The size, in bytes, of the document content."
)
fileType: Optional[str] = Field(
None,
description="The type of the file uploaded. Supported file types include the following: <code>pdf</code>, <code>jpeg</code>, <code>jpg</code>, and <code>png</code>.",
)
class DocumentResponse(BaseModel):
"""
This type provides information returned about a created document ID, which may or may not have been uploaded.
"""
documentId: Optional[str] = Field(
None, description="The unique ID of the document."
)
documentMetadata: Optional[DocumentMetadata] = Field(
None,
description="This container provides the name, size, and type of the specified file.",
)
documentStatus: Optional[str] = Field(
None,
description="The status of the document resource.<br><br>Once a document has been uploaded using the <b>uploadDocument</b> method, the <b>documentStatus</b> will be <code>SUBMITTED</code>. The document will then either be accepted or rejected. Only documents with the status of <code>ACCEPTED</code> are available to be added to a listing. For implementation help, refer to <a href='https://developer.ebay.com/api-docs/commerce/media/types/api:DocumentStatusEnum'>eBay API documentation</a>",
)
documentType: Optional[str] = Field(
None,
description="The type of the document uploaded. For example, <code>USER_GUIDE_OR_MANUAL</code>. For implementation help, refer to <a href='https://developer.ebay.com/api-docs/commerce/media/types/api:DocumentTypeEnum'>eBay API documentation</a>",
)
languages: Optional[List[str]] = Field(
None, description="This array shows the language(s) used in the document."
)
class ErrorParameter(BaseModel):
name: Optional[str] = Field(None, description="The object of the error.")
value: Optional[str] = Field(None, description="The value of the object.")
class Image(BaseModel):
"""
A type that provides the location of the image.
"""
imageUrl: Optional[str] = Field(
None, description="The URL of the image's location."
)
class ImageResponse(BaseModel):
"""
A type that provides an image's details including its URL and expiration.
"""
expirationDate: Optional[str] = Field(
None,
description="The date and time when an unused EPS image will expire and be removed from the EPS server, in Coordinated Universal Time (UTC). As long as an EPS image is being used in an active listing, that image will remain on the EPS server and be accessible.",
)
imageUrl: Optional[str] = Field(
None,
description="The EPS URL to access the uploaded image. This URL will be used in listing calls to add the image to a listing.",
)
maxDimensionImageUrl: Optional[str] = Field(
None,
description="The EPS URL to access the maximum dimension version of the uploaded image.",
)
class InputStream(BaseModel):
"""
The streaming input of the video source. The input source must be an .mp4 file of the type MPEG-4 Part 10 or Advanced Video Coding (MPEG-4 AVC).
"""
class Moderation(BaseModel):
"""
A container that provides video moderation information when calling the <strong>getVideo</strong> method.<br /><br />This container is returned if the specified video has been blocked by moderators.<br /><br /><span class="tablenote"><span style="color:#478415"><strong>Tip:</strong></span> See <a href="https://www.ebay.com/help/selling/listings/creating-managing-listings/add-video-to-listing?id=5272#section2" target="_blank">Video moderation and restrictions</a> in the eBay Seller Center for details about video moderation.</span>
"""
rejectReasons: Optional[List[str]] = Field(
None,
description="The reason(s) why the specified video was blocked by moderators.",
)
class Play(BaseModel):
"""
The two streaming video URLs available for a successfully uploaded video with a status of <code>LIVE</code>. The supported streaming video protocols are DASH (Dynamic Adaptive Streaming over HTTP) and HLS (HTTP Live Streaming).
"""
playUrl: Optional[str] = Field(None, description="The playable URL for this video.")
protocol: Optional[str] = Field(
None,
description="The protocol for the video playlist. Supported protocols are DASH (Dynamic Adaptive Streaming over HTTP) and HLS (HTTP\xa0Live Streaming). For implementation help, refer to <a href='https://developer.ebay.com/api-docs/commerce/media/types/api:ProtocolEnum'>eBay API documentation</a>",
)
class Video(BaseModel):
"""
A response field that retrieves all the metadata for the video, including its <b>title</b>, <b>classification</b>, <b>size</b>, <b>description</b>, <b>status</b>, <b>status message</b> (if any), and <b>expiration date</b>.
"""
classification: Optional[List[str]] = Field(
None,
description="The intended use for this video content. Currently, videos can only be added and associated with eBay listings, so the only supported value is <code>ITEM</code>.",
)
description: Optional[str] = Field(
None,
description='The description of the video. The video description is an optional field that can be set using the <a href=" /api-docs/commerce/media/resources/video/methods/createVideo" target="_blank">createVideo</a> method.',
)
expirationDate: Optional[str] = Field(
None,
description="The date and time when an unused video will expire and be removed from the eBay Video Services server, in Coordinated Universal Time (UTC).<br><br>As long as a video is being used in an active listing, that video will remain on the server and be accessible. If a video is not being used on an active listing, its expiration date is automatically set to 30 days after the video's initial upload.",
)
moderation: Optional[Moderation] = Field(
None,
description='The video moderation information that is returned if a video is blocked by moderators.<br /><br /><span class="tablenote"><span style="color:#478415"><strong>Tip:</strong></span> See <a href="https://www.ebay.com/help/selling/listings/creating-managing-listings/add-video-to-listing?id=5272#section2" target="_blank">Video moderation and restrictions</a> in the eBay Seller Center for details about video moderation.</span><br /><br />If the video status is <code>BLOCKED</code>, ensure that the video complies with eBay\'s video formatting and content guidelines. Afterwards, begin the video creation and upload procedure anew using the <strong>createVideo</strong> and <strong>uploadVideo</strong> methods.',
)
playLists: Optional[List[Play]] = Field(
None,
description="The playlist created for the uploaded video, which provides the streaming video URLs to play the video. The supported streaming video protocols are DASH (Dynamic Adaptive Streaming over HTTP) and HLS (HTTP\xa0Live Streaming). The playlist will only be generated if a video is successfully uploaded with a status of <code>LIVE</code>.",
)
size: Optional[int] = Field(
None, description="The size, in bytes, of the video content."
)
status: Optional[str] = Field(
None,
description="The status of the current video resource. For implementation help, refer to <a href='https://developer.ebay.com/api-docs/commerce/media/types/api:VideoStatusEnum'>eBay API documentation</a>",
)
statusMessage: Optional[str] = Field(
None,
description="The <b>statusMessage</b> field contains additional information on the status. For example, information on why processing might have failed or if the video was blocked.",
)
thumbnail: Optional[Image] = Field(
None,
description="The URL of the thumbnail image of the video. The thumbnail image's URL must be an eBayPictureURL (EPS URL).",
)
title: Optional[str] = Field(None, description="The title of the video.")
videoId: Optional[str] = Field(None, description="The unique ID of the video.")
class Error(BaseModel):
"""
This type defines the fields that can be returned in an error.
"""
category: Optional[str] = Field(None, description="Identifies the type of erro.")
domain: Optional[str] = Field(
None,
description="Name for the primary system where the error occurred. This is relevant for application errors.",
)
errorId: Optional[int] = Field(
None, description="A unique number to identify the error."
)
inputRefIds: Optional[List[str]] = Field(
None,
description="An array of request elements most closely associated to the error.",
)
longMessage: Optional[str] = Field(
None, description="A more detailed explanation of the error."
)
message: Optional[str] = Field(
None,
description="Information on how to correct the problem, in the end user's terms and language where applicable.",
)
outputRefIds: Optional[List[str]] = Field(
None,
description="An array of request elements most closely associated to the error.",
)
parameters: Optional[List[ErrorParameter]] = Field(
None,
description="An array of name/value pairs that describe details the error condition. These are useful when multiple errors are returned.",
)
subdomain: Optional[str] = Field(
None,
description="Further helps indicate which subsystem the error is coming from. System subcategories include: Initialization, Serialization, Security, Monitoring, Rate Limiting, etc.",
)

10
ebay_client/inventory/client.py
View file

@ -8,9 +8,7 @@ from ebay_client.generated.inventory.models import (
Offers,
)
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:
@ -22,7 +20,7 @@ class InventoryClient:
InventoryItemWithSkuLocaleGroupid,
"GET",
f"/sell/inventory/v1/inventory_item/{sku}",
scope_options=INVENTORY_READ_SCOPE_OPTIONS,
scopes=[INVENTORY_READ_SCOPE],
)
def get_inventory_items(self, *, limit: int | None = None, offset: int | None = None) -> InventoryItems:
@ -30,7 +28,7 @@ class InventoryClient:
InventoryItems,
"GET",
"/sell/inventory/v1/inventory_item",
scope_options=INVENTORY_READ_SCOPE_OPTIONS,
scopes=[INVENTORY_READ_SCOPE],
params={"limit": limit, "offset": offset},
)
@ -39,7 +37,7 @@ class InventoryClient:
OfferResponseWithListingId,
"GET",
f"/sell/inventory/v1/offer/{offer_id}",
scope_options=INVENTORY_READ_SCOPE_OPTIONS,
scopes=[INVENTORY_READ_SCOPE],
)
def get_offers(self, *, limit: int | None = None, offset: int | None = None, sku: str | None = None) -> Offers:
@ -47,6 +45,6 @@ class InventoryClient:
Offers,
"GET",
"/sell/inventory/v1/offer",
scope_options=INVENTORY_READ_SCOPE_OPTIONS,
scopes=[INVENTORY_READ_SCOPE],
params={"limit": limit, "offset": offset, "sku": sku},
)

3
ebay_client/media/__init__.py
View file

@ -1,3 +0,0 @@
from ebay_client.media.client import CreatedMediaResource, MediaClient, extract_resource_id
__all__ = ["CreatedMediaResource", "MediaClient", "extract_resource_id"]

294
ebay_client/media/client.py
View file

@ -1,294 +0,0 @@
from __future__ import annotations
from time import monotonic, sleep
from urllib.parse import urlparse
from pydantic import BaseModel
from ebay_client.core.http.transport import ApiTransport
from ebay_client.generated.media.models import (
CreateDocumentFromUrlRequest,
CreateDocumentRequest,
CreateDocumentResponse,
CreateImageFromUrlRequest,
CreateVideoRequest,
DocumentResponse,
ImageResponse,
Video,
)
MEDIA_SCOPE = "https://api.ebay.com/oauth/api_scope/sell.inventory"
class CreatedMediaResource(BaseModel):
location: str | None = None
resource_id: str | None = None
def extract_resource_id(location: str | None) -> str | None:
if not location:
return None
path = urlparse(location).path.rstrip("/")
if not path:
return None
_, _, resource_id = path.rpartition("/")
return resource_id or None
class MediaClient:
def __init__(self, transport: ApiTransport) -> None:
self.transport = transport
def create_image_from_file(
self,
*,
file_name: str,
content: bytes,
content_type: str = "application/octet-stream",
) -> ImageResponse:
return self.transport.request_model(
ImageResponse,
"POST",
"/commerce/media/v1_beta/image/create_image_from_file",
scopes=[MEDIA_SCOPE],
files={"image": (file_name, content, content_type)},
)
def create_image_from_url(self, payload: CreateImageFromUrlRequest) -> ImageResponse:
return self.transport.request_model(
ImageResponse,
"POST",
"/commerce/media/v1_beta/image/create_image_from_url",
scopes=[MEDIA_SCOPE],
headers={"Content-Type": "application/json"},
json_body=payload.model_dump(by_alias=True, exclude_none=True),
)
def get_image(self, image_id: str) -> ImageResponse:
return self.transport.request_model(
ImageResponse,
"GET",
f"/commerce/media/v1_beta/image/{image_id}",
scopes=[MEDIA_SCOPE],
)
def create_video(self, payload: CreateVideoRequest) -> CreatedMediaResource:
response = self.transport.request(
"POST",
"/commerce/media/v1_beta/video",
scopes=[MEDIA_SCOPE],
headers={"Content-Type": "application/json"},
json_body=payload.model_dump(by_alias=True, exclude_none=True),
)
location = response.headers.get("Location")
return CreatedMediaResource(location=location, resource_id=extract_resource_id(location))
def create_upload_and_wait_video(
self,
payload: CreateVideoRequest,
*,
content: bytes,
content_length: int | None = None,
content_range: str | None = None,
timeout_seconds: float = 30.0,
poll_interval_seconds: float = 1.0,
) -> Video:
created = self.create_video(payload)
video_id = self._require_resource_id(created.resource_id, "video resource ID")
self.upload_video(
video_id,
content=content,
content_length=content_length if content_length is not None else len(content),
content_range=content_range,
)
return self.wait_for_video(
video_id,
timeout_seconds=timeout_seconds,
poll_interval_seconds=poll_interval_seconds,
)
def get_video(self, video_id: str) -> Video:
return self.transport.request_model(
Video,
"GET",
f"/commerce/media/v1_beta/video/{video_id}",
scopes=[MEDIA_SCOPE],
)
def upload_video(
self,
video_id: str,
*,
content: bytes,
content_length: int | None = None,
content_range: str | None = None,
) -> None:
headers = {"Content-Type": "application/octet-stream"}
if content_length is not None:
headers["Content-Length"] = str(content_length)
if content_range is not None:
headers["Content-Range"] = content_range
self.transport.request_json(
"POST",
f"/commerce/media/v1_beta/video/{video_id}/upload",
scopes=[MEDIA_SCOPE],
headers=headers,
content=content,
)
def create_document(self, payload: CreateDocumentRequest) -> CreateDocumentResponse:
return self.transport.request_model(
CreateDocumentResponse,
"POST",
"/commerce/media/v1_beta/document",
scopes=[MEDIA_SCOPE],
headers={"Content-Type": "application/json"},
json_body=payload.model_dump(by_alias=True, exclude_none=True),
)
def create_upload_and_wait_document(
self,
payload: CreateDocumentRequest,
*,
file_name: str,
content: bytes,
content_type: str = "application/octet-stream",
timeout_seconds: float = 30.0,
poll_interval_seconds: float = 1.0,
) -> DocumentResponse:
created = self.create_document(payload)
document_id = self._require_resource_id(created.documentId, "documentId")
self.upload_document(
document_id,
file_name=file_name,
content=content,
content_type=content_type,
)
return self.wait_for_document(
document_id,
timeout_seconds=timeout_seconds,
poll_interval_seconds=poll_interval_seconds,
)
def create_document_from_url(self, payload: CreateDocumentFromUrlRequest) -> CreateDocumentResponse:
return self.transport.request_model(
CreateDocumentResponse,
"POST",
"/commerce/media/v1_beta/document/create_document_from_url",
scopes=[MEDIA_SCOPE],
headers={"Content-Type": "application/json"},
json_body=payload.model_dump(by_alias=True, exclude_none=True),
)
def create_document_from_url_and_wait(
self,
payload: CreateDocumentFromUrlRequest,
*,
timeout_seconds: float = 30.0,
poll_interval_seconds: float = 1.0,
) -> DocumentResponse:
created = self.create_document_from_url(payload)
document_id = self._require_resource_id(created.documentId, "documentId")
return self.wait_for_document(
document_id,
timeout_seconds=timeout_seconds,
poll_interval_seconds=poll_interval_seconds,
)
def get_document(self, document_id: str) -> DocumentResponse:
return self.transport.request_model(
DocumentResponse,
"GET",
f"/commerce/media/v1_beta/document/{document_id}",
scopes=[MEDIA_SCOPE],
)
def upload_document(
self,
document_id: str,
*,
file_name: str,
content: bytes,
content_type: str = "application/octet-stream",
) -> DocumentResponse:
return self.transport.request_model(
DocumentResponse,
"POST",
f"/commerce/media/v1_beta/document/{document_id}/upload",
scopes=[MEDIA_SCOPE],
files={"file": (file_name, content, content_type)},
)
def wait_for_video(
self,
video_id: str,
*,
success_statuses: set[str] | None = None,
failure_statuses: set[str] | None = None,
timeout_seconds: float = 30.0,
poll_interval_seconds: float = 1.0,
) -> Video:
desired_statuses = success_statuses or {"LIVE"}
terminal_failures = failure_statuses or {"BLOCKED", "PROCESSING_FAILED"}
return self._wait_for_media_state(
fetch=lambda: self.get_video(video_id),
get_status=lambda payload: payload.status,
desired_statuses=desired_statuses,
terminal_failures=terminal_failures,
timeout_seconds=timeout_seconds,
poll_interval_seconds=poll_interval_seconds,
resource_label=f"video {video_id}",
)
def wait_for_document(
self,
document_id: str,
*,
success_statuses: set[str] | None = None,
failure_statuses: set[str] | None = None,
timeout_seconds: float = 30.0,
poll_interval_seconds: float = 1.0,
) -> DocumentResponse:
desired_statuses = success_statuses or {"ACCEPTED"}
terminal_failures = failure_statuses or {"REJECTED"}
return self._wait_for_media_state(
fetch=lambda: self.get_document(document_id),
get_status=lambda payload: payload.documentStatus,
desired_statuses=desired_statuses,
terminal_failures=terminal_failures,
timeout_seconds=timeout_seconds,
poll_interval_seconds=poll_interval_seconds,
resource_label=f"document {document_id}",
)
def _wait_for_media_state(
self,
*,
fetch,
get_status,
desired_statuses: set[str],
terminal_failures: set[str],
timeout_seconds: float,
poll_interval_seconds: float,
resource_label: str,
):
deadline = monotonic() + timeout_seconds
while True:
payload = fetch()
status = get_status(payload)
if status in desired_statuses:
return payload
if status in terminal_failures:
raise ValueError(f"{resource_label} reached terminal failure status: {status}")
if monotonic() >= deadline:
raise TimeoutError(f"Timed out while waiting for {resource_label}; last status was {status!r}")
sleep(poll_interval_seconds)
@staticmethod
def _require_resource_id(value: str | None, field_name: str) -> str:
if not value:
raise RuntimeError(f"eBay did not return a required {field_name}")
return value

6
ebay_client/notification/__init__.py
View file

@ -1,11 +1,8 @@
from ebay_client.notification.client import NotificationClient
from ebay_client.notification.webhook import (
WebhookChallengeHandler,
WebhookDispatchResult,
WebhookEventEnvelope,
WebhookHttpResponse,
WebhookPublicKeyResolver,
WebhookRequestHandler,
WebhookSignatureParser,
WebhookSignatureValidator,
)
@ -13,11 +10,8 @@ from ebay_client.notification.webhook import (
__all__ = [
"NotificationClient",
"WebhookChallengeHandler",
"WebhookDispatchResult",
"WebhookEventEnvelope",
"WebhookHttpResponse",
"WebhookPublicKeyResolver",
"WebhookRequestHandler",
"WebhookSignatureParser",
"WebhookSignatureValidator",
]

119
ebay_client/notification/client.py
View file

@ -2,16 +2,10 @@ from __future__ import annotations
from ebay_client.core.http.transport import ApiTransport
from ebay_client.generated.notification.models import (
Config,
CreateSubscriptionFilterRequest,
CreateSubscriptionRequest,
Destination,
DestinationRequest,
DestinationSearchResponse,
PublicKey,
Subscription,
SubscriptionFilter,
UpdateSubscriptionRequest,
SubscriptionSearchResponse,
Topic,
TopicSearchResponse,
@ -20,33 +14,12 @@ from ebay_client.generated.notification.models import (
NOTIFICATION_SCOPE = "https://api.ebay.com/oauth/api_scope"
NOTIFICATION_SUBSCRIPTION_SCOPE = "https://api.ebay.com/oauth/api_scope/commerce.notification.subscription"
NOTIFICATION_SUBSCRIPTION_READ_SCOPE = "https://api.ebay.com/oauth/api_scope/commerce.notification.subscription.readonly"
NOTIFICATION_SUBSCRIPTION_READ_SCOPE_OPTIONS = [
[NOTIFICATION_SCOPE, NOTIFICATION_SUBSCRIPTION_READ_SCOPE],
[NOTIFICATION_SCOPE, NOTIFICATION_SUBSCRIPTION_SCOPE],
]
class NotificationClient:
def __init__(self, transport: ApiTransport) -> None:
self.transport = transport
def get_config(self) -> Config:
return self.transport.request_model(
Config,
"GET",
"/commerce/notification/v1/config",
scopes=[NOTIFICATION_SCOPE],
)
def update_config(self, payload: Config) -> None:
self.transport.request_json(
"PUT",
"/commerce/notification/v1/config",
scopes=[NOTIFICATION_SCOPE],
headers={"Content-Type": "application/json"},
json_body=payload.model_dump(by_alias=True, exclude_none=True),
)
def get_topics(self, *, limit: int | None = None, continuation_token: str | None = None) -> TopicSearchResponse:
return self.transport.request_model(
TopicSearchResponse,
@ -82,36 +55,12 @@ class NotificationClient:
json_body=payload.model_dump(by_alias=True, exclude_none=True),
)
def get_destination(self, destination_id: str) -> Destination:
return self.transport.request_model(
Destination,
"GET",
f"/commerce/notification/v1/destination/{destination_id}",
scopes=[NOTIFICATION_SCOPE],
)
def update_destination(self, destination_id: str, payload: DestinationRequest) -> None:
self.transport.request_json(
"PUT",
f"/commerce/notification/v1/destination/{destination_id}",
scopes=[NOTIFICATION_SCOPE],
headers={"Content-Type": "application/json"},
json_body=payload.model_dump(by_alias=True, exclude_none=True),
)
def delete_destination(self, destination_id: str) -> None:
self.transport.request_json(
"DELETE",
f"/commerce/notification/v1/destination/{destination_id}",
scopes=[NOTIFICATION_SCOPE],
)
def get_subscriptions(self, *, limit: int | None = None, continuation_token: str | None = None) -> SubscriptionSearchResponse:
return self.transport.request_model(
SubscriptionSearchResponse,
"GET",
"/commerce/notification/v1/subscription",
scope_options=NOTIFICATION_SUBSCRIPTION_READ_SCOPE_OPTIONS,
scopes=[NOTIFICATION_SCOPE, NOTIFICATION_SUBSCRIPTION_READ_SCOPE],
params={"limit": limit, "continuation_token": continuation_token},
)
@ -124,72 +73,6 @@ class NotificationClient:
json_body=payload.model_dump(by_alias=True, exclude_none=True),
)
def get_subscription(self, subscription_id: str) -> Subscription:
return self.transport.request_model(
Subscription,
"GET",
f"/commerce/notification/v1/subscription/{subscription_id}",
scope_options=NOTIFICATION_SUBSCRIPTION_READ_SCOPE_OPTIONS,
)
def update_subscription(self, subscription_id: str, payload: UpdateSubscriptionRequest) -> None:
self.transport.request_json(
"PUT",
f"/commerce/notification/v1/subscription/{subscription_id}",
scopes=[NOTIFICATION_SCOPE, NOTIFICATION_SUBSCRIPTION_SCOPE],
headers={"Content-Type": "application/json"},
json_body=payload.model_dump(by_alias=True, exclude_none=True),
)
def delete_subscription(self, subscription_id: str) -> None:
self.transport.request_json(
"DELETE",
f"/commerce/notification/v1/subscription/{subscription_id}",
scopes=[NOTIFICATION_SCOPE, NOTIFICATION_SUBSCRIPTION_SCOPE],
)
def create_subscription_filter(
self,
subscription_id: str,
payload: CreateSubscriptionFilterRequest,
) -> dict[str, object] | None:
return self.transport.request_json(
"POST",
f"/commerce/notification/v1/subscription/{subscription_id}/filter",
scopes=[NOTIFICATION_SCOPE, NOTIFICATION_SUBSCRIPTION_SCOPE],
headers={"Content-Type": "application/json"},
json_body=payload.model_dump(by_alias=True, exclude_none=True),
)
def get_subscription_filter(self, subscription_id: str, filter_id: str) -> SubscriptionFilter:
return self.transport.request_model(
SubscriptionFilter,
"GET",
f"/commerce/notification/v1/subscription/{subscription_id}/filter/{filter_id}",
scope_options=NOTIFICATION_SUBSCRIPTION_READ_SCOPE_OPTIONS,
)
def delete_subscription_filter(self, subscription_id: str, filter_id: str) -> None:
self.transport.request_json(
"DELETE",
f"/commerce/notification/v1/subscription/{subscription_id}/filter/{filter_id}",
scopes=[NOTIFICATION_SCOPE, NOTIFICATION_SUBSCRIPTION_SCOPE],
)
def disable_subscription(self, subscription_id: str) -> None:
self.transport.request_json(
"POST",
f"/commerce/notification/v1/subscription/{subscription_id}/disable",
scopes=[NOTIFICATION_SCOPE, NOTIFICATION_SUBSCRIPTION_SCOPE],
)
def enable_subscription(self, subscription_id: str) -> None:
self.transport.request_json(
"POST",
f"/commerce/notification/v1/subscription/{subscription_id}/enable",
scopes=[NOTIFICATION_SCOPE, NOTIFICATION_SUBSCRIPTION_SCOPE],
)
def test_subscription(self, subscription_id: str) -> None:
self.transport.request_json(
"POST",

176
ebay_client/notification/webhook.py
View file

@ -2,7 +2,6 @@ from __future__ import annotations
import base64
import hashlib
import json
from collections.abc import Callable
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
@ -11,7 +10,6 @@ from typing import Any
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives.serialization import load_der_public_key
from pydantic import BaseModel, Field
from ebay_client.generated.notification.models import PublicKey
@ -24,37 +22,6 @@ class WebhookEventEnvelope(BaseModel):
topic_id: str | None = None
data: dict[str, Any] | list[Any] | None = None
@classmethod
def from_payload(cls, payload: dict[str, Any] | list[Any]) -> WebhookEventEnvelope:
if isinstance(payload, list):
return cls(data=payload)
metadata = dict(payload.get("metadata") or {}) if isinstance(payload.get("metadata"), dict) else {}
for key, value in payload.items():
if key not in {"metadata", "notificationId", "notification_id", "publishDate", "publish_date", "topicId", "topic_id", "data"}:
metadata[key] = value
return cls(
metadata=metadata,
notification_id=payload.get("notificationId") or payload.get("notification_id"),
publish_date=payload.get("publishDate") or payload.get("publish_date"),
topic_id=payload.get("topicId") or payload.get("topic_id"),
data=payload.get("data"),
)
@dataclass(slots=True)
class WebhookHttpResponse:
status_code: int
headers: dict[str, str]
body: bytes = b""
@dataclass(slots=True)
class WebhookDispatchResult:
response: WebhookHttpResponse
event: WebhookEventEnvelope | None = None
@dataclass(slots=True)
class ParsedSignatureHeader:
@ -72,28 +39,6 @@ class WebhookSignatureParser:
DIGEST_ALIASES = ("digest",)
def parse(self, header_value: str) -> ParsedSignatureHeader:
parts = self._parse_parts(header_value)
signature_value = self._first_match(parts, self.SIGNATURE_ALIASES)
signature = self._decode_base64(signature_value) if signature_value else None
return ParsedSignatureHeader(
key_id=self._first_match(parts, self.KEY_ALIASES),
signature=signature,
algorithm=self._first_match(parts, self.ALGORITHM_ALIASES),
digest=self._first_match(parts, self.DIGEST_ALIASES),
raw_parts=parts,
)
def _parse_parts(self, header_value: str) -> dict[str, str]:
decoded = self._decode_base64_to_text(header_value)
if decoded:
try:
payload = json.loads(decoded)
except json.JSONDecodeError:
payload = None
if isinstance(payload, dict):
return {str(key).strip().lower(): str(value).strip() for key, value in payload.items() if value is not None}
separators = header_value.replace(";", ",").split(",")
parts: dict[str, str] = {}
for item in separators:
@ -101,31 +46,16 @@ class WebhookSignatureParser:
continue
key, value = item.split("=", 1)
parts[key.strip().lower()] = value.strip().strip('"')
return parts
@staticmethod
def _decode_base64_to_text(value: str) -> str | None:
decoded = WebhookSignatureParser._decode_base64(value)
if decoded is None:
return None
try:
return decoded.decode("utf-8")
except UnicodeDecodeError:
return None
@staticmethod
def _decode_base64(value: str) -> bytes | None:
normalized = value.strip()
if not normalized:
return None
padding = (-len(normalized)) % 4
normalized = normalized + ("=" * padding)
for decoder in (base64.b64decode, base64.urlsafe_b64decode):
try:
return decoder(normalized)
except Exception:
continue
return None
signature_value = self._first_match(parts, self.SIGNATURE_ALIASES)
signature = base64.b64decode(signature_value) if signature_value else None
return ParsedSignatureHeader(
key_id=self._first_match(parts, self.KEY_ALIASES),
signature=signature,
algorithm=self._first_match(parts, self.ALGORITHM_ALIASES),
digest=self._first_match(parts, self.DIGEST_ALIASES),
raw_parts=parts,
)
@staticmethod
def _first_match(parts: dict[str, str], aliases: tuple[str, ...]) -> str | None:
@ -177,12 +107,10 @@ class WebhookSignatureValidator:
if not parsed.key_id or not parsed.signature:
return False
key_payload = self.resolver.resolve(parsed.key_id)
key_value = key_payload.key
if not key_value:
return False
public_key = self._load_public_key(key_value)
if public_key is None:
pem = key_payload.key
if not pem:
return False
public_key = serialization.load_pem_public_key(pem.encode("utf-8"))
digest_name = (parsed.digest or key_payload.digest or "SHA256").upper()
digest = self._hash_algorithm(digest_name)
message = message_builder(body) if message_builder else body
@ -202,83 +130,3 @@ class WebhookSignatureValidator:
if normalized == "SHA512":
return hashes.SHA512()
return hashes.SHA256()
@staticmethod
def _load_public_key(key_value: str):
normalized = key_value.strip()
if not normalized:
return None
if "BEGIN PUBLIC KEY" in normalized:
return serialization.load_pem_public_key(normalized.encode("utf-8"))
decoded = WebhookSignatureParser._decode_base64(normalized)
if decoded is None:
return None
return load_der_public_key(decoded)
class WebhookRequestHandler:
def __init__(
self,
*,
signature_validator: WebhookSignatureValidator,
challenge_handler: WebhookChallengeHandler | None = None,
event_parser: Callable[[bytes], WebhookEventEnvelope] | None = None,
) -> None:
self.signature_validator = signature_validator
self.challenge_handler = challenge_handler or WebhookChallengeHandler()
self.event_parser = event_parser or self.parse_event
def handle_challenge(self, *, challenge_code: str | None, verification_token: str, endpoint: str) -> WebhookHttpResponse:
if not challenge_code:
return self._json_response(400, {"error": "challenge_code is required"})
payload = self.challenge_handler.build_response_body(
challenge_code=challenge_code,
verification_token=verification_token,
endpoint=endpoint,
)
return self._json_response(200, payload)
def handle_notification(
self,
*,
signature_header: str | None,
body: bytes,
message_builder: Callable[[bytes], bytes] | None = None,
) -> WebhookDispatchResult:
if not signature_header or not self.signature_validator.validate(
header_value=signature_header,
body=body,
message_builder=message_builder,
):
return WebhookDispatchResult(
response=WebhookHttpResponse(status_code=412, headers={}),
)
try:
event = self.event_parser(body)
except (UnicodeDecodeError, ValueError, TypeError):
return WebhookDispatchResult(
response=self._json_response(400, {"error": "notification body is not valid JSON"}),
)
return WebhookDispatchResult(
response=WebhookHttpResponse(status_code=200, headers={}),
event=event,
)
@staticmethod
def parse_event(body: bytes) -> WebhookEventEnvelope:
payload = json.loads(body.decode("utf-8"))
if isinstance(payload, dict) or isinstance(payload, list):
return WebhookEventEnvelope.from_payload(payload)
raise ValueError("notification payload must be a JSON object or array")
@staticmethod
def _json_response(status_code: int, payload: dict[str, Any]) -> WebhookHttpResponse:
return WebhookHttpResponse(
status_code=status_code,
headers={"Content-Type": "application/json"},
body=json.dumps(payload, separators=(",", ":")).encode("utf-8"),
)

71
examples/fastapi_notification_webhook.py
View file

@ -1,71 +0,0 @@
from __future__ import annotations
import os
from fastapi import FastAPI, Header, Request, Response
from ebay_client.core.auth.models import EbayOAuthConfig
from ebay_client.client import EbayClient
from ebay_client.notification import WebhookPublicKeyResolver, WebhookRequestHandler, WebhookSignatureValidator
app = FastAPI()
oauth_config = EbayOAuthConfig(
client_id=os.environ["EBAY_CLIENT_ID"],
client_secret=os.environ["EBAY_CLIENT_SECRET"],
default_scopes=["https://api.ebay.com/oauth/api_scope"],
)
ebay_client = EbayClient(oauth_config)
verification_token = os.environ["EBAY_NOTIFICATION_VERIFICATION_TOKEN"]
public_key_resolver = WebhookPublicKeyResolver(ebay_client.notification.get_public_key)
webhook_handler = WebhookRequestHandler(
signature_validator=WebhookSignatureValidator(public_key_resolver)
)
def process_event(topic_id: str | None, payload: dict | list | None) -> None:
if topic_id == "MARKETPLACE_ACCOUNT_DELETION":
return
@app.get("/webhooks/ebay")
async def ebay_challenge(challenge_code: str | None = None, request: Request | None = None) -> Response:
if request is None:
return Response(status_code=500)
endpoint = str(request.url.replace(query=""))
result = webhook_handler.handle_challenge(
challenge_code=challenge_code,
verification_token=verification_token,
endpoint=endpoint,
)
return Response(
content=result.body,
status_code=result.status_code,
headers=result.headers,
media_type=result.headers.get("Content-Type"),
)
@app.post("/webhooks/ebay")
async def ebay_notification(
request: Request,
x_ebay_signature: str | None = Header(default=None, alias="X-EBAY-SIGNATURE"),
) -> Response:
body = await request.body()
result = webhook_handler.handle_notification(
signature_header=x_ebay_signature,
body=body,
)
if result.event is not None:
process_event(result.event.topic_id, result.event.data)
return Response(
content=result.response.body,
status_code=result.response.status_code,
headers=result.response.headers,
media_type=result.response.headers.get("Content-Type"),
)

86
examples/media_workflows.py
View file

@ -1,86 +0,0 @@
from __future__ import annotations
import os
from pathlib import Path
from ebay_client.client import EbayClient
from ebay_client.core.auth.models import EbayOAuthConfig
from ebay_client.generated.media.models import (
CreateDocumentRequest,
CreateImageFromUrlRequest,
CreateVideoRequest,
)
def build_client() -> EbayClient:
oauth_config = EbayOAuthConfig(
client_id=os.environ["EBAY_CLIENT_ID"],
client_secret=os.environ["EBAY_CLIENT_SECRET"],
default_scopes=["https://api.ebay.com/oauth/api_scope/sell.inventory"],
)
return EbayClient(oauth_config)
def upload_image_from_file(client: EbayClient, image_path: Path) -> None:
image = client.media.create_image_from_file(
file_name=image_path.name,
content=image_path.read_bytes(),
content_type="image/jpeg",
)
print("image_url:", image.imageUrl)
def upload_image_from_url(client: EbayClient, image_url: str) -> None:
image = client.media.create_image_from_url(CreateImageFromUrlRequest(imageUrl=image_url))
print("image_url:", image.imageUrl)
def upload_document_and_wait(client: EbayClient, document_path: Path) -> None:
accepted = client.media.create_upload_and_wait_document(
CreateDocumentRequest(
documentType="USER_GUIDE_OR_MANUAL",
languages=["en-US"],
),
file_name=document_path.name,
content=document_path.read_bytes(),
content_type="application/pdf",
timeout_seconds=60.0,
)
print("document_final_status:", accepted.documentStatus)
def upload_video_and_wait(client: EbayClient, video_path: Path) -> None:
live_video = client.media.create_upload_and_wait_video(
CreateVideoRequest(
title=video_path.stem,
size=video_path.stat().st_size,
classification=["ITEM"],
description="Example upload from the ebay-rest-client workspace.",
),
content=video_path.read_bytes(),
timeout_seconds=120.0,
)
print("video_status:", live_video.status)
print("video_id:", live_video.videoId)
def main() -> None:
client = build_client()
image_file = os.environ.get("EBAY_MEDIA_IMAGE_FILE")
image_url = os.environ.get("EBAY_MEDIA_IMAGE_URL")
document_file = os.environ.get("EBAY_MEDIA_DOCUMENT_FILE")
video_file = os.environ.get("EBAY_MEDIA_VIDEO_FILE")
if image_file:
upload_image_from_file(client, Path(image_file))
if image_url:
upload_image_from_url(client, image_url)
if document_file:
upload_document_and_wait(client, Path(document_file))
if video_file:
upload_video_and_wait(client, Path(video_file))
if __name__ == "__main__":
main()

17
scripts/generate_clients.py
View file

@ -20,11 +20,6 @@ class ApiSpec:
API_SPECS = {
"media": ApiSpec(
name="media",
spec_path=ROOT / "commerce_media_v1_beta_oas3.yaml",
output_path=GENERATED_ROOT / "media",
),
"notification": ApiSpec(
name="notification",
spec_path=ROOT / "commerce_notification_v1_oas3.yaml",
@ -95,24 +90,12 @@ def run_generation(spec: ApiSpec, *, fail_on_warning: bool) -> None:
command.append("--disable-warnings")
subprocess.run(command, check=True, cwd=str(ROOT))
normalize_generated_module(spec.output_path / "models.py")
(spec.output_path / "__init__.py").write_text(
'"""Generated Pydantic models from the OpenAPI contract."""\n\nfrom .models import *\n',
encoding="utf-8",
)
def normalize_generated_module(file_path: Path) -> None:
raw_bytes = file_path.read_bytes()
try:
content = raw_bytes.decode("utf-8")
except UnicodeDecodeError:
content = raw_bytes.decode("cp1252")
content = content.replace("\u00a0", " ")
file_path.write_text(content, encoding="utf-8")
def main() -> int:
args = parse_args()
specs = [API_SPECS[args.api]] if args.api else [API_SPECS[name] for name in sorted(API_SPECS)]

16
tests/test_auth_oauth.py
View file

@ -33,19 +33,3 @@ def test_get_valid_token_reuses_unexpired_token() -> None:
token = client.get_valid_token(scopes=["scope.a"])
assert token.access_token == "cached-token"
def test_get_valid_token_reuses_token_when_any_scope_option_matches() -> None:
config = EbayOAuthConfig(client_id="client-id", client_secret="client-secret")
store = InMemoryTokenStore()
store.set_token(OAuthToken(access_token="cached-token", scope="scope.base scope.write"))
client = EbayOAuthClient(config, token_store=store)
token = client.get_valid_token(
scope_options=[
["scope.base", "scope.read"],
["scope.base", "scope.write"],
]
)
assert token.access_token == "cached-token"

168
tests/test_notification_webhook.py
View file

@ -1,18 +1,6 @@
from __future__ import annotations
import base64
import json
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import ec
from ebay_client.generated.notification.models import PublicKey
from ebay_client.notification.webhook import WebhookChallengeHandler, WebhookSignatureParser
from ebay_client.notification.webhook import (
WebhookPublicKeyResolver,
WebhookRequestHandler,
WebhookSignatureValidator,
)
def test_challenge_handler_builds_sha256_response() -> None:
@ -33,158 +21,4 @@ def test_signature_parser_extracts_known_fields() -> None:
assert parsed.key_id == "public-key-1"
assert parsed.algorithm == "ECDSA"
assert parsed.digest == "SHA256"
assert parsed.signature == b"foo"
def test_signature_parser_decodes_base64_json_header() -> None:
parser = WebhookSignatureParser()
header = base64.b64encode(
json.dumps(
{
"alg": "ECDSA",
"kid": "public-key-1",
"signature": base64.b64encode(b"signed-bytes").decode("ascii"),
"digest": "SHA256",
}
).encode("utf-8")
).decode("ascii")
parsed = parser.parse(header)
assert parsed.key_id == "public-key-1"
assert parsed.algorithm == "ECDSA"
assert parsed.digest == "SHA256"
assert parsed.signature == b"signed-bytes"
def test_signature_validator_verifies_base64_json_header_and_der_key() -> None:
private_key = ec.generate_private_key(ec.SECP256R1())
public_key = private_key.public_key()
public_key_der = public_key.public_bytes(
encoding=serialization.Encoding.DER,
format=serialization.PublicFormat.SubjectPublicKeyInfo,
)
body = b'{"notificationId":"abc-123"}'
signature = private_key.sign(body, ec.ECDSA(hashes.SHA256()))
header = base64.b64encode(
json.dumps(
{
"alg": "ECDSA",
"kid": "public-key-1",
"signature": base64.b64encode(signature).decode("ascii"),
"digest": "SHA256",
}
).encode("utf-8")
).decode("ascii")
resolver = WebhookPublicKeyResolver(
lambda key_id: PublicKey(
key=base64.b64encode(public_key_der).decode("ascii"),
algorithm="ECDSA",
digest="SHA256",
)
)
validator = WebhookSignatureValidator(resolver)
assert validator.validate(header_value=header, body=body) is True
def test_request_handler_returns_json_challenge_response() -> None:
handler = WebhookRequestHandler(
signature_validator=WebhookSignatureValidator(
WebhookPublicKeyResolver(lambda _: PublicKey(key="", algorithm="ECDSA", digest="SHA256"))
)
)
response = handler.handle_challenge(
challenge_code="challenge",
verification_token="verification",
endpoint="https://example.test/webhook",
)
assert response.status_code == 200
assert response.headers["Content-Type"] == "application/json"
body = json.loads(response.body.decode("utf-8"))
assert "challengeResponse" in body
def test_request_handler_accepts_verified_notification_and_parses_event() -> None:
private_key = ec.generate_private_key(ec.SECP256R1())
public_key = private_key.public_key()
public_key_der = public_key.public_bytes(
encoding=serialization.Encoding.DER,
format=serialization.PublicFormat.SubjectPublicKeyInfo,
)
body = json.dumps(
{
"notificationId": "abc-123",
"publishDate": "2026-04-07T00:00:00.000Z",
"topicId": "MARKETPLACE_ACCOUNT_DELETION",
"data": {"userId": "user-1"},
"schemaVersion": "1.0",
}
).encode("utf-8")
signature = private_key.sign(body, ec.ECDSA(hashes.SHA256()))
header = base64.b64encode(
json.dumps(
{
"alg": "ECDSA",
"kid": "public-key-1",
"signature": base64.b64encode(signature).decode("ascii"),
"digest": "SHA256",
}
).encode("utf-8")
).decode("ascii")
resolver = WebhookPublicKeyResolver(
lambda key_id: PublicKey(
key=base64.b64encode(public_key_der).decode("ascii"),
algorithm="ECDSA",
digest="SHA256",
)
)
handler = WebhookRequestHandler(signature_validator=WebhookSignatureValidator(resolver))
result = handler.handle_notification(signature_header=header, body=body)
assert result.response.status_code == 200
assert result.event is not None
assert result.event.notification_id == "abc-123"
assert result.event.topic_id == "MARKETPLACE_ACCOUNT_DELETION"
assert result.event.metadata["schemaVersion"] == "1.0"
def test_request_handler_returns_412_for_invalid_signature() -> None:
private_key = ec.generate_private_key(ec.SECP256R1())
public_key = private_key.public_key()
public_key_der = public_key.public_bytes(
encoding=serialization.Encoding.DER,
format=serialization.PublicFormat.SubjectPublicKeyInfo,
)
body = b'{"notificationId":"abc-123"}'
wrong_body = b'{"notificationId":"def-456"}'
signature = private_key.sign(body, ec.ECDSA(hashes.SHA256()))
header = base64.b64encode(
json.dumps(
{
"alg": "ECDSA",
"kid": "public-key-1",
"signature": base64.b64encode(signature).decode("ascii"),
"digest": "SHA256",
}
).encode("utf-8")
).decode("ascii")
resolver = WebhookPublicKeyResolver(
lambda key_id: PublicKey(
key=base64.b64encode(public_key_der).decode("ascii"),
algorithm="ECDSA",
digest="SHA256",
)
)
handler = WebhookRequestHandler(signature_validator=WebhookSignatureValidator(resolver))
result = handler.handle_notification(signature_header=header, body=wrong_body)
assert result.response.status_code == 412
assert result.event is None
assert parsed.signature == b"foo"

723
tests/test_public_wrappers.py
View file

@ -1,723 +0,0 @@
from __future__ import annotations
import json
from pytest_httpx import HTTPXMock
from ebay_client.account.client import AccountClient
from ebay_client.core.auth.models import OAuthToken
from ebay_client.core.http.transport import ApiTransport
from ebay_client.feed.client import FeedClient
from ebay_client.fulfillment.client import FulfillmentClient
from ebay_client.generated.account.models import Programs
from ebay_client.generated.feed.models import TaskCollection
from ebay_client.generated.fulfillment.models import Order
from ebay_client.generated.inventory.models import InventoryItemWithSkuLocaleGroupid
from ebay_client.generated.media.models import (
CreateDocumentFromUrlRequest,
CreateDocumentRequest,
CreateDocumentResponse,
CreateImageFromUrlRequest,
CreateVideoRequest,
DocumentResponse,
ImageResponse,
Video,
)
from ebay_client.generated.notification.models import (
Config,
CreateSubscriptionFilterRequest,
CreateSubscriptionRequest,
DeliveryConfig,
Subscription,
SubscriptionFilter,
SubscriptionPayloadDetail,
DestinationRequest,
TopicSearchResponse,
UpdateSubscriptionRequest,
)
from ebay_client.inventory.client import InventoryClient
from ebay_client.media.client import CreatedMediaResource, MediaClient, extract_resource_id
from ebay_client.notification.client import NotificationClient
class DummyOAuthClient:
def get_valid_token(
self,
*,
scopes: list[str] | None = None,
scope_options: list[list[str]] | None = None,
) -> OAuthToken:
if scopes:
resolved_scopes = scopes
elif scope_options:
resolved_scopes = scope_options[0]
else:
resolved_scopes = []
return OAuthToken(access_token="test-token", scope=" ".join(resolved_scopes))
def build_transport() -> ApiTransport:
return ApiTransport(base_url="https://api.ebay.com", oauth_client=DummyOAuthClient())
class RecordingOAuthClient:
def __init__(self) -> None:
self.last_scopes: list[str] | None = None
self.last_scope_options: list[list[str]] | None = None
def get_valid_token(
self,
*,
scopes: list[str] | None = None,
scope_options: list[list[str]] | None = None,
) -> OAuthToken:
self.last_scopes = scopes
self.last_scope_options = scope_options
resolved_scopes = scopes or (scope_options[0] if scope_options else [])
return OAuthToken(access_token="recorded-token", scope=" ".join(resolved_scopes))
def test_notification_wrapper_returns_pydantic_model(httpx_mock: HTTPXMock) -> None:
httpx_mock.add_response(
method="GET",
url="https://api.ebay.com/commerce/notification/v1/topic?limit=10",
json={"topics": [{"topicId": "MARKETPLACE_ACCOUNT_DELETION", "description": "topic"}], "total": 1},
)
client = NotificationClient(build_transport())
result = client.get_topics(limit=10)
assert isinstance(result, TopicSearchResponse)
assert result.total == 1
assert result.topics and result.topics[0].topicId == "MARKETPLACE_ACCOUNT_DELETION"
request = httpx_mock.get_requests()[0]
assert request.headers["Authorization"] == "Bearer test-token"
def test_notification_wrapper_serializes_typed_request_model(httpx_mock: HTTPXMock) -> None:
httpx_mock.add_response(
method="POST",
url="https://api.ebay.com/commerce/notification/v1/destination",
json={},
status_code=201,
)
client = NotificationClient(build_transport())
payload = DestinationRequest(
name="main-destination",
status="ENABLED",
deliveryConfig=DeliveryConfig(
endpoint="https://example.test/webhook",
verificationToken="verification_token_1234567890123456",
),
)
client.create_destination(payload)
request = httpx_mock.get_requests()[0]
body = json.loads(request.content.decode("utf-8"))
assert body["name"] == "main-destination"
assert body["deliveryConfig"]["endpoint"] == "https://example.test/webhook"
def test_notification_config_wrapper_round_trip(httpx_mock: HTTPXMock) -> None:
httpx_mock.add_response(
method="GET",
url="https://api.ebay.com/commerce/notification/v1/config",
json={"alertEmail": "alerts@example.test"},
)
httpx_mock.add_response(
method="PUT",
url="https://api.ebay.com/commerce/notification/v1/config",
status_code=204,
)
client = NotificationClient(build_transport())
config = client.get_config()
client.update_config(Config(alertEmail="ops@example.test"))
assert isinstance(config, Config)
assert config.alertEmail == "alerts@example.test"
request = httpx_mock.get_requests()[1]
body = json.loads(request.content.decode("utf-8"))
assert body["alertEmail"] == "ops@example.test"
def test_notification_subscription_wrapper_serializes_and_returns_models(httpx_mock: HTTPXMock) -> None:
httpx_mock.add_response(
method="POST",
url="https://api.ebay.com/commerce/notification/v1/subscription",
json={},
status_code=201,
)
httpx_mock.add_response(
method="GET",
url="https://api.ebay.com/commerce/notification/v1/subscription/SUB-1",
json={"subscriptionId": "SUB-1", "topicId": "TOPIC-1", "status": "ENABLED"},
)
httpx_mock.add_response(
method="PUT",
url="https://api.ebay.com/commerce/notification/v1/subscription/SUB-1",
status_code=204,
)
httpx_mock.add_response(
method="POST",
url="https://api.ebay.com/commerce/notification/v1/subscription/SUB-1/enable",
status_code=204,
)
httpx_mock.add_response(
method="POST",
url="https://api.ebay.com/commerce/notification/v1/subscription/SUB-1/test",
status_code=202,
)
client = NotificationClient(build_transport())
create_payload = CreateSubscriptionRequest(
destinationId="DEST-1",
topicId="TOPIC-1",
status="DISABLED",
payload=SubscriptionPayloadDetail(
deliveryProtocol="HTTPS",
format="JSON",
schemaVersion="1.0",
),
)
update_payload = UpdateSubscriptionRequest(status="ENABLED", destinationId="DEST-1")
client.create_subscription(create_payload)
subscription = client.get_subscription("SUB-1")
client.update_subscription("SUB-1", update_payload)
client.enable_subscription("SUB-1")
client.test_subscription("SUB-1")
assert isinstance(subscription, Subscription)
assert subscription.subscriptionId == "SUB-1"
create_request = httpx_mock.get_requests()[0]
create_body = json.loads(create_request.content.decode("utf-8"))
assert create_body["destinationId"] == "DEST-1"
assert create_body["payload"]["deliveryProtocol"] == "HTTPS"
update_request = httpx_mock.get_requests()[2]
update_body = json.loads(update_request.content.decode("utf-8"))
assert update_body == {"destinationId": "DEST-1", "status": "ENABLED"}
def test_notification_subscription_filter_wrapper_serializes_and_returns_model(httpx_mock: HTTPXMock) -> None:
httpx_mock.add_response(
method="POST",
url="https://api.ebay.com/commerce/notification/v1/subscription/SUB-1/filter",
json={},
status_code=201,
)
httpx_mock.add_response(
method="GET",
url="https://api.ebay.com/commerce/notification/v1/subscription/SUB-1/filter/FILTER-1",
json={"filterId": "FILTER-1", "subscriptionId": "SUB-1", "filterStatus": "ENABLED"},
)
httpx_mock.add_response(
method="DELETE",
url="https://api.ebay.com/commerce/notification/v1/subscription/SUB-1/filter/FILTER-1",
status_code=204,
)
client = NotificationClient(build_transport())
payload = CreateSubscriptionFilterRequest(
filterSchema={
"properties": {
"data": {
"type": "object",
}
}
}
)
client.create_subscription_filter("SUB-1", payload)
subscription_filter = client.get_subscription_filter("SUB-1", "FILTER-1")
client.delete_subscription_filter("SUB-1", "FILTER-1")
assert isinstance(subscription_filter, SubscriptionFilter)
assert subscription_filter.filterId == "FILTER-1"
request = httpx_mock.get_requests()[0]
body = json.loads(request.content.decode("utf-8"))
assert body["filterSchema"]["properties"]["data"]["type"] == "object"
def test_inventory_wrapper_returns_inventory_item_model(httpx_mock: HTTPXMock) -> None:
httpx_mock.add_response(
method="GET",
url="https://api.ebay.com/sell/inventory/v1/inventory_item/SKU-1",
json={"sku": "SKU-1"},
)
client = InventoryClient(build_transport())
result = client.get_inventory_item("SKU-1")
assert isinstance(result, InventoryItemWithSkuLocaleGroupid)
assert result.sku == "SKU-1"
def test_fulfillment_wrapper_returns_order_model(httpx_mock: HTTPXMock) -> None:
httpx_mock.add_response(
method="GET",
url="https://api.ebay.com/sell/fulfillment/v1/order/ORDER-1",
json={"orderId": "ORDER-1"},
)
client = FulfillmentClient(build_transport())
result = client.get_order("ORDER-1")
assert isinstance(result, Order)
assert result.orderId == "ORDER-1"
def test_account_wrapper_returns_programs_model(httpx_mock: HTTPXMock) -> None:
httpx_mock.add_response(
method="GET",
url="https://api.ebay.com/sell/account/v1/program/get_opted_in_programs",
json={"programs": [{"programType": "OUT_OF_STOCK_CONTROL"}]},
)
client = AccountClient(build_transport())
result = client.get_opted_in_programs()
assert isinstance(result, Programs)
assert result.programs and result.programs[0].programType == "OUT_OF_STOCK_CONTROL"
def test_feed_wrapper_returns_task_collection_model(httpx_mock: HTTPXMock) -> None:
httpx_mock.add_response(
method="GET",
url="https://api.ebay.com/sell/feed/v1/task?feed_type=LMS_ORDER_REPORT",
json={"tasks": [{"taskId": "TASK-1"}], "total": 1},
)
client = FeedClient(build_transport())
result = client.get_tasks(feed_type="LMS_ORDER_REPORT")
assert isinstance(result, TaskCollection)
assert result.total == 1
assert result.tasks and result.tasks[0].taskId == "TASK-1"
def test_inventory_wrapper_accepts_readonly_or_full_scope_options(httpx_mock: HTTPXMock) -> None:
oauth_client = RecordingOAuthClient()
transport = ApiTransport(base_url="https://api.ebay.com", oauth_client=oauth_client)
httpx_mock.add_response(
method="GET",
url="https://api.ebay.com/sell/inventory/v1/inventory_item/SKU-1",
json={"sku": "SKU-1"},
)
InventoryClient(transport).get_inventory_item("SKU-1")
assert oauth_client.last_scopes is None
assert oauth_client.last_scope_options == [
["https://api.ebay.com/oauth/api_scope/sell.inventory.readonly"],
["https://api.ebay.com/oauth/api_scope/sell.inventory"],
]
def test_fulfillment_wrapper_accepts_readonly_or_full_scope_options(httpx_mock: HTTPXMock) -> None:
oauth_client = RecordingOAuthClient()
transport = ApiTransport(base_url="https://api.ebay.com", oauth_client=oauth_client)
httpx_mock.add_response(
method="GET",
url="https://api.ebay.com/sell/fulfillment/v1/order/ORDER-1",
json={"orderId": "ORDER-1"},
)
FulfillmentClient(transport).get_order("ORDER-1")
assert oauth_client.last_scopes is None
assert oauth_client.last_scope_options == [
["https://api.ebay.com/oauth/api_scope/sell.fulfillment.readonly"],
["https://api.ebay.com/oauth/api_scope/sell.fulfillment"],
]
def test_account_wrapper_accepts_readonly_or_full_scope_options(httpx_mock: HTTPXMock) -> None:
oauth_client = RecordingOAuthClient()
transport = ApiTransport(base_url="https://api.ebay.com", oauth_client=oauth_client)
httpx_mock.add_response(
method="GET",
url="https://api.ebay.com/sell/account/v1/privilege",
json={},
)
AccountClient(transport).get_privileges()
assert oauth_client.last_scopes is None
assert oauth_client.last_scope_options == [
["https://api.ebay.com/oauth/api_scope/sell.account.readonly"],
["https://api.ebay.com/oauth/api_scope/sell.account"],
]
def test_feed_wrapper_accepts_any_documented_feed_scope_option(httpx_mock: HTTPXMock) -> None:
oauth_client = RecordingOAuthClient()
transport = ApiTransport(base_url="https://api.ebay.com", oauth_client=oauth_client)
httpx_mock.add_response(
method="GET",
url="https://api.ebay.com/sell/feed/v1/task?feed_type=LMS_ORDER_REPORT",
json={"tasks": [{"taskId": "TASK-1"}], "total": 1},
)
FeedClient(transport).get_tasks(feed_type="LMS_ORDER_REPORT")
assert oauth_client.last_scopes is None
assert oauth_client.last_scope_options == [
["https://api.ebay.com/oauth/api_scope/sell.inventory"],
["https://api.ebay.com/oauth/api_scope/sell.fulfillment"],
["https://api.ebay.com/oauth/api_scope/sell.marketing"],
["https://api.ebay.com/oauth/api_scope/commerce.catalog.readonly"],
["https://api.ebay.com/oauth/api_scope/sell.analytics.readonly"],
]
def test_media_wrapper_returns_image_model_from_url(httpx_mock: HTTPXMock) -> None:
httpx_mock.add_response(
method="POST",
url="https://api.ebay.com/commerce/media/v1_beta/image/create_image_from_url",
json={"imageUrl": "https://i.ebayimg.com/images/g/demo.jpg", "expirationDate": "2026-12-31T00:00:00Z"},
status_code=201,
)
client = MediaClient(build_transport())
result = client.create_image_from_url(
CreateImageFromUrlRequest(imageUrl="https://example.test/demo.jpg")
)
assert isinstance(result, ImageResponse)
assert result.imageUrl == "https://i.ebayimg.com/images/g/demo.jpg"
request = httpx_mock.get_requests()[0]
body = json.loads(request.content.decode("utf-8"))
assert body["imageUrl"] == "https://example.test/demo.jpg"
def test_media_wrapper_serializes_multipart_image_upload(httpx_mock: HTTPXMock) -> None:
httpx_mock.add_response(
method="POST",
url="https://api.ebay.com/commerce/media/v1_beta/image/create_image_from_file",
json={"imageUrl": "https://i.ebayimg.com/images/g/uploaded.jpg"},
status_code=201,
)
client = MediaClient(build_transport())
result = client.create_image_from_file(
file_name="demo.jpg",
content=b"binary-image-content",
content_type="image/jpeg",
)
assert isinstance(result, ImageResponse)
request = httpx_mock.get_requests()[0]
assert request.headers["Content-Type"].startswith("multipart/form-data;")
assert b"name=\"image\"" in request.content
assert b"filename=\"demo.jpg\"" in request.content
assert b"binary-image-content" in request.content
def test_media_wrapper_returns_created_resource_location_for_video(httpx_mock: HTTPXMock) -> None:
httpx_mock.add_response(
method="POST",
url="https://api.ebay.com/commerce/media/v1_beta/video",
status_code=201,
headers={"Location": "https://api.ebay.com/commerce/media/v1_beta/video/VIDEO-1"},
)
client = MediaClient(build_transport())
result = client.create_video(
CreateVideoRequest(title="Demo", size=1024, classification=["ITEM"])
)
assert isinstance(result, CreatedMediaResource)
assert result.location == "https://api.ebay.com/commerce/media/v1_beta/video/VIDEO-1"
assert result.resource_id == "VIDEO-1"
def test_extract_media_resource_id_handles_location_header() -> None:
assert extract_resource_id("https://api.ebay.com/commerce/media/v1_beta/video/VIDEO-1") == "VIDEO-1"
assert extract_resource_id("https://api.ebay.com/commerce/media/v1_beta/document/DOC-1/") == "DOC-1"
assert extract_resource_id(None) is None
def test_media_wrapper_returns_video_model(httpx_mock: HTTPXMock) -> None:
httpx_mock.add_response(
method="GET",
url="https://api.ebay.com/commerce/media/v1_beta/video/VIDEO-1",
json={"videoId": "VIDEO-1", "status": "LIVE", "title": "Demo"},
)
client = MediaClient(build_transport())
result = client.get_video("VIDEO-1")
assert isinstance(result, Video)
assert result.videoId == "VIDEO-1"
def test_media_wrapper_uploads_video_bytes(httpx_mock: HTTPXMock) -> None:
httpx_mock.add_response(
method="POST",
url="https://api.ebay.com/commerce/media/v1_beta/video/VIDEO-1/upload",
status_code=200,
)
client = MediaClient(build_transport())
client.upload_video(
"VIDEO-1",
content=b"video-bytes",
content_length=11,
content_range="bytes 0-10/11",
)
request = httpx_mock.get_requests()[0]
assert request.headers["Content-Type"] == "application/octet-stream"
assert request.headers["Content-Length"] == "11"
assert request.headers["Content-Range"] == "bytes 0-10/11"
assert request.content == b"video-bytes"
def test_media_wrapper_returns_document_models(httpx_mock: HTTPXMock) -> None:
httpx_mock.add_response(
method="POST",
url="https://api.ebay.com/commerce/media/v1_beta/document",
json={"documentId": "DOC-1", "documentStatus": "PENDING_UPLOAD"},
status_code=201,
)
httpx_mock.add_response(
method="POST",
url="https://api.ebay.com/commerce/media/v1_beta/document/create_document_from_url",
json={"documentId": "DOC-2", "documentStatus": "SUBMITTED"},
status_code=201,
)
httpx_mock.add_response(
method="GET",
url="https://api.ebay.com/commerce/media/v1_beta/document/DOC-1",
json={"documentId": "DOC-1", "documentStatus": "ACCEPTED", "documentType": "USER_GUIDE_OR_MANUAL"},
)
httpx_mock.add_response(
method="POST",
url="https://api.ebay.com/commerce/media/v1_beta/document/DOC-1/upload",
json={"documentId": "DOC-1", "documentStatus": "SUBMITTED"},
status_code=200,
)
client = MediaClient(build_transport())
created = client.create_document(
CreateDocumentRequest(documentType="USER_GUIDE_OR_MANUAL", languages=["en-US"])
)
created_from_url = client.create_document_from_url(
CreateDocumentFromUrlRequest(
documentType="USER_GUIDE_OR_MANUAL",
documentUrl="https://example.test/guide.pdf",
languages=["en-US"],
)
)
fetched = client.get_document("DOC-1")
uploaded = client.upload_document(
"DOC-1",
file_name="guide.pdf",
content=b"%PDF-1.7",
content_type="application/pdf",
)
assert created.documentId == "DOC-1"
assert created_from_url.documentId == "DOC-2"
assert isinstance(fetched, DocumentResponse)
assert fetched.documentStatus == "ACCEPTED"
assert uploaded.documentStatus == "SUBMITTED"
upload_request = httpx_mock.get_requests()[3]
assert upload_request.headers["Content-Type"].startswith("multipart/form-data;")
assert b"filename=\"guide.pdf\"" in upload_request.content
assert b"%PDF-1.7" in upload_request.content
def test_media_wait_for_video_returns_live_payload(monkeypatch) -> None:
client = MediaClient(build_transport())
states = iter(
[
Video(videoId="VIDEO-1", status="PENDING_UPLOAD"),
Video(videoId="VIDEO-1", status="PROCESSING"),
Video(videoId="VIDEO-1", status="LIVE"),
]
)
monkeypatch.setattr(client, "get_video", lambda _video_id: next(states))
monkeypatch.setattr("ebay_client.media.client.sleep", lambda _seconds: None)
result = client.wait_for_video("VIDEO-1", poll_interval_seconds=0.0)
assert result.status == "LIVE"
def test_media_wait_for_document_raises_on_terminal_failure(monkeypatch) -> None:
client = MediaClient(build_transport())
monkeypatch.setattr(
client,
"get_document",
lambda _document_id: DocumentResponse(documentId="DOC-1", documentStatus="REJECTED"),
)
try:
client.wait_for_document("DOC-1", poll_interval_seconds=0.0)
except ValueError as exc:
assert "REJECTED" in str(exc)
else:
raise AssertionError("Expected wait_for_document to raise on terminal failure status")
def test_media_create_upload_and_wait_video_orchestrates_flow(monkeypatch) -> None:
client = MediaClient(build_transport())
calls: list[tuple[str, object]] = []
monkeypatch.setattr(
client,
"create_video",
lambda payload: calls.append(("create_video", payload)) or CreatedMediaResource(resource_id="VIDEO-9"),
)
monkeypatch.setattr(
client,
"upload_video",
lambda video_id, **kwargs: calls.append(("upload_video", {"video_id": video_id, **kwargs})),
)
monkeypatch.setattr(
client,
"wait_for_video",
lambda video_id, **kwargs: calls.append(("wait_for_video", {"video_id": video_id, **kwargs}))
or Video(videoId=video_id, status="LIVE"),
)
result = client.create_upload_and_wait_video(
CreateVideoRequest(title="Demo", size=4, classification=["ITEM"]),
content=b"demo",
poll_interval_seconds=0.0,
)
assert result.videoId == "VIDEO-9"
assert calls[0][0] == "create_video"
assert calls[1] == (
"upload_video",
{"video_id": "VIDEO-9", "content": b"demo", "content_length": 4, "content_range": None},
)
assert calls[2] == (
"wait_for_video",
{"video_id": "VIDEO-9", "timeout_seconds": 30.0, "poll_interval_seconds": 0.0},
)
def test_media_create_upload_and_wait_document_orchestrates_flow(monkeypatch) -> None:
client = MediaClient(build_transport())
calls: list[tuple[str, object]] = []
monkeypatch.setattr(
client,
"create_document",
lambda payload: calls.append(("create_document", payload))
or CreateDocumentResponse(documentId="DOC-9", documentStatus="PENDING_UPLOAD"),
)
monkeypatch.setattr(
client,
"upload_document",
lambda document_id, **kwargs: calls.append(("upload_document", {"document_id": document_id, **kwargs}))
or DocumentResponse(documentId=document_id, documentStatus="SUBMITTED"),
)
monkeypatch.setattr(
client,
"wait_for_document",
lambda document_id, **kwargs: calls.append(("wait_for_document", {"document_id": document_id, **kwargs}))
or DocumentResponse(documentId=document_id, documentStatus="ACCEPTED"),
)
result = client.create_upload_and_wait_document(
CreateDocumentRequest(documentType="USER_GUIDE_OR_MANUAL", languages=["en-US"]),
file_name="guide.pdf",
content=b"%PDF-1.7",
content_type="application/pdf",
poll_interval_seconds=0.0,
)
assert result.documentStatus == "ACCEPTED"
assert calls[0][0] == "create_document"
assert calls[1] == (
"upload_document",
{
"document_id": "DOC-9",
"file_name": "guide.pdf",
"content": b"%PDF-1.7",
"content_type": "application/pdf",
},
)
assert calls[2] == (
"wait_for_document",
{"document_id": "DOC-9", "timeout_seconds": 30.0, "poll_interval_seconds": 0.0},
)
def test_media_create_document_from_url_and_wait_orchestrates_flow(monkeypatch) -> None:
client = MediaClient(build_transport())
calls: list[tuple[str, object]] = []
monkeypatch.setattr(
client,
"create_document_from_url",
lambda payload: calls.append(("create_document_from_url", payload))
or CreateDocumentResponse(documentId="DOC-10", documentStatus="SUBMITTED"),
)
monkeypatch.setattr(
client,
"wait_for_document",
lambda document_id, **kwargs: calls.append(("wait_for_document", {"document_id": document_id, **kwargs}))
or DocumentResponse(documentId=document_id, documentStatus="ACCEPTED"),
)
result = client.create_document_from_url_and_wait(
CreateDocumentFromUrlRequest(
documentType="USER_GUIDE_OR_MANUAL",
documentUrl="https://example.test/guide.pdf",
languages=["en-US"],
),
poll_interval_seconds=0.0,
)
assert result.documentStatus == "ACCEPTED"
assert calls[0][0] == "create_document_from_url"
assert calls[1] == (
"wait_for_document",
{"document_id": "DOC-10", "timeout_seconds": 30.0, "poll_interval_seconds": 0.0},
)
def test_media_convenience_methods_raise_when_required_ids_are_missing(monkeypatch) -> None:
client = MediaClient(build_transport())
monkeypatch.setattr(client, "create_video", lambda payload: CreatedMediaResource(resource_id=None))
monkeypatch.setattr(client, "create_document", lambda payload: CreateDocumentResponse(documentId=None))
monkeypatch.setattr(client, "create_document_from_url", lambda payload: CreateDocumentResponse(documentId=None))
for action in (
lambda: client.create_upload_and_wait_video(
CreateVideoRequest(title="Demo", size=4, classification=["ITEM"]),
content=b"demo",
poll_interval_seconds=0.0,
),
lambda: client.create_upload_and_wait_document(
CreateDocumentRequest(documentType="USER_GUIDE_OR_MANUAL", languages=["en-US"]),
file_name="guide.pdf",
content=b"%PDF-1.7",
poll_interval_seconds=0.0,
),
lambda: client.create_document_from_url_and_wait(
CreateDocumentFromUrlRequest(
documentType="USER_GUIDE_OR_MANUAL",
documentUrl="https://example.test/guide.pdf",
languages=["en-US"],
),
poll_interval_seconds=0.0,
),
):
try:
action()
except RuntimeError:
pass
else:
raise AssertionError("Expected convenience method to raise when eBay omits the required identifier")