Add support for sync-specific or async-specific auth flows

[florimondmanca]

Closes #1176, prompted by #1176 (comment)

Allow authors or auth classes to provide a sync-specific and/or async-specific implementation of the auth flow, so that they can do I/O, synchronization or concurrency without blocking the event loop:

class Auth:
    def auth_flow(self, request) -> Generator[Request, Response, None]:
        yield request  # Current API

    def sync_auth_flow(self, request: Request) -> Generator[Request, Response, None]:
        ...  # Defer to `.auth_flow()` by default

    async def async_auth_flow(self, request: Request) -> AsyncGenerator[Request, Response]:
        ...  # Defer to `.auth_flow()`

Doesn't affect the user-side auth API, nor existing auth implementations.

Rendered docs:

[tomchristie]

Great! A bit more discussion about this over at #1176 (comment)

[tomchristie]

Okay, I'm getting warm fuzzies about this. 😻

Suggestion that I'm curious to get your feedback on...

What do we think to also:

• Adding a sync_auth_flow mirroring async_auth_flow
• Bringing in the request/response body loading logic into the Auth class itself.

So, something like this...

class Auth:
    """
    Base class for all authentication schemes.
    To implement a custom authentication scheme, subclass `Auth` and override
    the `.auth_flow()` method.
    If the authentication scheme does I/O, such as disk access or network calls, or uses
    synchronization primitives such as locks, you should override `.async_auth_flow()`
    to provide an async-friendly implementation that will be used by the `AsyncClient`.
    Usage of sync I/O within an async codebase would block the event loop, and could
    cause performance issues.
    """

    requires_request_body = False
    requires_response_body = False

    def auth_flow(self, request: Request) -> typing.Generator[Request, Response, None]:
        """
        Execute the authentication flow.
        To dispatch a request, `yield` it:
        ```
        yield request
        ```
        The client will `.send()` the response back into the flow generator. You can
        access it like so:
        ```
        response = yield request
        ```
        A `return` (or reaching the end of the generator) will result in the
        client returning the last response obtained from the server.
        You can dispatch as many requests as is necessary.
        """
        # We could *potentially* do something more clever here to return
        # more helpful errors if `sync_auth_flow`/`async_auth_flow` *has* been overridden,
        # but the auth flow is being used in the incorrect sync/async context.
        raise NotImplementedError("Override 'auth_flow' to implement a custom auth class.")
        yield request  # pragma: nocover 

    def sync_auth_flow(
        self, request: Request
    ) -> typing.AsyncGenerator[Request, Response]:
        """
        Execute the authentication flow synchronously.

        By default, this defers to `.auth_flow()`. You should override this method
        when the authentication scheme does I/O, such as disk access or network calls,
        or uses concurrency primitives such as locks.
        """
        if self.requires_request_body:
            request.read()

        flow = self.auth_flow(request)
        request = next(flow)

        while True:
            response = yield request
            if self.requires_response_body:
                response.read()

            try:
                request = flow.send(response)
            except StopIteration:
                break

    async def async_auth_flow(
        self, request: Request
    ) -> typing.AsyncGenerator[Request, Response]:
        """
        Execute the authentication flow asynchronously.

        By default, this defers to `.auth_flow()`. You should override this method
        when the authentication scheme does I/O, such as disk access or network calls,
        or uses concurrency primitives such as locks.
        """
        if self.requires_request_body:
            await request.aread()

        flow = self.auth_flow(request)
        request = next(flow)

        while True:
            response = yield request
            if self.requires_response_body:
                await response.aread()

            try:
                request = flow.send(response)
            except StopIteration:
                break

Here's why that might be a nice thing to do...

• Most users will override auth_flow, that'll support sync+async. Works for any cases where the only I/O you need is sending auth requests and perhaps reading request/response bodies.
• For sync I/O users can override sync_auth_flow. The class will error if you attempt to use it with an async client.
• For async I/O users can override `async_auth_flow. The class will error if you attempt to use it with a sync client.
• Or users can override sync_auth_flow+async_auth_flow.

Also...

• The request/response body logic is more tightly encapsulated, rather than being spread partly across the auth class, and partly in the client.
• Testing the auth class becomes a bit easier, as you don't need to think about handling the request/response body logic in the test case.

So I think testing looks something like this?...

(Okay, we can't actually create Response instances quite like I've done in this example just yet, but I've got some work to talk about planned here, so just roll with it for now...)

Test for Basic Auth

auth = httpx.BasicAuth()
request = httpx.Request("GET", "https://www.example.com")

# The initial request should include a basic auth header.
flow = auth.sync_auth_flow(request)
request = next(flow)
assert request.headers['Authorization'].startswith('Basic')

# No other requests are made.
response = httpx.Response(text='Hello, world!', status_code=200)
with pytest.raises(StopIteration):
    flow.send(response)

Test for Digest Auth with a 200 response

auth = httpx.DigestAuth()
request = httpx.Request("GET", "https://www.example.com")

# The initial request should not include an auth header.
flow = auth.sync_auth_flow(request)
request = next(flow)
assert 'Authorization' not in request.headers

# If a 200 response is returned, then no other requests are made.
response = httpx.Response(text='Hello, world!', status_code=200)
with pytest.raises(StopIteration):
    flow.send(response)

Test for Digest Auth with a 401 response

auth = httpx.DigestAuth()
request = httpx.Request("GET", "https://www.example.com")

# The initial request should not include an auth header.
flow = auth.sync_auth_flow(request)
request = next(flow)
assert 'Authorization' not in request.headers

# If a 401 response is returned, then a digest auth request is made.
headers = {'WWW-Authenticate': 'Digest realm="...", qop="...", nonce="...", opaque="..."'}
response = httpx.Response(text='Auth required', status_code=401, headers=headers)
request = flow.send(response)
assert request.headers['Authorization'].startswith('Digest')

# No other requests are made.
response = httpx.Response(text='Hello, world!', status_code=200)
with pytest.raises(StopIteration):
    flow.send(response)

Testing in an async context looks the same except calling .async_auth_flow(), anext, and asend.

What do we think?

[florimondmanca]

Added docs that were missing to make this PR complete — ready for a new round of reviews!

[tomchristie]

Added in auth unit tests for good measure.
We might(?) even want to doc up unit testing auth classes at some point, but can treat that separately if we do want to.

Anyways, this one is fantastic - very pleased with it! ✨🍰✨