HTTPX is a next-generation HTTP client for Python. Authlib enables OAuth 1.0 and OAuth 2.0 for HTTPX with its async versions:
OAuth1Client
OAuth2Client
AssertionClient
AsyncOAuth1Client
AsyncOAuth2Client
AsyncAssertionClient
Note
HTTPX is still in its “alpha” stage, use it with caution.
There are three steps in OAuth 1 to obtain an access token:
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.
In OAuth 2 Session, there are many grant types, including:
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.
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.
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.
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(...)
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(...)
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
.
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(...)
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.