feat(proxy): add proxy support to AmazonCreatorsApi

amazon_creatorsapi/aio/api.py

@@ -104,6 +104,9 @@ class AsyncAmazonCreatorsApi:
         country: Country code (e.g., "ES", "US"). Used to determine marketplace.
         marketplace: Marketplace URL (e.g., "www.amazon.es"). Overrides country.
         throttling: Wait time in seconds between API calls. Defaults to 1 second.
+        proxy: Optional HTTP proxy URL, e.g. ``"http://user:pass@proxy:3128"``.
+            Applied to both API calls and OAuth2 token refresh (httpx handles
+            credentials embedded in the URL). Defaults to no proxy.
 
     Raises:
         InvalidArgumentError: If neither country nor marketplace is provided.
@@ -121,6 +124,7 @@ def __init__(
         country: CountryCode | None = None,
         marketplace: str | None = None,
         throttling: float = DEFAULT_THROTTLING,
+        proxy: str | None = None,
     ) -> None:
         """Initialize the async Amazon Creators API client."""
         # Validate version early to fail fast (before token manager initialization)
@@ -139,10 +143,13 @@ def __init__(
 
         # HTTP client and token manager (initialized lazily or via context manager)
         self._http_client: AsyncHttpClient | None = None
+        # Normalize empty string to None so httpx doesn't reject proxy="".
+        self._proxy = proxy or None
         self._token_manager = AsyncOAuth2TokenManager(
             credential_id=credential_id,
             credential_secret=credential_secret,
             version=version,
+            proxy=self._proxy,
         )
         self._owns_client = False
 
@@ -163,7 +170,7 @@ def _validate_version(self, version: str) -> None:
 
     async def __aenter__(self) -> Self:
         """Enter async context manager, creating a persistent HTTP client."""
-        self._http_client = AsyncHttpClient(host=API_HOST)
+        self._http_client = AsyncHttpClient(host=API_HOST, proxy=self._proxy)
         await self._http_client.__aenter__()
         self._owns_client = True
         return self
@@ -498,7 +505,7 @@ async def _make_request(
         if self._http_client is not None:
             response = await self._http_client.post(endpoint, headers, body)
         else:
-            async with AsyncHttpClient(host=API_HOST) as client:
+            async with AsyncHttpClient(host=API_HOST, proxy=self._proxy) as client:
                 response = await client.post(endpoint, headers, body)
 
         # Handle errors

amazon_creatorsapi/aio/auth.py

@@ -55,6 +55,9 @@ class AsyncOAuth2TokenManager:
         credential_secret: OAuth2 credential secret.
         version: API version (determines auth endpoint).
         auth_endpoint: Optional custom auth endpoint URL.
+        proxy: Optional HTTP proxy URL, e.g. ``"http://user:pass@proxy:3128"``.
+            Applied to token refresh requests (httpx handles credentials
+            embedded in the URL). Defaults to no proxy.
 
     """
 
@@ -64,12 +67,15 @@ def __init__(
         credential_secret: str,
         version: str,
         auth_endpoint: str | None = None,
+        proxy: str | None = None,
     ) -> None:
         """Initialize the async OAuth2 token manager."""
         self._credential_id = credential_id
         self._credential_secret = credential_secret
         self._version = version
         self._auth_endpoint = self._determine_auth_endpoint(version, auth_endpoint)
+        # Normalize empty string to None so httpx doesn't reject proxy="".
+        self._proxy = proxy or None
 
         self._access_token: str | None = None
         self._expires_at: float | None = None
@@ -186,7 +192,7 @@ async def refresh_token(self) -> str:
         }
 
         try:
-            async with httpx.AsyncClient() as client:
+            async with httpx.AsyncClient(proxy=self._proxy) as client:
                 if self.is_lwa():
                     response = await client.post(
                         self._auth_endpoint,

amazon_creatorsapi/aio/client.py

@@ -65,17 +65,23 @@ class AsyncHttpClient:
     Args:
         host: Base URL for API requests. Defaults to Amazon Creators API.
         timeout: Request timeout in seconds. Defaults to 30.
+        proxy: Optional HTTP proxy URL, e.g. ``"http://user:pass@proxy:3128"``.
+            httpx applies it to every request (and handles credentials
+            embedded in the URL). Defaults to no proxy.
 
     """
 
     def __init__(
         self,
         host: str = DEFAULT_HOST,
         timeout: float = DEFAULT_TIMEOUT,
+        proxy: str | None = None,
     ) -> None:
         """Initialize the async HTTP client."""
         self._host = host
         self._timeout = timeout
+        # Normalize empty string to None so httpx doesn't reject proxy="".
+        self._proxy = proxy or None
         self._client: httpx.AsyncClient | None = None
         self._owns_client = False
 
@@ -85,6 +91,7 @@ async def __aenter__(self) -> Self:
             base_url=self._host,
             timeout=self._timeout,
             headers={"User-Agent": USER_AGENT},
+            proxy=self._proxy,
         )
         self._owns_client = True
         return self
@@ -132,6 +139,7 @@ async def post(
             async with httpx.AsyncClient(
                 base_url=self._host,
                 timeout=self._timeout,
+                proxy=self._proxy,
             ) as client:
                 response = await client.post(
                     path,

amazon_creatorsapi/api.py

@@ -16,6 +16,7 @@
 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.configuration import Configuration
 from creatorsapi_python_sdk.exceptions import ApiException
 from creatorsapi_python_sdk.models.get_browse_nodes_request_content import (
     GetBrowseNodesRequestContent,
@@ -58,6 +59,8 @@ class AmazonCreatorsApi:
         country: Country code (e.g., "ES", "US"). Used to determine marketplace.
         marketplace: Marketplace URL (e.g., "www.amazon.es"). Overrides country.
         throttling: Wait time in seconds between API calls. Defaults to 1 second.
+        proxy: Optional HTTP proxy URL, e.g. ``"http://user:pass@proxy:3128"``.
+            Applied to both regular API calls and OAuth2 token refresh.
 
     Raises:
         InvalidArgumentError: If neither country nor marketplace is provided.
@@ -83,6 +86,7 @@ def __init__(
         country: CountryCode | None = None,
         marketplace: str | None = None,
         throttling: float = DEFAULT_THROTTLING,
+        proxy: str | None = None,
     ) -> None:
         """Initialize the Amazon Creators API client."""
         self._credential_id = credential_id
@@ -95,7 +99,11 @@ def __init__(
         # Determine marketplace from country or direct value
         self.marketplace = validate_and_get_marketplace(country, marketplace)
 
+        configuration = Configuration()
+        configuration.proxy = proxy
+
         self._api_client = ApiClient(
+            configuration=configuration,
             credential_id=credential_id,
             credential_secret=credential_secret,
             version=version,

creatorsapi_python_sdk/api_client.py

@@ -384,7 +384,9 @@ def call_api(
                             self.credential_id, self.credential_secret,
                             self.version, self.auth_endpoint
                         )
-                        self._token_manager = OAuth2TokenManager(config)
+                        proxy = self.configuration.proxy
+                        proxies = {"http": proxy, "https": proxy} if proxy else None
+                        self._token_manager = OAuth2TokenManager(config, proxies=proxies)
             # Get token (will use cached token if valid)
             token = self._token_manager.get_token()
             # Add Authorization headers - Version only for v2.x

creatorsapi_python_sdk/auth/oauth2_token_manager.py

@@ -30,13 +30,15 @@
 class OAuth2TokenManager:
     """Manages OAuth2 token lifecycle including acquisition, caching, and automatic refresh"""
 
-    def __init__(self, config):
+    def __init__(self, config, proxies=None):
         """
         Creates an OAuth2TokenManager instance
-        
+
         :param config: The OAuth2Config instance
+        :param proxies: Optional dict of proxy URLs, e.g. {"http": "http://proxy:3128", "https": "http://proxy:3128"}
         """
         self.config = config
+        self.proxies = proxies
         self.access_token = None
         self.expires_at = None
 
@@ -67,6 +69,10 @@ def refresh_token(self):
         :raises Exception: If token refresh fails
         """
         try:
+            session = requests.Session()
+            if self.proxies:
+                session.proxies.update(self.proxies)
+
             if self.config.is_lwa():
                 # LWA (v3.x) uses JSON body
                 request_data = {
@@ -76,7 +82,7 @@ def refresh_token(self):
                     'scope': self.config.get_scope()
                 }
                 headers = {'Content-Type': 'application/json'}
-                response = requests.post(
+                response = session.post(
                     self.config.get_cognito_endpoint(),
                     json=request_data,
                     headers=headers
@@ -90,7 +96,7 @@ def refresh_token(self):
                     'scope': self.config.get_scope()
                 }
                 headers = {'Content-Type': 'application/x-www-form-urlencoded'}
-                response = requests.post(
+                response = session.post(
                     self.config.get_cognito_endpoint(),
                     data=request_data,
                     headers=headers

creatorsapi_python_sdk/rest.py

@@ -110,8 +110,28 @@ def __init__(self, configuration) -> None:
                 pool_args["headers"] = configuration.proxy_headers
                 self.pool_manager = SOCKSProxyManager(**pool_args)
             else:
-                pool_args["proxy_url"] = configuration.proxy
-                pool_args["proxy_headers"] = configuration.proxy_headers
+                proxy_url = configuration.proxy
+                proxy_headers = configuration.proxy_headers
+                # urllib3 ProxyManager ignores credentials embedded in the
+                # proxy URL for HTTPS CONNECT tunneling — they must be passed
+                # via proxy_headers instead.  Extract them here so callers can
+                # pass a plain "http://user:pass@host:port" URL and get correct
+                # CONNECT auth without any extra configuration.
+                if proxy_headers is None:
+                    from urllib.parse import urlparse
+                    _parsed = urlparse(proxy_url)
+                    if _parsed.username:
+                        proxy_headers = urllib3.make_headers(
+                            proxy_basic_auth=(
+                                f"{_parsed.username}:{_parsed.password}"
+                            )
+                        )
+                        proxy_url = (
+                            f"{_parsed.scheme}://"
+                            f"{_parsed.hostname}:{_parsed.port}"
+                        )
+                pool_args["proxy_url"] = proxy_url
+                pool_args["proxy_headers"] = proxy_headers
                 self.pool_manager = urllib3.ProxyManager(**pool_args)
         else:
             self.pool_manager = urllib3.PoolManager(**pool_args)

tests/amazon_creatorsapi/aio/api_test.py

@@ -82,6 +82,37 @@ def test_accepts_lwa_version(self, mock_token_manager: MagicMock) -> None:
 
         self.assertEqual(api.marketplace, "www.amazon.com")
 
+    @patch("amazon_creatorsapi.aio.api.AsyncOAuth2TokenManager")
+    def test_init_with_proxy(self, mock_token_manager: MagicMock) -> None:
+        """Test proxy URL is passed through to the token manager."""
+        proxy_url = "http://user:pass@proxy.example.com:3128"
+        api = AsyncAmazonCreatorsApi(
+            credential_id="test_id",
+            credential_secret="test_secret",
+            version="2.2",
+            tag="test-tag",
+            country="ES",
+            proxy=proxy_url,
+        )
+
+        self.assertEqual(api._proxy, proxy_url)
+        call_kwargs = mock_token_manager.call_args.kwargs
+        self.assertEqual(call_kwargs["proxy"], proxy_url)
+
+    @patch("amazon_creatorsapi.aio.api.AsyncOAuth2TokenManager")
+    def test_init_without_proxy(self, mock_token_manager: MagicMock) -> None:
+        """Test token manager receives proxy=None when not provided."""
+        AsyncAmazonCreatorsApi(
+            credential_id="test_id",
+            credential_secret="test_secret",
+            version="2.2",
+            tag="test-tag",
+            country="ES",
+        )
+
+        call_kwargs = mock_token_manager.call_args.kwargs
+        self.assertIsNone(call_kwargs["proxy"])
+
     @patch("amazon_creatorsapi.aio.api.AsyncOAuth2TokenManager")
     def test_raises_error_when_no_country_or_marketplace(
         self, mock_token_manager: MagicMock
@@ -157,6 +188,31 @@ async def test_context_manager_creates_and_closes_client(
 
         mock_client.__aexit__.assert_called_once()
 
+    @patch("amazon_creatorsapi.aio.api.AsyncOAuth2TokenManager")
+    @patch("amazon_creatorsapi.aio.api.AsyncHttpClient")
+    async def test_context_manager_passes_proxy_to_client(
+        self,
+        mock_http_client_class: MagicMock,
+        mock_token_manager: MagicMock,
+    ) -> None:
+        """Test context manager passes proxy to AsyncHttpClient."""
+        proxy_url = "http://user:pass@proxy.example.com:3128"
+        mock_client = AsyncMock()
+        mock_http_client_class.return_value = mock_client
+
+        async with AsyncAmazonCreatorsApi(
+            credential_id="test_id",
+            credential_secret="test_secret",
+            version="2.2",
+            tag="test-tag",
+            country="ES",
+            proxy=proxy_url,
+        ) as api:
+            self.assertEqual(api._proxy, proxy_url)
+
+        call_kwargs = mock_http_client_class.call_args.kwargs
+        self.assertEqual(call_kwargs["proxy"], proxy_url)
+
     @patch("amazon_creatorsapi.aio.api.AsyncOAuth2TokenManager")
     async def test_context_manager_exit_without_client(
         self,
@@ -1265,6 +1321,46 @@ async def test_request_without_context_manager(
 
         self.assertEqual(len(items), 1)
 
+    @patch("amazon_creatorsapi.aio.api.AsyncOAuth2TokenManager")
+    @patch("amazon_creatorsapi.aio.api.AsyncHttpClient")
+    async def test_request_without_context_manager_passes_proxy(
+        self,
+        mock_http_client_class: MagicMock,
+        mock_token_manager_class: MagicMock,
+    ) -> None:
+        """Test standalone request creates temp client with proxy."""
+        proxy_url = "http://user:pass@proxy.example.com:3128"
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.json.return_value = {
+            "itemsResult": {"items": [{"ASIN": "B0DLFMFBJW"}]}
+        }
+
+        mock_client = AsyncMock()
+        mock_client.post.return_value = mock_response
+        mock_client.__aenter__.return_value = mock_client
+        mock_http_client_class.return_value = mock_client
+
+        mock_token_manager = AsyncMock()
+        mock_token_manager.get_token.return_value = "test_token"
+        mock_token_manager_class.return_value = mock_token_manager
+
+        api = AsyncAmazonCreatorsApi(
+            credential_id="test_id",
+            credential_secret="test_secret",
+            version="2.2",
+            tag="test-tag",
+            country="ES",
+            throttling=0,
+            proxy=proxy_url,
+        )
+
+        items = await api.get_items(["B0DLFMFBJW"])
+
+        self.assertEqual(len(items), 1)
+        call_kwargs = mock_http_client_class.call_args.kwargs
+        self.assertEqual(call_kwargs["proxy"], proxy_url)
+
     @patch("amazon_creatorsapi.aio.api.AsyncOAuth2TokenManager")
     @patch("amazon_creatorsapi.aio.api.AsyncHttpClient")
     async def test_request_uses_v2_authorization_header(