如何实现FastAPI接口并行/异步处理多请求？

首先得指出你当前代码的两个核心问题，它们会直接影响多请求处理的效率和安全性：

• 直接用format拼接用户输入到SQL语句中，存在严重的SQL注入风险，绝对不能这么做；
• session.execute是同步IO操作，放在async接口函数里会阻塞FastAPI的事件循环，导致其他请求只能排队等待，根本没法实现并行处理。

下面是具体的优化方案：

1. 优先使用异步数据库驱动 #

如果你的数据库支持异步驱动（比如Cassandra官方已经提供异步版本），直接切换成异步操作，这样数据库查询不会阻塞事件循环，ASGI服务器就能同时处理更多请求。

以Cassandra为例，优化后的代码：

from cassandra.cluster import AsyncCluster
from cassandra.query import SimpleStatement
from fastapi import FastAPI, status
from fastapi.responses import JSONResponse
from fastapi.encoders import jsonable_encoder

app = FastAPI()

# 启动时初始化异步数据库连接
@app.on_event("startup")
async def startup_event():
    app.state.async_cluster = AsyncCluster(nodes)  # nodes是你的Cassandra节点列表
    app.state.async_session = await app.state.async_cluster.connect("testbd")

# 关闭时释放资源
@app.on_event("shutdown")
async def shutdown_event():
    await app.state.async_session.shutdown()
    await app.state.async_cluster.shutdown()

@app.get("/data/{user}")
async def getusername(user: str):
    # 使用参数化查询避免SQL注入
    query = SimpleStatement("SELECT * FROM testbd WHERE user = %s")
    # 异步执行查询，不阻塞事件循环
    rows = await app.state.async_session.execute(query, (user,))
    data = [dict(row) for row in rows]  # 转换为可序列化的字典
    return JSONResponse(content=jsonable_encoder(data), status_code=status.HTTP_200_OK)

2. 无法切换异步驱动？用线程池执行同步操作 #

如果暂时没法用异步驱动，就把同步的数据库操作放到线程池里执行，避免阻塞事件循环：

from cassandra.cluster import Cluster
from fastapi import FastAPI, status
from fastapi.responses import JSONResponse
from fastapi.encoders import jsonable_encoder
import asyncio

app = FastAPI()

# 初始化同步集群和会话（Cassandra的session是线程安全的，全局复用没问题）
cluster = Cluster(nodes)
session = cluster.connect("testbd")

@app.get("/data/{user}")
async def getusername(user: str):
    # 封装同步查询逻辑
    def sync_query():
        query = "SELECT * FROM testbd WHERE user = %s"
        rows = session.execute(query, (user,))
        return [dict(row) for row in rows]
    
    # 把同步操作放到线程池 separating
Same invalidate UpdateOrange刑美告知煜准备.goss监控线程池执行，不阻塞事件循环
    data = await asyncio.to_thread(sync_query)
    return JSONResponse(content=jsonable_encoder(data), status_code=status.HTTP_200_OK)

3. 关键注意事项 #

• 必须用参数化查询：永远不要直接拼接用户输入到SQL语句中，这是防范SQL注入的基础；
• 部署用ASGI服务器：用Uvicorn、Hypercorn这类原生支持异步的ASGI服务器，搭配Gunicorn时也要用Uvicorn worker，才能发挥FastAPI的并发优势；
• 合理管理数据库连接：不要为每个请求新建连接，复用全局会话或使用连接池，降低连接开销。

内容的提问来源于stack exchange，提问作者Hmm