python: Enable configuring proxies in SvixOptions

Author: svix-mman

python/svix/api/client.py

@@ -1,5 +1,5 @@
 import ssl
-from typing import Dict, List, Union
+from typing import Dict, List, Union, Optional
 
 import attr
 
@@ -60,6 +60,7 @@ class AuthenticatedClient(Client):
     token: str
     prefix: str = "Bearer"
     auth_header_name: str = "Authorization"
+    proxy: Optional[str] = attr.ib(default=None)
 
     def get_headers(self) -> Dict[str, str]:
         """Get headers to be used in authenticated endpoints"""

python/svix/api/common.py

@@ -75,11 +75,29 @@ class ApiBase:
 
     def __init__(self, client: AuthenticatedClient) -> None:
         self._client = client
+
+        if self._client.proxy is not None:
+            proxy_mounts = {
+                "http://": httpx.HTTPTransport(proxy=httpx.Proxy(self._client.proxy)),
+                "https://": httpx.HTTPTransport(proxy=httpx.Proxy(self._client.proxy)),
+            }
+            async_proxy_mounts = {
+                "http://": httpx.AsyncHTTPTransport(
+                    proxy=httpx.Proxy(self._client.proxy)
+                ),
+                "https://": httpx.AsyncHTTPTransport(
+                    proxy=httpx.Proxy(self._client.proxy)
+                ),
+            }
+        else:
+            proxy_mounts = None
+            async_proxy_mounts = None
+
         self._httpx_client = httpx.Client(
-            verify=client.verify_ssl, cookies=self._client.get_cookies()
+            mounts=proxy_mounts, cookies=self._client.get_cookies()
         )
         self._httpx_async_client = httpx.AsyncClient(
-            verify=client.verify_ssl, cookies=self._client.get_cookies()
+            mounts=async_proxy_mounts, cookies=self._client.get_cookies()
         )
 
     def _get_httpx_kwargs(
@@ -168,7 +186,6 @@ def _request_sync(
             header_params=header_params,
             json_body=json_body,
         )
-
         response = self._httpx_client.request(**httpx_kwargs)
         for retry_count, sleep_time in enumerate(self._client.retry_schedule):
             if response.status_code < 500:

python/svix/api/svix.py

@@ -45,6 +45,7 @@ class SvixOptions:
     """
     timeout: float = 15.0
 
+    proxy: t.Optional[str] = None
 
 class ClientBase:
     _client: AuthenticatedClient
@@ -78,6 +79,7 @@ def __init__(self, auth_token: str, options: SvixOptions = SvixOptions()) -> Non
             timeout=options.timeout,
             follow_redirects=False,
             raise_on_unexpected_status=True,
+            proxy=options.proxy,
         )
         self._client = client