Disabling event hooks per request

[mijolabs]

This is somewhat similar to 2649, but not quite the same.

My question is, is it possible to disable an event hook (or all) on a per-request basis?

I think the specific use case that I have in mind is best illustrated with a code block:

async def request_hook(self, request):
    if self.token is None or self.token_has_expired():
        self.token = await self.client.get("/token_endpoint")

Obviously that creates an infinite recursion.

Here's one method of disabling it for a single request:

client.event_hooks = {"request": []}
self.token = await self.client.get("/token_endpoint")
client.event_hooks = {"request": [self.request_hook]}    # Re-enabling it after

But it seems prone to race conditions (and really doesn't look very nice).

It would have been great if one could do, e.g.

self.token = await self.client.get("/token_endpoint", event_hooks={"request": []})

But that doesn't seem supported.

Is there a recommended way of accomplishing this?

[karpetrosyan]

Hi!
I believe you can simply replace event hooks with custom transports as follows:

import httpx

async def request_hook(self, request):
    if self.token is None or self.token_has_expired():
        self.token = await self.client.get("/token_endpoint")

class CustomTransport(httpx.AsyncHTTPTransport):
    async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
        if not request.extensions.get("disable_hooks", False):
            await request_hook(request)

        return await super().handle_async_request(request)


client = httpx.AsyncClient(transport=CustomTransport())

await client.get("/token_endpoint")  # hook_works
await client.get("/token_endpoint", extensions={"disable_hooks": True})  # hook doesn't work

[mijolabs]

Thanks, this approach works. I didn't know I could use a custom transport for this.