并发更新错误：FastAPI中的竞态条件错误

编程语言 2026-07-10

我有一个函数，它更新数据库

@router.put("/{task_id}/status")
def update_status(task_id: str, task_data: UpdateStatus, db: Session = Depends(get_db)):
    task = retry_filter_by_id(db, model_name=Task, id=task_id)
    if task:
        task.external_status = task_data.status
        task.external_status_update_time = datetime.now()
        retry_commit(db)
        return JSONResponse(content={
            "status": "Updated",
            "task_id": task_id
        })
    raise HTTPException(status_code=400, detail=f"Task {task_id} not found")

# retry functions
@retry_db_operation(max_retries=3)
def retry_filter_by_id(db, model_name, id: str):
    return db.query(model_name).filter_by(id=id).one_or_none()

@retry_db_operation(max_retries=3)
def retry_commit(db):
    db.commit()

避免竞争条件的正确做法是什么？

我的解决方案是

@retry_db_operation(max_retries=3)
def retry_filter_by_id(db, model_name, id: str, lock=False):
    if not lock:
        return db.query(model_name).filter_by(id=id).one_or_none()
    return db.query(model_name).filter_by(id=id).with_for_update().one_or_none()

解决方案

没有一个单一的答案：解决方案取决于正在应用哪种更新，以及应用在什么样的约束条件下运行。

在本例中，似乎你只有一个包含两列且经常被更新的表，这会导致并发更新错误。一种处理方法可能是把这两列移动到一个单独的表中，并将每个新状态插入到该表，而不是更新单行记录。由于你现在只进行 INSERT，而不是 UPDATE，你将不再出现并发更新的问题。当然，这样做的代价是现在必须查询该表以获取最新状态，并且你必须设计一种方法定期清理旧记录，以防止表变得过大。

由于你使用SQLAlchemy，你可以做一些技巧来创建一个关系，以获取任务的最新状态（请注意它是只读的：必须把新状态追加到任务的 statuses 属性中）。

import datetime as dt

import sqlalchemy as sa
from sqlalchemy import orm
from sqlalchemy.orm import Mapped, mapped_column


class Base(orm.DeclarativeBase):
    pass


class TaskStatus(Base):
    __tablename__ = 'task_statuses'

    id: Mapped[int] = mapped_column(primary_key=True)
    status: Mapped[str]
    update_time: Mapped[dt.datetime]
    task_id: Mapped[int] = mapped_column(sa.ForeignKey('tasks.id'))

    task: Mapped['Task'] = orm.relationship('Task', back_populates='statuses')


class Task(Base):
    __tablename__ = 'tasks'

    id: Mapped[int] = mapped_column(primary_key=True)
    statuses: Mapped[list['TaskStatus']] = orm.relationship('TaskStatus', back_populates='task')

    @orm.declared_attr
    def __mapper_args__(cls) -> dict:
        statuses = TaskStatus.__table__
        most_recent_child = (
            sa.select(statuses.c.id)
            .where(statuses.c.task_id == cls.id)
            .order_by(statuses.c.update_time.desc())
            .limit(1)
            .correlate(cls.__table__)
            .scalar_subquery()
        )

        rel: Mapped[TaskStatus] = orm.relationship(
            TaskStatus,
            primaryjoin=sa.and_(
                TaskStatus.id == most_recent_child, TaskStatus.task_id == cls.id
            ),
            uselist=False,
            viewonly=True,
        )
        return {'properties': {'latest_status': rel}}


engine = sa.create_engine('sqlite://', echo=True)
Base.metadata.create_all(engine)
Session = orm.sessionmaker(engine)

with Session.begin() as s:
    task = Task()
    s.add(task)
    for status in ['first', 'second', 'third']:
        ts = TaskStatus(status=status, update_time=dt.datetime.now())
        task.statuses.append(ts)

with Session() as s:
    task = s.scalar(sa.select(Task))
    assert task.latest_status.status == 'third'

    task.statuses.append(TaskStatus(status='fourth', update_time=dt.datetime.now()))

    s.commit()

    assert len(task.statuses) == 4
    assert task.latest_status.status == 'fourth'

（这段代码是我在这里的答案基础上定制的版本）

error