FastAPI 异步编程：提升 API 性能

在 Web 开发中，性能至关重要，尤其是在同时处理多个请求时。FastAPI 专为速度和性能而设计，允许开发人员利用异步编程来构建响应式、非阻塞的 API。本文深入探讨了使用 FastAPI 进行异步编程，并向您展示了如何通过并发处理请求来优化 API 性能。

为什么要异步编程？

异步编程允许部分代码同时运行，而不是等待每个任务完成后再继续。这对于 Web 应用程序特别有用，因为请求可能涉及长时间运行的任务，例如数据库操作或外部 API 调用。异步代码可缩短响应时间，使服务器能够同时处理多个请求而不会阻塞。

FastAPI 支持使用async和await关键字的异步编程，这使其成为高性能应用程序的理想框架。

在 FastAPI 中设置异步端点

要实现异步端点，请使用async def端点函数的语法。以下是如何在 FastAPI 中定义简单异步端点的示例。

from fastapi import FastAPI

import asyncio



app = FastAPI()



@app.get("/async-example")

async def async_example():

    await asyncio.sleep(2)  # Simulate an I/O operation

    return {"message": "This endpoint is asynchronous!"}

在此示例中：

async_example被定义为使用的异步函数async def。
await asyncio.sleep(2)模拟阻塞 I/O 操作，这不会阻止服务器处理其他请求。

这意味着当服务器等待asyncio.sleep(2)完成时，它可以处理其他传入的请求。

使用异步函数进行数据库访问

在 FastAPI 应用程序中访问数据库时，建议使用异步数据库驱动程序以避免阻塞操作。以下是如何使用 FastAPI 设置异步数据库查询的示例。

from fastapi import FastAPI

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine

from sqlalchemy.orm import sessionmaker



# Setup database connection

DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"

engine = create_async_engine(DATABASE_URL, echo=True)

async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)



app = FastAPI()



@app.get("/items/")

async def get_items():

    async with async_session() as session:

        result = await session.execute("SELECT * FROM items")

        items = result.fetchall()

        return {"items": items}

在此设置中：

我们使用create_async_enginefromsqlalchemy.ext.asyncio进行异步数据库连接。
我们在一个块内访问数据库async with，当函数遇到任何阻塞操作时，允许它放弃控制。

与后台任务并行处理

对于不需要立即响应客户端的操作（例如发送电子邮件或处理大量数据），FastAPI 提供了一项BackgroundTasks功能。这允许您在向用户发送响应后卸载要运行的任务。

from fastapi import FastAPI, BackgroundTasks



app = FastAPI()



def write_log(message: str):

    with open("log.txt", "a") as log:

        log.write(message + "\n")



@app.post("/send-notification/")

async def send_notification(email: str, background_tasks: BackgroundTasks):

    background_tasks.add_task(write_log, f"Notification sent to {email}")

    return {"message": f"Notification scheduled for {email}"}

在此示例中：