101个FastAPI技巧作者：[FastAPI专家]

本仓库包含FastAPI的技巧和窍门。如果您有任何认为有用的技巧，欢迎开issue或提交pull request。

考虑在GitHub上赞助我以支持我的工作。有了您的支持，我将能够创作更多像这样的内容。

[!提示] 记得关注这个仓库以接收新技巧的通知。

1. 安装`uvloop`和`httptools`

默认情况下，[Uvicorn][uvicorn]不包含uvloop和httptools，它们比默认的asyncio事件循环和HTTP解析器更快。您可以使用以下命令安装它们：

pip install uvloop httptools

如果它们安装在您的环境中，[Uvicorn][uvicorn]会自动使用它们。

[!警告] uvloop无法在Windows上安装。如果您在本地使用Windows，但在生产环境中使用Linux，可以使用环境标记来避免在Windows上安装uvloop，例如：uvloop; sys_platform != 'win32'。

2. 谨慎使用非异步函数

在FastAPI中使用非异步函数会带来性能损失。因此，始终优先使用异步函数。性能损失来自于FastAPI将调用[run_in_threadpool][run_in_threadpool]，它会使用线程池运行函数。

[!注意] 内部实现中，[run_in_threadpool][run_in_threadpool]将使用[anyio.to_thread.run_sync][run_sync]在线程池中运行函数。

[!提示] 线程池中只有40个可用线程。如果您用完了所有线程，您的应用程序将被阻塞。

要更改可用线程的数量，您可以使用以下代码：

import anyio
from contextlib import asynccontextmanager
from typing import Iterator

from fastapi import FastAPI


@asynccontextmanager
async def lifespan(app: FastAPI) -> Iterator[None]:
    limiter = anyio.to_thread.current_default_thread_limiter()
    limiter.total_tokens = 100
    yield

app = FastAPI(lifespan=lifespan)

您可以在[AnyIO的文档][increase-threadpool]中阅读更多相关信息。

3. 在WebSocket中使用`async for`而不是`while True`

您在网上找到的大多数示例都使用while True来从WebSocket读取消息。

我认为使用这种不太优雅的表示法主要是因为Starlette文档长期以来没有展示async for表示法。

不要使用while True：

from fastapi import FastAPI
from starlette.websockets import WebSocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket) -> None:
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"消息内容是：{data}")

您可以使用async for表示法：

from fastapi import FastAPI
from starlette.websockets import WebSocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket) -> None:
    await websocket.accept()
    async for data in websocket.iter_text():
        await websocket.send_text(f"消息内容是：{data}")

您可以在[Starlette文档][websockets-iter-data]中阅读更多相关信息。

4. 忽略`WebSocketDisconnect`异常

如果您使用while True表示法，您需要捕获WebSocketDisconnect。 async for表示法会自动为您捕获它。

from fastapi import FastAPI
from starlette.websockets import WebSocket, WebSocketDisconnect

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket) -> None:
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"消息内容是：{data}")
    except WebSocketDisconnect:
        pass

如果您需要在WebSocket断开连接时释放资源，可以使用该异常来处理。

如果您使用的是旧版FastAPI，只有receive方法会引发WebSocketDisconnect异常。 send方法不会引发它。在最新版本中，所有方法都会引发该异常。在这种情况下，您需要将send方法也放在try块内。

5. 使用HTTPX的`AsyncClient`而不是`TestClient`

由于您在应用程序中使用async函数，使用HTTPX的AsyncClient而不是Starlette的TestClient会更容易。

from fastapi import FastAPI


app = FastAPI()


@app.get("/")
async def read_root():
    return {"Hello": "World"}


# 使用TestClient
from starlette.testclient import TestClient

client = TestClient(app)
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"Hello": "World"}

# 使用AsyncClient
import anyio
from httpx import AsyncClient, ASGITransport


async def main():
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        response = await client.get("/")
        assert response.status_code == 200
        assert response.json() == {"Hello": "World"}


anyio.run(main)

如果您使用生命周期事件（on_startup、on_shutdown或lifespan参数），可以使用[asgi-lifespan][asgi-lifespan]包来运行这些事件。

from contextlib import asynccontextmanager
from typing import AsyncIterator

import anyio
from asgi_lifespan import LifespanManager
from httpx import AsyncClient, ASGITransport
from fastapi import FastAPI


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    print("启动应用")
    yield
    print("停止应用")


app = FastAPI(lifespan=lifespan)


@app.get("/")
async def read_root():
    return {"Hello": "World"}


async def main():
    async with LifespanManager(app) as manager:
        async with AsyncClient(transport=ASGITransport(app=manager.app)) as client:
            response = await client.get("/")
            assert response.status_code == 200
            assert response.json() == {"Hello": "World"}


anyio.run(main)

[!注意] 考虑通过GitHub Sponsors支持[asgi-lifespan][asgi-lifespan]的创建者[Florimond Manca][florimondmanca]。

6. 使用生命周期状态而不是`app.state`

不久前，FastAPI开始支持[生命周期状态]，它定义了一种标准方式来管理需要在启动时创建并在请求-响应周期中使用的对象。

不再推荐使用app.state。您应该改用[生命周期状态]。

使用app.state时，您会这样做：

from contextlib import asynccontextmanager
from typing import AsyncIterator

from fastapi import FastAPI, Request
from httpx import AsyncClient


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    async with AsyncClient(app=app) as client:
        app.state.client = client
        yield


app = FastAPI(lifespan=lifespan)


@app.get("/")
async def read_root(request: Request):
    client = request.app.state.client
    response = await client.get("/")
    return response.json()

使用生命周期状态时，您会这样做：

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import Any, TypedDict, cast

from fastapi import FastAPI, Request
from httpx import AsyncClient


class State(TypedDict):
    client: AsyncClient


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[State]:
    async with AsyncClient(app=app) as client:
        yield {"client": client}


app = FastAPI(lifespan=lifespan)


@app.get("/")
async def read_root(request: Request) -> dict[str, Any]:
    client = cast(AsyncClient, request.state.client)
    response = await client.get("/")
    return response.json()

7. 启用AsyncIO调试模式

如果您想找出阻塞事件循环的端点，可以启用AsyncIO调试模式。

启用后，当一个任务执行时间超过100毫秒时，Python会打印一条警告消息。

使用PYTHONASYNCIODEBUG=1 python main.py运行以下代码：

import os
import time

import uvicorn
from fastapi import FastAPI


app = FastAPI()


@app.get("/")
async def read_root():
    time.sleep(1)  # 阻塞调用
    return {"Hello": "World"}


if __name__ == "__main__":
    uvicorn.run(app, loop="uvloop")

如果您调用该端点，您将看到以下消息：

INFO:     Started server process [19319]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     127.0.0.1:50036 - "GET / HTTP/1.1" 200 OK
Executing <Task finished name='Task-3' coro=<RequestResponseCycle.run_asgi() done, defined at /uvicorn/uvicorn/protocols/http/httptools_impl.py:408> result=None created at /uvicorn/uvicorn/protocols/http/httptools_impl.py:291> took 1.009 seconds

您可以在官方文档中阅读更多相关信息。

8. 实现纯ASGI中间件而不是`BaseHTTPMiddleware`

[BaseHTTPMiddleware][base-http-middleware]是在FastAPI中创建中间件最简单的方法。

[!注意] @app.middleware("http")装饰器是BaseHTTPMiddleware的包装器。

BaseHTTPMiddleware曾存在一些问题，但最新版本中大多数问题已经修复。尽管如此，使用它仍然会带来性能损失。

为了避免性能损失，您可以实现[纯ASGI中间件]。缺点是实现起来更复杂。查看Starlette的文档了解如何实现[纯ASGI中间件]。

9. 你的依赖项可能在线程上运行

如果函数是非异步的，并且你将其用作依赖项，它将在线程中运行。

在以下示例中，http_client函数将在线程中运行：

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from httpx import AsyncClient
from fastapi import FastAPI, Request, Depends


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[dict[str, AsyncClient]]:
    async with AsyncClient() as client:
        yield {"client": client}


app = FastAPI(lifespan=lifespan)


def http_client(request: Request) -> AsyncClient:
    return request.state.client


@app.get("/")
async def read_root(client: AsyncClient = Depends(http_client)):
    return await client.get("/")

要在事件循环中运行，你需要将函数改为异步：

# ...

async def http_client(request: Request) -> AsyncClient:
    return request.state.client

# ...

作为读者的练习，让我们进一步了解如何检查正在运行的线程。

你可以使用python main.py运行以下代码：

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

import anyio
from anyio.to_thread import current_default_thread_limiter
from httpx import AsyncClient
from fastapi import FastAPI, Request, Depends


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[dict[str, AsyncClient]]:
    async with AsyncClient() as client:
        yield {"client": client}


app = FastAPI(lifespan=lifespan)


# 将此函数改为异步，并重新运行此应用程序。
def http_client(request: Request) -> AsyncClient:
    return request.state.client


@app.get("/")
async def read_root(client: AsyncClient = Depends(http_client)): ...


async def monitor_thread_limiter():
    limiter = current_default_thread_limiter()
    threads_in_use = limiter.borrowed_tokens
    while True:
        if threads_in_use != limiter.borrowed_tokens:
            print(f"使用中的线程数: {limiter.borrowed_tokens}")
            threads_in_use = limiter.borrowed_tokens
        await anyio.sleep(0)


if __name__ == "__main__":
    import uvicorn

    config = uvicorn.Config(app="main:app")
    server = uvicorn.Server(config)

    async def main():
        async with anyio.create_task_group() as tg:
            tg.start_soon(monitor_thread_limiter)
            await server.serve()

    anyio.run(main)

如果你调用该端点，你将看到以下消息：

❯ python main.py
INFO:     Started server process [23966]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
使用中的线程数: 1
INFO:     127.0.0.1:57848 - "GET / HTTP/1.1" 200 OK
使用中的线程数: 0