如何在Flask+SQLAlchemy+PostgreSQL中实现加锁校验后插入行

PostgreSQL 的所有锁机制都和事务绑定，事务提交/回滚时会自动释放锁，无需手动编写释放锁的代码，只需控制好事务边界即可。以下是两种生产环境常用的正确实现方式：

方式1：行级写锁 + 事务（适合需要复杂前置校验的场景） #

前置要求 #

给你用于判断行是否存在的查询字段添加唯一索引/唯一约束，避免并发场景下的幻读问题。

代码示例 #

from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import text

db = SQLAlchemy()

# 示例模型，假设我们通过username判断行是否存在
class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    # 必须加唯一约束
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120))

def create_user_if_not_exists(username: str, email: str):
    # 显式开启事务，事务结束自动提交/回滚并释放锁
    with db.session.begin():
        # 步骤1+2合并：查询的同时给命中的行加写锁，等价于SQL: SELECT * FROM user WHERE username=? FOR UPDATE
        exist_user = User.query.filter_by(username=username).with_for_update().first()
        if exist_user:
            return exist_user, False
        
        # 步骤3：无匹配行则新增数据
        new_user = User(username=username, email=email)
        db.session.add(new_user)
        db.session.flush()
    
    return new_user, True

特殊说明 #

如果你的场景需要锁整张表（仅推荐低并发小表使用），可以在事务开始后执行表锁语句：

db.session.execute(text("LOCK TABLE user IN EXCLUSIVE MODE"))

方式2：PostgreSQL 原生UPSERT（推荐大部分场景使用） #

无需显式加锁，PostgreSQL 内部原子处理插入和冲突判断，性能远高于显式加锁方案，代码更简洁。

前置要求 #

和方式1一致，必须给冲突判断字段添加唯一索引/唯一约束。

代码示例 #

from sqlalchemy.dialects.postgresql import insert

def upsert_user(username: str, email: str):
    with db.session.begin():
        # 构造插入语句
        insert_stmt = insert(User).values(username=username, email=email)
        # 冲突时不执行操作，也可以根据需求用on_conflict_do_update更新已有字段
        on_conflict_stmt = insert_stmt.on_conflict_do_nothing(index_elements=['username'])
        
        result = db.session.execute(on_conflict_stmt)
        if result.rowcount > 0:
            # 新增成功
            new_user = User.query.filter_by(username=username).first()
            return new_user, True
        else:
            # 行已存在
            exist_user = User.query.filter_by(username=username).first()
            return exist_user, False

冲突时更新字段的写法示例 #

如果存在匹配行时需要更新对应字段，可以替换冲突处理逻辑：

on_conflict_stmt = insert_stmt.on_conflict_do_update(
    index_elements=['username'],
    set_=dict(email=insert_stmt.excluded.email)
)

通用注意事项 #

• 不要在事务内部执行耗时的非数据库操作，会导致锁长时间占用，引发性能问题甚至死锁
• 优先选择UPSERT方案，并发性能比显式加锁高30%以上
• 唯一约束是避免并发插入冲突的最后一道防线，两种方案都必须配置

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