OAuth for HTTPX

HTTPX is a next-generation HTTP client for Python. Authlib enables OAuth 1.0 and OAuth 2.0 for HTTPX with its async versions:

Note

HTTPX is still in its “alpha” stage, use it with caution.

HTTPX OAuth 1.0

There are three steps in OAuth 1 to obtain an access token:

  1. fetch a temporary credential

  2. visit the authorization page

  3. exchange access token with the temporary credential

It shares a common API design with OAuth for Requests.

Read the common guide of OAuth 1 Session to understand the whole OAuth 1.0 flow.

HTTPX OAuth 2.0

In OAuth 2 Session, there are many grant types, including:

  1. Authorization Code Flow

  2. Implicit Flow

  3. Password Flow

  4. Client Credentials Flow

And also, Authlib supports non Standard OAuth 2.0 providers via Compliance Fix.

Read the common guide of OAuth 2 Session to understand the whole OAuth 2.0 flow.

Using client_secret_jwt in HTTPX

Here is how you could register and use client_secret_jwt client authentication method for HTTPX:

from authlib.integrations.httpx_client import AsyncOAuth2Client
from authlib.oauth2.rfc7523 import ClientSecretJWT

client = AsyncOAuth2Client(
    'your-client-id', 'your-client-secret',
    token_endpoint_auth_method='client_secret_jwt'
)
token_endpoint = 'https://example.com/oauth/token'
client.register_client_auth_method(ClientSecretJWT(token_endpoint))
client.fetch_token(token_endpoint)

The ClientSecretJWT is provided by RFC7523: JWT Profile for OAuth 2.0 Client Authentication and Authorization Grants.

Using private_key_jwt in HTTPX

Here is how you could register and use private_key_jwt client authentication method for HTTPX:

from authlib.integrations.httpx_client import AsyncOAuth2Client
from authlib.oauth2.rfc7523 import PrivateKeyJWT

with open('your-private-key.pem', 'rb') as f:
    private_key = f.read()

client = AsyncOAuth2Client(
    'your-client-id', private_key,
    token_endpoint_auth_method='private_key_jwt',
)
token_endpoint = 'https://example.com/oauth/token'
client.register_client_auth_method(PrivateKeyJWT(token_endpoint))
client.fetch_token(token_endpoint)

The PrivateKeyJWT is provided by RFC7523: JWT Profile for OAuth 2.0 Client Authentication and Authorization Grants.

Async OAuth 1.0

The async version of AsyncOAuth1Client works the same as OAuth 1 Session, except that we need to add await when required:

# fetching request token
request_token = await client.fetch_request_token(request_token_url)

# fetching access token
access_token = await client.fetch_access_token(access_token_url)

# normal requests
await client.get(...)
await client.post(...)
await client.put(...)
await client.delete(...)

Async OAuth 2.0

The async version of AsyncOAuth2Client works the same as OAuth 2 Session, except that we need to add await when required:

# fetching access token
token = await client.fetch_token(token_endpoint, ...)

# normal requests
await client.get(...)
await client.post(...)
await client.put(...)
await client.delete(...)

Auto Update Token

The AsyncOAuth2Client also supports update_token parameter, the update_token can either be sync and async. For instance:

async def update_token(token, refresh_token=None, access_token=None):
    if refresh_token:
        item = await OAuth2Token.find(name=name, refresh_token=refresh_token)
    elif access_token:
        item = await OAuth2Token.find(name=name, access_token=access_token)
    else:
        return

    # update old token
    item.access_token = token['access_token']
    item.refresh_token = token.get('refresh_token')
    item.expires_at = token['expires_at']
    await item.save()

Then pass this update_token into AsyncOAuth2Client.

Async Service Account

AsyncAssertionClient is the async version for Assertion Framework of OAuth 2.0 Authorization Grants. It is also know as service account. A configured AsyncAssertionClient will handle token authorization automatically, which means you can just use it.

Take Google Service Account as an example, with the information in your service account JSON configure file:

import json
from authlib.integrations.httpx_client import AsyncAssertionClient

with open('MyProject-1234.json') as f:
    conf = json.load(f)

token_uri = conf['token_uri']
header = {'alg': 'RS256'}
key_id = conf.get('private_key_id')
if key_id:
    header['kid'] = key_id

# Google puts scope in payload
claims = {'scope': scope}

async def main():
    client = AsyncAssertionClient(
        token_endpoint=token_uri,
        issuer=conf['client_email'],
        audience=token_uri,
        claims=claims,
        subject=None,
        key=conf['private_key'],
        header=header,
    )
    resp = await client.get(...)
    resp = await client.post(...)

Close Client Hint

Developers SHOULD close a HTTPX Session when the jobs are done. You can call .close() manually, or use a with context to automatically close the session:

client = OAuth2Client(client_id, client_secret)
client.get(url)
client.close()

with OAuth2Client(client_id, client_secret) as client:
    client.get(url)

For async OAuth Client, use await client.close():

client = AsyncOAuth2Client(client_id, client_secret)
await client.get(url)
await client.close()

async with AsyncOAuth2Client(client_id, client_secret) as client:
    await client.get(url)

Our Web OAuth Clients will close every session automatically, no need to worry.