PostgreSQL+SQLAlchemy API并发请求数据一致性问题求助

你遇到的是典型的并发竞态条件问题：多个请求同时执行「检查是否存在→插入新数据」的操作，由于默认事务隔离级别（READ COMMITTED）下，每个事务只能看到已提交的数据，导致后续请求看不到前序请求未提交的插入操作，最终重复插入数据。

问题根源分析 #

你的代码里，select检查和session.add虽然在同一个事务中，但READ COMMITTED隔离级别下，事务的查询只能看到事务启动前已提交的快照。当第一个请求还未提交插入时，第二个请求的事务依然会查询到「无对应条目」，进而执行插入，最终两个事务提交后产生重复数据。

可行解决方案 #

1. 数据库层面加唯一约束（最根本的防护）

给bearer_token字段添加唯一约束，让数据库直接阻止重复数据插入，这是最可靠的底层防护，即使代码逻辑有漏洞，数据库也会抛出IntegrityError：

from sqlalchemy import Column, String, Boolean, UUID
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Token(Base):
    __tablename__ = "tokens"
    id = Column(UUID, primary_key=True)
    bearer_token = Column(String, unique=True, nullable=False)  # 添加唯一约束
    valid = Column(Boolean, default=True)

之后执行数据库迁移，让PostgreSQL生效这个约束。业务代码中可以捕获IntegrityError，返回「数据已存在」的提示。

2. 使用PostgreSQL原子插入语法（推荐方案）

用INSERT ... ON CONFLICT语法把「检查+插入」合并为一个原子操作，避免竞态条件。SQLAlchemy对PostgreSQL的该语法有原生支持：

from sqlalchemy.dialects.postgresql import insert
import uuid

async def insert(session, bearer_token):
    # 构造插入语句，冲突时忽略
    stmt = insert(Token).values(
        bearer_token=bearer_token,
        id=uuid.uuid1(),
        valid=True
    ).on_conflict_do_nothing(index_elements=["bearer_token"])
    
    async with session.begin():
        await session.execute(stmt)

这个操作完全在数据库层面原子执行，不需要先查询再判断，性能和可靠性都远高于原逻辑。

3. 调整事务隔离级别+重试机制（你当前的方案优化）

你使用的SERIALIZABLE隔离级别会强制事务串行执行，避免并发冲突，但可能降低性能，且会抛出SerializationFailure异常，需要配合重试逻辑：

from sqlalchemy.exc import SerializationFailure
import uuid
from sqlalchemy import select

async def insert(session, bearer_token):
    newdbrow = Token()
    entity = {"bearer_token": bearer_token, 'id': uuid.uuid1(), 'valid': True}
    for key, value in entity.items():
        setattr(newdbrow, key, value)
    
    while True:
        try:
            async with session.begin():
                row = await session.scalar(select(Token).filter_by(bearer_token=bearer_token))
                if row is None:
                    session.add(newdbrow)
            break
        except SerializationFailure:
            # 事务冲突，重试操作
            continue

这种方案适合并发量不高的场景，若并发量大，推荐优先用前两种方案。

内容的提问来源于stack exchange，提问作者Nhật Nguyễn