如何在Flask-SQLAlchemy（PostgreSQL）中创建可空自增列？

针对你的需求，这里提供两种可行的实现方式：

方式一：数据库层面（触发器+序列） #

利用PostgreSQL的序列和触发器，在插入数据时自动为column赋值，这种方式性能高且能避免并发冲突。

步骤1：创建自定义序列 #

创建一个初始值为0的序列，满足你示例中第一个非NULL值为0的需求：

CREATE SEQUENCE user_column_seq START WITH 0 INCREMENT BY 1;

步骤2：编写触发器函数 #

定义一个函数，当插入user表且column为NULL时，从序列中取值赋值：

CREATE OR REPLACE FUNCTION set_user_column()
RETURNS TRIGGER AS $$
BEGIN
    IF NEW.column IS NULL THEN
        NEW.column := nextval('user_column_seq');
    END IF;
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

步骤3：绑定触发器到User表 #

为user表添加BEFORE INSERT触发器，触发上述函数：

CREATE TRIGGER trigger_set_user_column
BEFORE INSERT ON "user"
FOR EACH ROW EXECUTE FUNCTION set_user_column();

在Flask-SQLAlchemy中初始化 #

你可以在应用启动时执行上述SQL，避免手动操作数据库：

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import text

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://user:password@localhost/dbname'
db = SQLAlchemy(app)

class User(db.Model):
    id: Mapped[int] = mapped_column(primary_key=True)
    column: Mapped[int] = mapped_column(nullable=True, unique=True)

# 应用启动时创建序列和触发器（仅当不存在时）
with app.app_context():
    seq_exists = db.engine.execute(text("SELECT EXISTS (SELECT 1 FROM pg_class WHERE relname='user_column_seq')")).scalar()
    if not seq_exists:
        db.engine.execute(text("CREATE SEQUENCE user_column_seq START WITH 0 INCREMENT BY 1"))
        db.engine.execute(text("""
            CREATE OR REPLACE FUNCTION set_user_column()
            RETURNS TRIGGER AS $$
            BEGIN
                IF NEW.column IS NULL THEN
                    NEW.column := nextval('user_column_seq');
                END IF;
                RETURN NEW;
            END;
            $$ LANGUAGE plpgsql;
        """))
        db.engine.execute(text("""
            CREATE TRIGGER trigger_set_user_column
            BEFORE INSERT ON "user"
            FOR EACH ROW EXECUTE FUNCTION set_user_column();
        """))

方式二：SQLAlchemy事件监听 #

通过SQLAlchemy的before_insert事件，在Python层面自动计算并赋值column。注意需处理并发场景，建议加表锁避免重复值。

代码实现 #

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import event, text

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://user:password@localhost/dbname'
db = SQLAlchemy(app)

class User(db.Model):
    id: Mapped[int] = mapped_column(primary_key=True)
    column: Mapped[int] = mapped_column(nullable=True, unique=True)

@event.listens_for(User, 'before_insert')
def set_user_column_before_insert(mapper, connection, target):
    if target.column is None:
        # 加表锁防止并发插入时重复赋值
        connection.execute(text("LOCK TABLE \"user\" IN SHARE ROW EXCLUSIVE MODE"))
        # 查询当前最大的column值
        max_col = connection.execute(text("SELECT MAX(column) FROM \"user\"")).scalar()
        target.column = 0 if max_col is None else max_col + 1

注意事项 #

• 方式二在高并发场景下可能存在性能瓶颈，因为加了表锁；方式一的触发器+序列是PostgreSQL原生支持的，性能更优。
• 两种方式都能保证column的唯一性，因为模型中已设置unique=True，数据库会自动校验重复值。

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