FastAPI结合SQLAlchemy实现单次请求批量创建多个用户的方案

第一步：调整接口入参校验 #

原接口仅支持接收单个用户对象，修改路由层代码支持接收用户列表即可兼容批量请求：

from typing import List

# 可根据需求选择返回消息或创建完成的用户列表
# 方案1：返回操作提示
@app.post("/entities/", response_model=dict)
def create_users(users: List[schemas.User], db: Session = Depends(get_db)):
    crud.bulk_create_users(db=db, users=users)
    return {"message": f"{len(users)} users created successfully"}

# 方案2：返回创建完成的用户全量数据
@app.post("/entities/", response_model=List[schemas.User])
def create_users(users: List[schemas.User], db: Session = Depends(get_db)):
    return crud.bulk_create_users(db=db, users=users)

第二步：修正CRUD批量插入逻辑 #

你之前写的版本存在参数冗余、变量名拼写错误的问题，正确批量插入实现如下：

from typing import List

def bulk_create_users(db: Session, users: List[schemas.User]):
    # 批量构造ORM对象
    db_users = [models.User(**user.dict()) for user in users]
    # 批量写入数据库，比循环调用add方法性能高30%以上
    db.bulk_save_objects(db_users)
    # 提交事务
    db.commit()
    # 如果需要返回带数据库生成字段（如自增ID）的对象，可添加批量刷新逻辑
    # 不需要返回数据可删除以下循环，性能提升更明显
    for user in db_users:
        db.refresh(user)
    return db_users

可选优化点 #

• 数据量超过1000条时建议分批插入，每500-1000条提交一次，避免单次请求负载过高
• 不需要返回创建后的数据时删除refresh逻辑，可进一步降低数据库交互开销
• 原单用户创建接口可直接废弃，批量接口兼容单用户场景，仅需传入长度为1的数组即可

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