如何通过SQLAlchemy实现按分组自增的version列？

问题背景 #

使用SQLAlchemy ORM操作PostgreSQL数据库，现有Network模型如下：

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

class Network(Base):
    __tablename__ = 'networks'

    id: Mapped[int] = mapped_column(sa.Integer, primary_key=True)
    name: Mapped[str] = mapped_column(sa.String(60))
    type: Mapped[str] = mapped_column(sa.String(60))
    version: Mapped[int] = mapped_column(sa.Integer)

需求：插入新行时，若存在相同name和type的记录，新行的version为该组最大version+1；若不存在，则version默认值为1。例如已有数据：

id    name    type    version
0     foo     big     1
1     bar     big     1

插入新的(foo, big)行后，新增行的version应为2。

解决方案 #

方案一：数据库触发器实现（推荐，无并发冲突） #

通过PostgreSQL的函数和触发器实现，这是最可靠的方式，能在数据库层面保证逻辑正确性，避免并发插入时的版本重复问题。

1. 创建计算版本号的PL/pgSQL函数：

CREATE OR REPLACE FUNCTION set_network_version()
RETURNS TRIGGER AS $$
BEGIN
    SELECT COALESCE(MAX(version), 0) + 1
    INTO NEW.version
    FROM networks
    WHERE name = NEW.name AND type = NEW.type;
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

2. 创建绑定到表的前置触发器：

CREATE TRIGGER trigger_set_network_version
BEFORE INSERT ON networks
FOR EACH ROW
EXECUTE FUNCTION set_network_version();

配置完成后，插入新行时数据库会自动计算并填充version字段，ORM层无需额外处理。

方案二：SQLAlchemy ORM事件监听（需注意并发） #

如果希望在ORM层处理逻辑，可以通过before_insert事件监听实现，但高并发场景下可能出现版本重复（多个请求同时查询最大值后插入，会导致相同version）。

修改Network模型并添加事件监听：

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

class Network(Base):
    __tablename__ = 'networks'

    id: Mapped[int] = mapped_column(sa.Integer, primary_key=True)
    name: Mapped[str] = mapped_column(sa.String(60))
    type: Mapped[str] = mapped_column(sa.String(60))
    version: Mapped[int] = mapped_column(sa.Integer)

@event.listens_for(Network, 'before_insert')
def set_version_before_insert(mapper, connection, target):
    # 查询同(name, type)组的最大version
    max_version = connection.execute(
        sa.select(sa.func.max(Network.version))
        .where(Network.name == target.name)
        .where(Network.type == target.type)
    ).scalar()
    target.version = (max_version or 0) + 1

若使用此方案，高并发场景建议搭配SERIALIZABLE事务隔离级别，或优先选择方案一。

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