Logo
Project Icon

fastapi-tips

FastAPI开发技巧集锦，提升应用效率和质量的101条建议

FastAPI异步性能优化中间件WebSocketGithub开源项目

该项目汇集了101个FastAPI开发技巧，内容涉及性能优化、异步编程、WebSocket处理和测试方法等关键领域。项目介绍了依赖安装优化、异步函数使用、WebSocket代码改进、测试客户端选择以及生命周期状态管理等实用方法。这些建议旨在帮助开发者提高FastAPI应用的性能和代码质量，是一个全面且实用的技术参考资源。

访问官网访问官网GithubGithub文档文档论文论文
介绍相关项目

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

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

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

在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

将def http_client替换为async def http_client并重新运行应用程序。 你将不会看到使用中的线程数: 1的消息，因为该函数在事件循环中运行。

[!提示] 你可以使用我构建的[FastAPI Dependency]包，明确指定依赖项何时应该在线程中运行。

相关项目
项目侧边栏1项目侧边栏2
推荐项目
Project Cover

豆包MarsCode

豆包 MarsCode 是一款革命性的编程助手，通过AI技术提供代码补全、单测生成、代码解释和智能问答等功能，支持100+编程语言，与主流编辑器无缝集成，显著提升开发效率和代码质量。

Project Cover

AI写歌

Suno AI是一个革命性的AI音乐创作平台，能在短短30秒内帮助用户创作出一首完整的歌曲。无论是寻找创作灵感还是需要快速制作音乐，Suno AI都是音乐爱好者和专业人士的理想选择。

Project Cover

有言AI

有言平台提供一站式AIGC视频创作解决方案，通过智能技术简化视频制作流程。无论是企业宣传还是个人分享，有言都能帮助用户快速、轻松地制作出专业级别的视频内容。

Project Cover

Kimi

Kimi AI助手提供多语言对话支持，能够阅读和理解用户上传的文件内容，解析网页信息，并结合搜索结果为用户提供详尽的答案。无论是日常咨询还是专业问题，Kimi都能以友好、专业的方式提供帮助。

Project Cover

阿里绘蛙

绘蛙是阿里巴巴集团推出的革命性AI电商营销平台。利用尖端人工智能技术，为商家提供一键生成商品图和营销文案的服务，显著提升内容创作效率和营销效果。适用于淘宝、天猫等电商平台，让商品第一时间被种草。

Project Cover

吐司

探索Tensor.Art平台的独特AI模型，免费访问各种图像生成与AI训练工具，从Stable Diffusion等基础模型开始，轻松实现创新图像生成。体验前沿的AI技术，推动个人和企业的创新发展。

Project Cover

SubCat字幕猫

SubCat字幕猫APP是一款创新的视频播放器，它将改变您观看视频的方式！SubCat结合了先进的人工智能技术，为您提供即时视频字幕翻译，无论是本地视频还是网络流媒体，让您轻松享受各种语言的内容。

Project Cover

美间AI

美间AI创意设计平台，利用前沿AI技术，为设计师和营销人员提供一站式设计解决方案。从智能海报到3D效果图，再到文案生成，美间让创意设计更简单、更高效。

Project Cover

稿定AI

稿定设计 是一个多功能的在线设计和创意平台，提供广泛的设计工具和资源，以满足不同用户的需求。从专业的图形设计师到普通用户，无论是进行图片处理、智能抠图、H5页面制作还是视频剪辑，稿定设计都能提供简单、高效的解决方案。该平台以其用户友好的界面和强大的功能集合，帮助用户轻松实现创意设计。

使用协议隐私政策广告服务
投诉举报邮箱: service@vectorlightyear.com
@2024 懂AI·鲁ICP备2024100362号-6·鲁公网安备37021002001498号