Supporting async auth flows.

[flvndh]

Hello,

I try to make asynchronous calls to the Microsoft Graph API using httpx.AsyncClient. For authentication I use the azure-identity package that provides several client classes to easily get an access token:

from azure.identity.aio import EnvironmentCredential

async with EnvironmentCredential() as credential:
    token = await credential.get_token(*scopes)

I'd like then to subclass httpx.Auth and implement the auth_flow method

class AzureAuth(Auth):

    async def auth_flow(self, request):
        token = await self._credential.get_token(*self._scopes) # Get the access token or retrieve it from cache
        request.headers["Authorization"] = "Bearer " + token.token
        yield request

so that I can do things like

async with AsyncClient(base_url="https://graph.microsoft.com", auth=AzureAuth(...)) as client:
    response = await client.get("/groups")

The problem is that httpx.Auth.auth_flow is synchronous. How can I make it asynchronous ?

[florimondmanca]

Hi!

This is not supported currently.

After thinking about this with alternatives such as running await calls in the event loop (works for asyncio, but trio doesn't support it, and we lose any concurrency benefits), it seems indeed that we might want to consider how to add async support to auth implementations - your use case here is legit.

It's a bit tricky in terms of design and API, but let's consider options…

• Allow auth_flow to be either sync, or async (as you propose above), which means def auth_flow() or async def auth_flow().
  • Cons:
    • Doesn't allow providing both sync and async implementations, so most likely a no-no.
• Add a new async def async_auth_flow() method to the Auth interface.
  • On the client side, AsyncClient must be able to handle both methods. It would prefer async_auth_flow() and use auth_flow() as a fallback. Client will only call into auth_flow().
  • Pros:
    • Backwards compatible (fully additive).
    • We can implement both sync and async on the same Auth subclass, with the choice of implementing them up to the author (perhaps with different libraries).
    • Transparent to the user: authors can add async support without requiring the user to change their code.
  • Cons:
    • Quite verbose in the async case.
• Introduce AsyncAuth with an async def auth_flow() method.
  • The AsyncClient must be able to handle both kinds of auth (Auth and AsyncAuth) transparently, while Client only accepts Auth.
  • Pros:
    • Backwards compatible: when the auth flow doesn't do I/O, authors and users can stick to Auth.
    • When the auth flow must do I/O, author can provide an async class, and tell users to use the async one with AsyncClient.
  • Cons:
    • Instead of a single class, developers now need to deal with "two kinds of classes". But this is only required when the auth class does I/O (if it doesn't, there's no point in having an AsyncAuth). So, pretty confusing.

Right now I'm thinking 2/ might be the best option…

[florimondmanca]

Now, another option: "do nothing". If an auth scheme really did require to perform I/O on each request, that would be quite wasteful, right? In practice, this doesn't seem like something that would pop up very often, if at all.

Indeed, for your specific use case it looks like the acquisition of a token does not depend on the request.

So you could perhaps just write a generic BearerAuth class, that takes a static token…

class BearerAuth:
    def __init__(self, token):
        self._token = token

    def auth_flow(self, request):
        request.headers["Authorization"] = f"Bearer {self.token}"
        yield request

And then use it like this…

from azure.identity.aio import EnvironmentCredential

scopes = ...

async with EnvironmentCredential() as credential:
    token = await credential.get_token(*scopes)

    async with httpx.AsyncClient(auth=BearerAuth(token.token)):
        ...

[florimondmanca]

Closing since from the information provided, I think my suggestion above should solve your use case. If it doesn't, feel free to reach back!

[tomchristie]

Other options here...

• Create an AsyncClient subclass and override .send(...) to deal with 403 responses, and renew the auth token when needed.
• Figure out what the actual HTTP request flow is, so that you can implement it as a custom auth flow, rather than by relying on the helper lib.

Of those two, the first would be tied to async and is marginally less convenient if you wanted to provide it as a third party package, but is likely to provide the hook you need. The second option gives you a single implementation that works with both, and will end up making either sync or async requests as required, but can't give you eg async disk accesses or other async non-HTTP I/O (which for almost all auth schemes shouldn't really matter).

[victoraugustolls]

Hi! Sorry if I'm being intrusive on this matter, but as in my company we use httpx against multiple services, I've seen multiple different auth flows on our side, so:

If an auth scheme really did require to perform I/O on each request, that would be quite wasteful, right? In practice, this doesn't seem like something that would pop up very often, if at all.

That's true, but some requests (against the same service) might need I/O and some not. I think I can give two flow examples:

1st:

• Fetch a token from a remote API which is valid for only 30 minutes
• Save it in local cache (expired with TTL)
• While it is valid, no I/O is performed
• When it is invalid, repeat the process

2nd:

• Fetch a token from a remote API, which you don't know for how long it is going to be valid
• Use it until you receive a 401 response from some API
• Refetch the token

Those two flows wouldn't make an I/O for each request, but might need to.

I do have these flow on my end, and many more (and having I/O or not in them is around 50/50), and would be great if we could use the Auth class for this.
I solved those, and others, on my end bu having a custom HTTPService class with a custom flow in each.

[victoraugustolls]

Just one snippet example:

    async def get_token(self):
        token = await self._cache.get("TOKEN")
        if token is not None:
            return token

        await self._lock.acquire()
        try:
            token = await self._cache.get("TOKEN")
            if token is not None:
                return token

            token = await self._request_token()
            await self._cache.set("TOKEN", token, 27 * HALF_MINUTE)
            return token
        finally:
            self._lock.release()

await self._request_token() is actually a pretty complex endpoint request, but just wanted to give an example here.

[tomchristie]

The locking around token acquiry is a really good example case, yup.

[victoraugustolls]

@tomchristie So should this issue be reopened? Or maybe open another one to track the need for an async authentication flow?

[K900]

Seconding this request. We're dealing with a system that's very similar to the one described earlier, and having async auth would be really helpful.