如何在 FastAPI 应用程序中正确重用 httpx.AsyncClient？

Question

我有一个 FastAPI 应用程序，在多种不同的场合，需要调用外部 API。我使用 httpx.AsyncClient 进行这些调用。关键是我不完全理解应该如何使用它。

从httpx 的文档中我应该使用上下文管理器，

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")

但是，我明白，通过这种方式，每次我调用 时都会生成一个新客户端foo()，而这正是我们首先希望通过使用客户端来避免的情况。

我想另一种选择是在某处定义一些全局客户端，然后在需要时导入它，就像这样

aclient = httpx.AsyncClient()

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

不过，第二个选项看起来有点可疑，因为没有人负责关闭会话等。

所以问题是：如何httpx.AsyncClient()在 FastAPI 应用程序中正确（重新）使用？

Answer 1

您可以拥有一个在 FastApi 关闭事件中关闭的全局客户端。

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)

注意 - 这个答案的灵感来自于我很久以前在其他地方看到的类似答案aiohttp，我找不到参考资料，但感谢那个人！

编辑

我在示例中添加了 uvicorn bootstrapping，以便它现在功能齐全。我还添加了日志记录以显示启动和关闭时发生的情况，您可以访问localhost:8000/docs以触发端点并查看发生了什么（通过日志）。

从启动挂钩调用该方法的原因start()是，当调用该挂钩时，事件循环已经启动，因此我们知道我们将在异步上下文中实例化 httpx 客户端。

另外，我错过了方法async上的stop()，并使用了 aself.async_client = None而不是 just async_client = None，所以我在示例中修复了这些错误。

Answer 2

这个问题的答案取决于您如何构建 FastAPI 应用程序以及如何管理依赖项。使用 httpx.AsyncClient() 的一种可能方法是创建一个自定义依赖函数，该函数返回客户端的实例并在请求完成时关闭它。例如：

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()

这样，您无需创建全局客户端或担心手动关闭它。FastAPI 将为您处理依赖注入和上下文管理。您还可以对需要使用客户端的其他端点使用相同的依赖函数。

或者，您可以创建一个全局客户端并在应用程序关闭时将其关闭。例如：

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()

这样，您不需要为每个请求创建一个新的客户端，但需要确保在应用程序停止时正确关闭客户端。您可以使用 atexit 模块注册一个在应用程序退出时调用的函数，也可以使用其他方法，例如信号处理程序或事件挂钩。

两种方法都有其优点和缺点，您应该选择适合您的需要和偏好的一种。您还可以查看有关依赖项和测试的 FastAPI 文档，以获取更多示例和最佳实践。