refactor: Centralize API error handling, resource extraction, and marketplace validation into new core modules.

Author: sergioteula

amazon_creatorsapi/aio/api.py

@@ -13,15 +13,11 @@
 from typing_extensions import Self
 
 from amazon_creatorsapi.core.constants import DEFAULT_THROTTLING
-from amazon_creatorsapi.core.marketplaces import MARKETPLACES
+from amazon_creatorsapi.core.error_handling import handle_api_error
 from amazon_creatorsapi.core.parsers import get_asin, get_items_ids
-from amazon_creatorsapi.errors import (
-    AssociateValidationError,
-    InvalidArgumentError,
-    ItemsNotFoundError,
-    RequestError,
-    TooManyRequestsError,
-)
+from amazon_creatorsapi.core.resources import get_all_resources
+from amazon_creatorsapi.core.validation import validate_and_get_marketplace
+from amazon_creatorsapi.errors import ItemsNotFoundError
 
 try:
     from .auth import VERSION_ENDPOINTS, AsyncOAuth2TokenManager
@@ -132,21 +128,12 @@ def __init__(
         self._credential_secret = credential_secret
         self._version = version
         self._last_query_time = time.time() - throttling
-        self._throttle_lock = asyncio.Lock()
+        self._throttle_lock: asyncio.Lock | None = None
         self.tag = tag
         self.throttling = float(throttling)
 
         # Determine marketplace from country or direct value
-        if marketplace:
-            self.marketplace = marketplace
-        elif country:
-            if country not in MARKETPLACES:
-                msg = f"Country code '{country}' is not valid"
-                raise InvalidArgumentError(msg)
-            self.marketplace = MARKETPLACES[country]
-        else:
-            msg = "Either 'country' or 'marketplace' must be provided"
-            raise InvalidArgumentError(msg)
+        self.marketplace = validate_and_get_marketplace(country, marketplace)
 
         # HTTP client and token manager (initialized lazily or via context manager)
         self._http_client: AsyncHttpClient | None = None
@@ -218,7 +205,7 @@ async def get_items(
 
         """
         if resources is None:
-            resources = self._get_all_resources(GetItemsResource)
+            resources = get_all_resources(GetItemsResource)
 
         item_ids = get_items_ids(items)
 
@@ -299,7 +286,7 @@ async def search_items(  # noqa: PLR0912, C901
 
         """
         if resources is None:
-            resources = self._get_all_resources(SearchItemsResource)
+            resources = get_all_resources(SearchItemsResource)
 
         request_body: dict[str, Any] = {
             "partnerTag": self.tag,
@@ -382,7 +369,7 @@ async def get_variations(
 
         """
         if resources is None:
-            resources = self._get_all_resources(GetVariationsResource)
+            resources = get_all_resources(GetVariationsResource)
 
         asin = get_asin(asin)
 
@@ -433,7 +420,7 @@ async def get_browse_nodes(
 
         """
         if resources is None:
-            resources = self._get_all_resources(GetBrowseNodesResource)
+            resources = get_all_resources(GetBrowseNodesResource)
 
         request_body: dict[str, Any] = {
             "partnerTag": self.tag,
@@ -462,6 +449,10 @@ async def _throttle(self) -> None:
         Uses asyncio.Lock to prevent race conditions when multiple coroutines
         attempt to make concurrent requests.
         """
+        # Lazy initialization of the lock (ensures event loop is active)
+        if self._throttle_lock is None:
+            self._throttle_lock = asyncio.Lock()
+
         async with self._throttle_lock:
             wait_time = self.throttling - (time.time() - self._last_query_time)
             if wait_time > 0:
@@ -525,45 +516,7 @@ def _handle_error_response(self, status_code: int, body: str) -> None:
             RequestError: For other errors.
 
         """
-        http_not_found = 404
-        http_too_many_requests = 429
-
-        if status_code == http_not_found:
-            msg = "No items found for the request"
-            raise ItemsNotFoundError(msg)
-
-        if status_code == http_too_many_requests:
-            msg = "Rate limit exceeded, try increasing throttling"
-            raise TooManyRequestsError(msg)
-
-        if "InvalidParameterValue" in body:
-            msg = "Invalid parameter value provided in the request"
-            raise InvalidArgumentError(msg)
-
-        if "InvalidPartnerTag" in body:
-            msg = "The partner tag is invalid or not present"
-            raise InvalidArgumentError(msg)
-
-        if "InvalidAssociate" in body:
-            msg = "Credentials are not valid for the selected marketplace"
-            raise AssociateValidationError(msg)
-
-        # Generic error
-        body_info = f" - {body[:200]}" if body else ""
-        msg = f"Request failed with status {status_code}{body_info}"
-        raise RequestError(msg)
-
-    def _get_all_resources(self, resource_class: type[ResourceT]) -> list[ResourceT]:
-        """Extract all resource values from a resource enum class.
-
-        Args:
-            resource_class: Enum class containing resource definitions.
-
-        Returns:
-            List of all enum members from the resource class.
-
-        """
-        return list(resource_class)
+        handle_api_error(status_code, body)
 
     def _deserialize_items(self, items_data: list[dict[str, Any]]) -> list[Item]:
         """Deserialize item data from API response to Item models."""

amazon_creatorsapi/api.py

@@ -6,18 +6,14 @@
 from __future__ import annotations
 
 import time
-from typing import TYPE_CHECKING, Any, NoReturn
+from typing import TYPE_CHECKING, NoReturn
 
 from amazon_creatorsapi.core.constants import DEFAULT_THROTTLING
-from amazon_creatorsapi.core.marketplaces import MARKETPLACES
+from amazon_creatorsapi.core.error_handling import handle_api_error
 from amazon_creatorsapi.core.parsers import get_asin, get_items_ids
-from amazon_creatorsapi.errors import (
-    AssociateValidationError,
-    InvalidArgumentError,
-    ItemsNotFoundError,
-    RequestError,
-    TooManyRequestsError,
-)
+from amazon_creatorsapi.core.resources import get_all_resources
+from amazon_creatorsapi.core.validation import validate_and_get_marketplace
+from amazon_creatorsapi.errors import ItemsNotFoundError
 from creatorsapi_python_sdk.api.default_api import DefaultApi
 from creatorsapi_python_sdk.api_client import ApiClient
 from creatorsapi_python_sdk.exceptions import ApiException
@@ -41,8 +37,6 @@
 from creatorsapi_python_sdk.models.search_items_resource import SearchItemsResource
 
 if TYPE_CHECKING:
-    from enum import Enum
-
     from amazon_creatorsapi.core.marketplaces import CountryCode
     from creatorsapi_python_sdk.models.browse_node import BrowseNode
     from creatorsapi_python_sdk.models.condition import Condition
@@ -98,16 +92,7 @@ def __init__(
         self.throttling = float(throttling)
 
         # Determine marketplace from country or direct value
-        if marketplace:
-            self.marketplace = marketplace
-        elif country:
-            if country not in MARKETPLACES:
-                msg = f"Country code '{country}' is not valid"
-                raise InvalidArgumentError(msg)
-            self.marketplace = MARKETPLACES[country]
-        else:
-            msg = "Either 'country' or 'marketplace' must be provided"
-            raise InvalidArgumentError(msg)
+        self.marketplace = validate_and_get_marketplace(country, marketplace)
 
         self._api_client = ApiClient(
             credential_id=credential_id,
@@ -143,7 +128,7 @@ def get_items(
 
         """
         if resources is None:
-            resources = self._get_all_resources(GetItemsResource)
+            resources = get_all_resources(GetItemsResource)
 
         item_ids = get_items_ids(items)
 
@@ -228,7 +213,7 @@ def search_items(
 
         """
         if resources is None:
-            resources = self._get_all_resources(SearchItemsResource)
+            resources = get_all_resources(SearchItemsResource)
 
         request = SearchItemsRequestContent(
             partnerTag=self.tag,
@@ -298,7 +283,7 @@ def get_variations(
 
         """
         if resources is None:
-            resources = self._get_all_resources(GetVariationsResource)
+            resources = get_all_resources(GetVariationsResource)
 
         asin = get_asin(asin)
 
@@ -350,7 +335,7 @@ def get_browse_nodes(
 
         """
         if resources is None:
-            resources = self._get_all_resources(GetBrowseNodesResource)
+            resources = get_all_resources(GetBrowseNodesResource)
 
         request = GetBrowseNodesRequestContent(
             partnerTag=self.tag,
@@ -385,39 +370,11 @@ def _throttle(self) -> None:
             time.sleep(wait_time)
         self._last_query_time = time.time()
 
-    def _get_all_resources(self, resource_class: type[Enum]) -> list[Any]:
-        """Extract all resource values from a resource enum class."""
-        return [member.value for member in resource_class]
-
     def _handle_api_exception(self, error: ApiException) -> NoReturn:
         """Handle API exceptions and raise appropriate custom exceptions."""
-        http_not_found = 404
-        http_too_many_requests = 429
-
-        if error.status == http_not_found:
-            msg = "No items found for the request"
-            raise ItemsNotFoundError(msg) from error
-
-        if error.status == http_too_many_requests:
-            msg = "Rate limit exceeded, try increasing throttling"
-            raise TooManyRequestsError(msg) from error
-
         error_body = str(error.body) if error.body else ""
-
-        if "InvalidParameterValue" in error_body:
-            msg = "Invalid parameter value provided in the request"
-            raise InvalidArgumentError(msg) from error
-
-        if "InvalidPartnerTag" in error_body:
-            msg = "The partner tag is invalid or not present"
-            raise InvalidArgumentError(msg) from error
-
-        if "InvalidAssociate" in error_body:
-            msg = "Credentials are not valid for the selected marketplace"
-            raise AssociateValidationError(msg) from error
-
-        # Include error body in message for debugging
-        reason = error.reason or "Unknown error"
-        body_info = f" - {error_body[:200]}" if error_body else ""
-        msg = f"Request failed: {reason}{body_info}"
-        raise RequestError(msg) from error
+        try:
+            handle_api_error(error.status, error_body)
+        except Exception as exc:
+            # Re-raise with original exception as cause for better stack traces
+            raise exc from error

amazon_creatorsapi/core/constants.py

@@ -1,3 +1,7 @@
 """Constants for the Amazon Creators API."""
 
 DEFAULT_THROTTLING = 1
+
+# HTTP status codes
+HTTP_NOT_FOUND = 404
+HTTP_TOO_MANY_REQUESTS = 429

amazon_creatorsapi/core/error_handling.py

@@ -0,0 +1,55 @@
+"""Error handling utilities for the Amazon Creators API."""
+
+from __future__ import annotations
+
+from typing import NoReturn
+
+from amazon_creatorsapi.core.constants import HTTP_NOT_FOUND, HTTP_TOO_MANY_REQUESTS
+from amazon_creatorsapi.errors import (
+    AssociateValidationError,
+    InvalidArgumentError,
+    ItemsNotFoundError,
+    RequestError,
+    TooManyRequestsError,
+)
+
+
+def handle_api_error(status_code: int, body: str) -> NoReturn:
+    """Handle API error responses and raise appropriate exceptions.
+
+    Args:
+        status_code: HTTP status code.
+        body: Response body text.
+
+    Raises:
+        ItemsNotFoundError: For 404 errors.
+        TooManyRequestsError: For 429 errors.
+        InvalidArgumentError: For validation errors.
+        AssociateValidationError: For invalid associate credentials.
+        RequestError: For other errors.
+
+    """
+    if status_code == HTTP_NOT_FOUND:
+        msg = "No items found for the request"
+        raise ItemsNotFoundError(msg)
+
+    if status_code == HTTP_TOO_MANY_REQUESTS:
+        msg = "Rate limit exceeded, try increasing throttling"
+        raise TooManyRequestsError(msg)
+
+    if "InvalidParameterValue" in body:
+        msg = "Invalid parameter value provided in the request"
+        raise InvalidArgumentError(msg)
+
+    if "InvalidPartnerTag" in body:
+        msg = "The partner tag is invalid or not present"
+        raise InvalidArgumentError(msg)
+
+    if "InvalidAssociate" in body:
+        msg = "Credentials are not valid for the selected marketplace"
+        raise AssociateValidationError(msg)
+
+    # Generic error
+    body_info = f" - {body[:200]}" if body else ""
+    msg = f"Request failed with status {status_code}{body_info}"
+    raise RequestError(msg)

amazon_creatorsapi/core/resources.py

@@ -0,0 +1,22 @@
+"""Resource utilities for the Amazon Creators API."""
+
+from __future__ import annotations
+
+from enum import Enum
+from typing import TypeVar
+
+# TypeVar for generic resource handling
+ResourceT = TypeVar("ResourceT", bound=Enum)
+
+
+def get_all_resources(resource_class: type[ResourceT]) -> list[ResourceT]:
+    """Extract all resource values from a resource enum class.
+
+    Args:
+        resource_class: Enum class containing resource definitions.
+
+    Returns:
+        List of all enum members from the resource class.
+
+    """
+    return list(resource_class)

amazon_creatorsapi/core/validation.py

@@ -0,0 +1,40 @@
+"""Validation utilities for the Amazon Creators API."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from amazon_creatorsapi.core.marketplaces import MARKETPLACES
+from amazon_creatorsapi.errors import InvalidArgumentError
+
+if TYPE_CHECKING:
+    from amazon_creatorsapi.core.marketplaces import CountryCode
+
+
+def validate_and_get_marketplace(
+    country: CountryCode | None,
+    marketplace: str | None,
+) -> str:
+    """Validate and determine marketplace from country or direct value.
+
+    Args:
+        country: Country code (e.g., "ES", "US").
+        marketplace: Marketplace URL (e.g., "www.amazon.es").
+
+    Returns:
+        The marketplace URL.
+
+    Raises:
+        InvalidArgumentError: If neither country nor marketplace is provided,
+            or if the country code is invalid.
+
+    """
+    if marketplace:
+        return marketplace
+    if country:
+        if country not in MARKETPLACES:
+            msg = f"Country code '{country}' is not valid"
+            raise InvalidArgumentError(msg)
+        return MARKETPLACES[country]
+    msg = "Either 'country' or 'marketplace' must be provided"
+    raise InvalidArgumentError(msg)