FastAPI异步端点中同步CRUD操作的影响及解决方案咨询

我定义了一个异步FastAPI端点，负责将数据上传至S3：

@router.post("/upload", status_code=200, response_model=schemas.S3UploadSchema)
async def upload(
    file_name: str,
    data: str,
    s3_client=Depends(dependencies.get_s3_client),
    current_user: User = Depends(get_current_user)
):
    uploads3 = await s3_client.upload_fileobj(
        bucket=settings.RESULT_BUCKET,
        key=f"{settings.RESULT_FOLDER}/{file_name}",
        data=data
    )
    if uploads3:
        s3_url = f"https://{settings.RESULT_BUCKET}.s3.amazonaws.com/{settings.RESULT_FOLDER}/{file_name}"
        return {"status": "success",
                "image_url": s3_url}
    raise HTTPException(
        status_code=400, detail=f"Failed to upload '{file_name}' to S3.")

之后我在其中添加了CRUD操作（新增行已标注），修改后的代码如下：

@router.post("/upload", status_code=200, response_model=schemas.S3UploadSchema)
async def upload(
    file_name: str,
    data: str,
    s3_client=Depends(dependencies.get_s3_client),
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user)
):
    uploads3 = await s3_client.upload_fileobj(
        bucket=settings.RESULT_BUCKET,
        key=f"{settings.RESULT_FOLDER}/{file_name}",
        data=data
    )
    if uploads3:
        pattern = crud.create_pattern(db=db, pattern_name=file_name) # new line
        s3_url = f"https://{settings.RESULT_BUCKET}.s3.amazonaws.com/{settings.RESULT_FOLDER}/{file_name}"
        return {"status": "success",
                "image_url": s3_url,
                "pattern": pattern}
    raise HTTPException(
        status_code=400, detail=f"Failed to upload '{file_name}' to S3.")

执行CRUD操作使用的Session来自from sqlalchemy.orm import Session，我认为由于未使用异步Session，该CRUD操作会同步执行，导致服务器处理该请求时无法响应其他请求。请问我的理解是否正确？若正确，该如何处理这种情况？能否给出建议？

你的理解完全正确 #

FastAPI的异步端点运行在事件循环中，同步的数据库操作（比如使用SQLAlchemy的同步Session）会阻塞整个事件循环。一旦这个CRUD操作开始执行，服务器在这段时间内无法处理其他任何请求，直接降低了服务的并发能力。

处理方案建议 #

1. 改用SQLAlchemy异步Session #

这是最规范的长期解决方案：

• 替换同步Session为SQLAlchemy的异步Session（从sqlalchemy.ext.asyncio import AsyncSession导入）
• 调整数据库依赖get_db，返回异步Session实例
• 修改CRUD操作create_pattern，使用异步方法（比如await db.add(pattern_obj)、await db.commit()、await db.refresh(pattern_obj)）

示例调整后的关键代码片段：

# 依赖部分
from sqlalchemy.ext.asyncio import AsyncSession

async def get_db():
    async with async_session() as session:
        yield session

# CRUD操作示例
async def create_pattern(db: AsyncSession, pattern_name: str):
    db_pattern = models.Pattern(name=pattern_name)
    db.add(db_pattern)
    await db.commit()
    await db.refresh(db_pattern)
    return db_pattern

# 端点中的调用
pattern = await crud.create_pattern(db=db, pattern_name=file_name)

2. 将同步CRUD操作放入线程池执行 #

如果暂时不想重构整个数据库层为异步，可以用以下方式规避事件循环阻塞：

方式一：使用asyncio.to_thread

把同步操作放到单独线程中执行，不阻塞事件循环：

import asyncio

# 端点中的调用修改为
pattern = await asyncio.to_thread(crud.create_pattern, db=db, pattern_name=file_name)

注意：要确保同步Session在多线程环境下的安全性——SQLAlchemy的同步Session不是线程安全的，每个线程应该使用独立的Session实例。

方式二：使用BackgroundTasks（适合非关键路径操作）

如果这个CRUD操作不影响接口返回结果（比如不需要立即返回pattern数据），可以把它放到后台任务中，不阻塞当前请求：

from fastapi import BackgroundTasks

# 修改端点参数
async def upload(
    file_name: str,
    data: str,
    background_tasks: BackgroundTasks,
    s3_client=Depends(dependencies.get_s3_client),
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user)
):
    # ... 其他代码 ...
    if uploads3:
        background_tasks.add_task(crud.create_pattern, db=db, pattern_name=file_name)
        s3_url = f"https://{settings.RESULT_BUCKET}.s3.amazonaws.com/{settings.RESULT_FOLDER}/{file_name}"
        return {"status": "success",
                "image_url": s3_url}

3. 长期建议 #

优先选择迁移到异步数据库操作，更符合FastAPI异步架构的设计理念，能更好地发挥框架的并发优势。

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