Skip to content

Instantly share code, notes, and snippets.

@oshea00
Created February 6, 2025 00:36
Show Gist options
  • Save oshea00/af8037374356918b52e1fe4fa42c413e to your computer and use it in GitHub Desktop.
Save oshea00/af8037374356918b52e1fe4fa42c413e to your computer and use it in GitHub Desktop.

Custom auth for httpx

Let me help explain both parameters in detail: For the verify parameter:

It can accept several formats:

True/False (boolean) for enabling/disabling verification A path to a CA bundle file (string) A path to a directory containing CA certificates (string) A ssl.SSLContext instance For PEM format specifically - yes, you can pass a PEM-formatted certificate string, but it needs to be saved as a file first. HTTPX doesn't directly accept PEM strings in memory

For the auth parameter, it accepts objects that implement the Auth interface. Here are the key requirements: Example CustomAuth class to use with httpx.Client "auth" parameter

class OAuth2ClientCredentials(httpx.Auth):
    def __init__(self, token_url, client_id, client_secret):
        self.token_url = token_url
        self.client_id = client_id
        self.client_secret = client_secret
        self._token = None
        self._client = httpx.Client()

    def auth_flow(self, request):
        if not self._token:
            # Get initial token
            response = self._client.post(
                self.token_url,
                data={
                    "grant_type": "client_credentials",
                    "client_id": self.client_id,
                    "client_secret": self.client_secret,
                }
            )
            response.raise_for_status()
            self._token = response.json()["access_token"]

        # Add token to request
        request.headers["Authorization"] = f"Bearer {self._token}"
        
        yield request

        # Handle 401s by refreshing token and retrying
        if response.status_code == 401:
            self._token = None  # Force token refresh
            request.headers["Authorization"] = f"Bearer {self._token}"
            yield request

This implementation:

  1. Handles the initial token fetch
  2. Caches the token for reuse
  3. Automatically refreshes on 401 responses
  4. Uses the auth_flow generator pattern to allow for request modification before and after the response

Async methods

class OAuth2ClientCredentials(httpx.Auth):
    def __init__(self, token_url, client_id, client_secret):
        self.token_url = token_url
        self.client_id = client_id
        self.client_secret = client_secret
        self._token = None

    def sync_auth_flow(self, request):
        # Used when making requests with httpx.Client()
        if not self._token:
            with httpx.Client() as client:
                response = client.post(...)
                self._token = response.json()["access_token"]
        request.headers["Authorization"] = f"Bearer {self._token}"
        yield request

    async def auth_flow(self, request):
        # Used when making requests with httpx.AsyncClient()
        if not self._token:
            async with httpx.AsyncClient() as client:
                response = await client.post(...)
                self._token = response.json()["access_token"]
        request.headers["Authorization"] = f"Bearer {self._token}"
        yield request

Without sync_auth_flow, synchronous code would still work but would have to go through the async machinery unnecessarily. The sync version is particularly useful in applications that don't use any async code at all.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment