如何在SQLAlchemy中处理多会话并发竞争问题

你的问题核心是检查邮箱是否存在和创建用户这两步不是原子操作，导致并发请求绕过检查，触发数据库唯一约束后返回500错误。下面是几种靠谱的解决办法：

1. 捕获数据库唯一约束异常，返回正确的409响应 #

数据库已经给email字段加了unique约束，这是最可靠的防线。直接尝试创建用户，捕获插入时的完整性异常，转换成业务层面的"邮箱已被占用"提示即可，不需要提前检查。

修改CRUD的create函数：

from sqlalchemy.exc import IntegrityError
from fastapi import HTTPException

async def create(db: AsyncSession, user_scheme: UserCreate) -> User:
    user = User(
        **user_scheme.model_dump(exclude={"password"}),
        hashed_password=hash_password(user_scheme.password),
    )
    db.add(user)
    try:
        await db.commit()
        await db.refresh(user)  # 刷新获取数据库生成的字段（比如id）
        return user
    except IntegrityError as e:
        await db.rollback()
        # 判断是否是邮箱唯一约束冲突（不同数据库错误码不同，PostgreSQL是23505）
        if hasattr(e.orig, 'pgcode') and e.orig.pgcode == "23505":
            raise HTTPException(status_code=409, detail="Email is already taken")
        # MySQL可以用以下判断：
        # elif "Duplicate entry" in str(e.orig):
        #     raise HTTPException(status_code=409, detail="Email is already taken")
        raise  # 其他异常重新抛出

然后路由可以去掉前置检查，直接调用创建函数：

@router.post("/", status_code=201, response_model=UserRead)
async def register_user(db: SessionDepends, user_scheme: UserCreate) -> User:
    return await crud.user.create(db, user_scheme)

2. 用数据库原子操作实现"创建或检查" #

如果想在数据库层面一步完成检查和创建，可以用INSERT ... ON CONFLICT（PostgreSQL）或者类似语法，保证操作的原子性：

from sqlalchemy import insert, select

async def create_user(db: AsyncSession, user_scheme: UserCreate) -> User:
    # 插入用户，邮箱冲突时什么都不做
    insert_stmt = insert(User).values(
        email=user_scheme.email,
        hashed_password=hash_password(user_scheme.password)
    ).on_conflict_do_nothing(index_elements=["email"])
    
    result = await db.execute(insert_stmt)
    if result.rowcount == 1:
        # 插入成功，查询并返回用户
        return await db.scalar(select(User).where(User.email == user_scheme.email))
    else:
        # 邮箱已存在，抛出异常
        raise HTTPException(status_code=409, detail="Email is already taken")

为什么前置检查没用？ #

你原来的代码里先检查再创建，本质是"读取-修改-写入"的非原子操作。两个并发Session在读取阶段都看到邮箱未被占用，然后先后执行写入，必然有一个触发约束报错。数据库的约束是唯一能保证全局一致性的手段，前置检查只能作为优化（提前给用户反馈），不能替代约束和异常处理。

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