how do you properly reuse an httpx.AsyncClient within a FastAPI application? [duplicate]

Question

I have a FastAPI application which, in several different occasions, needs to call external APIs. I use httpx.AsyncClient for these calls. The point is that I don't fully understand how I shoud use it.

From httpx' documentation I should use context managers,

async def foo():
    """"
    I need to call foo quite often from different 
    parts of my application
    """
    async with httpx.AsyncClient() as aclient:
        # make some http requests, e.g.,
        await aclient.get("http://example.it")

However, I understand that in this way a new client is spawned each time I call foo(), and is precisely what we want to avoid by using a client in the first place.

I suppose an alternative would be to have some global client defined somewhere, and just import it whenever I need it like so

aclient = httpx.AsyncClient()

async def bar():
    # make some http requests using the global aclient, e.g.,
    await aclient.get("http://example.it")

This second option looks somewhat fishy, though, as nobody is taking care of closing the session and the like.

So the question is: how do I properly (re)use httpx.AsyncClient() within a FastAPI application?

Nuclide

This question has already been answered here and here. – Ma

Answer 1

You can have a global client that is closed in the FastApi shutdown event.

import logging
from fastapi import FastAPI
import httpx

logging.basicConfig(level=logging.INFO, format="%(levelname)-9s %(asctime)s - %(name)s - %(message)s")
LOGGER = logging.getLogger(__name__)


class HTTPXClientWrapper:

    async_client = None

    def start(self):
        """ Instantiate the client. Call from the FastAPI startup hook."""
        self.async_client = httpx.AsyncClient()
        LOGGER.info(f'httpx AsyncClient instantiated. Id {id(self.async_client)}')

    async def stop(self):
        """ Gracefully shutdown. Call from FastAPI shutdown hook."""
        LOGGER.info(f'httpx async_client.is_closed(): {self.async_client.is_closed} - Now close it. Id (will be unchanged): {id(self.async_client)}')
        await self.async_client.aclose()
        LOGGER.info(f'httpx async_client.is_closed(): {self.async_client.is_closed}. Id (will be unchanged): {id(self.async_client)}')
        self.async_client = None
        LOGGER.info('httpx AsyncClient closed')

    def __call__(self):
        """ Calling the instantiated HTTPXClientWrapper returns the wrapped singleton."""
        # Ensure we don't use it if not started / running
        assert self.async_client is not None
        LOGGER.info(f'httpx async_client.is_closed(): {self.async_client.is_closed}. Id (will be unchanged): {id(self.async_client)}')
        return self.async_client


httpx_client_wrapper = HTTPXClientWrapper()
app = FastAPI()


@app.get('/test-call-external')
async def call_external_api(url: str = 'https://stackoverflow.com'):
    async_client = httpx_client_wrapper()
    res = await async_client.get(url)
    result = res.text
    return {
        'result': result,
        'status': res.status_code
    }


@app.on_event("startup")
async def startup_event():
    httpx_client_wrapper.start()


@app.on_event("shutdown")
async def shutdown_event():
    await httpx_client_wrapper.stop()


if __name__ == '__main__':
    import uvicorn
    LOGGER.info(f'starting...')
    uvicorn.run(f"{__name__}:app", host="127.0.0.1", port=8000)

Note - this answer was inspired by a similar answer I saw elsewhere a long time ago for aiohttp, I can't find the reference but thanks to whoever that was!

EDIT

I've added uvicorn bootstrapping in the example so that it's now fully functional. I've also added logging to show what's going on on startup and shutdown, and you can visit localhost:8000/docs to trigger the endpoint and see what happens (via the logs).

The reason for calling the start() method from the startup hook is that by the time the hook is called the eventloop has already started, so we know we will be instantiating the httpx client in an async context.

Also I was missing the async on the stop() method, and had a self.async_client = None instead of just async_client = None, so I have fixed those errors in the example.

Thaothapa

This is not sufficient. You need to instantiate the client in an async context since it depends on the event loop being available. – Ventral

The start() method is called from the startup hook, which only happens once the event loop is running. I've added the uvicorn bootstrapping so that it's now fully executable as a single file to make it easier to try out locally. In a real world app you would import the wrapper from anywhere in your app (from my_app.main import httpx_client_wrapper) and call it to get the client - just as is done in the call_external_api() route here. – Thaothapa

In a real world app, I would use a Depends injection for the client. – Ventral

Answer 2

The answer to this question depends on how you structure your FastAPI application and how you manage your dependencies. One possible way to use httpx.AsyncClient() is to create a custom dependency function that returns an instance of the client and closes it when the request is finished. For example:

from fastapi import FastAPI, Depends
import httpx

app = FastAPI()

async def get_client():
    # create a new client for each request
    async with httpx.AsyncClient() as client:
        # yield the client to the endpoint function
        yield client
        # close the client when the request is done

@app.get("/foo")
async def foo(client: httpx.AsyncClient = Depends(get_client)):
    # use the client to make some http requests, e.g.,
    response = await client.get("http://example.it")
    return response.json()

This way, you don't need to create a global client or worry about closing it manually. FastAPI will handle the dependency injection and the context management for you. You can also use the same dependency function for other endpoints that need to use the client.

Alternatively, you can create a global client and close it when the application shuts down. For example:

from fastapi import FastAPI, Depends
import httpx
import atexit

app = FastAPI()

# create a global client
client = httpx.AsyncClient()

# register a function to close the client when the app exits
atexit.register(client.aclose)

@app.get("/bar")
async def bar():
    # use the global client to make some http requests, e.g.,
    response = await client.get("http://example.it")
    return response.json()

This way, you don't need to create a new client for each request, but you need to make sure that the client is closed properly when the application stops. You can use the atexit module to register a function that will be called when the app exits, or you can use other methods such as signal handlers or event hooks.

Both methods have their pros and cons, and you should choose the one that suits your needs and preferences. You can also check out the FastAPI documentation on dependencies and testing for more examples and best practices.

Mound

But using get_client as dependency injection still is spawninig client each time you call foo() right? – Tb