SQLAlchemy下SQL表基于外键分组的条件自增ID实现咨询

解决方案 #

方案1：MS SQL Server 原生触发器实现（最高并发安全性） #

这是完全由数据库原生实现的方案，不存在自行查询最大值+1的竞态条件问题：

• 首先需要移除ValueID列的Identity自增属性，该字段的值将由触发器统一生成
• 创建INSTEAD OF INSERT触发器，逻辑如下：

CREATE TRIGGER trg_CalcGroupedValueID
ON Table2
INSTEAD OF INSERT
AS
BEGIN
    SET NOCOUNT ON;
    -- 为插入的每一条记录计算对应DefinitionID下的递增ValueID
    INSERT INTO Table2 (DefinitionID, ValueID, [其他需要插入的字段])
    SELECT 
        i.DefinitionID,
        ISNULL(t.max_vid, 0) + ROW_NUMBER() OVER (PARTITION BY i.DefinitionID ORDER BY (SELECT 1)),
        i.[其他字段1],
        i.[其他字段2]
    FROM inserted i
    OUTER APPLY (
        SELECT MAX(ValueID) as max_vid 
        FROM Table2 t 
        WHERE t.DefinitionID = i.DefinitionID
    ) t
END

该触发器是原子操作，即使高并发场景下也不会出现重复序号的问题。

方案2：兼顾SQLAlchemy数据库无关性的实现 #

如果你需要适配多类数据库，不想依赖SQL Server特定的触发器语法，可以使用SQLAlchemy的事件监听配合行级锁实现：

• 注册before_insert事件钩子，插入前查询对应分组的最大序号，查询时加排他锁避免并发冲突

from sqlalchemy import event, select, func

@event.listens_for(Table2, 'before_insert')
def generate_grouped_valueid(mapper, connection, target):
    # 加行级锁避免并发查询导致的重复序号
    max_vid = connection.scalar(
        select(func.max(Table2.ValueID))
        .where(Table2.DefinitionID == target.DefinitionID)
        .with_for_update()
    )
    target.ValueID = (max_vid or 0) + 1

注意该方案需要确保你的数据库支持SELECT ... FOR UPDATE语法，主流关系型数据库均已兼容该特性。

额外建议 #

如果业务允许，建议保留全局唯一的自增主键（比如原来的全局自增ValueID作为主键），额外新增GroupSeq字段存储分组内的递增序号，这样既可以满足业务需要的分组递增逻辑，也不会破坏主键的全局唯一性，后续跨数据库适配也更灵活。

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