FastAPI+Python3多表插入事务提交/回滚：需关联用户主键作为外键

核心问题是你没必要先提交User事务来获取主键ID——SQLAlchemy在同一事务会话内，即使未提交，添加的User对象的自增ID也能直接访问（会话会自动跟踪并生成ID，无需手动commit或refresh）。以下是正确的原子性实现方式：

1. 使用ORM方式实现原子操作 #

直接将两个插入操作放在同一个事务中，通过异常捕获处理回滚：

try:
    # 所有操作在同一个会话事务内执行
    user_obj = MyUser(user_name='abc@myemail.com', password='abcd')
    db.session.add(user_obj)
    
    # 无需提交或刷新，直接访问user_obj.id获取自增主键
    user_enrolment = UserEnrollmentModel(user_id=user_obj.id, role_id=123)
    db.session.add(user_enrolment)
    
    # 统一提交事务
    db.session.commit()
except Exception as e:
    # 异常触发时整体回滚所有操作
    db.session.rollback()
    raise e  # 可根据业务需求自定义异常处理逻辑

原理说明 #

对于autoincrement=True的自增主键，SQLAlchemy在调用add()后，会通过会话内部维护对象状态，无需提交事务就能获取生成的ID——底层会自动执行INSERT ... RETURNING（或对应数据库方言的自增ID获取逻辑），但所有操作仍处于同一事务上下文，确保原子性。

2. 修正模型关系（可选） #

你的User模型中多对多关系的secondary参数可能存在定义问题，建议按规范调整：

# 先定义多对多关联表
user_enrolment = sa.Table(
    'user_enrolment',
    Model.metadata,
    sa.Column('user_id', sa.Integer(), sa.ForeignKey('user.id'), primary_key=True),
    sa.Column('enrolment_id', sa.Integer(), sa.ForeignKey('enrolment.id'), primary_key=True)
)

class User(Model):
    __tablename__ = "user"
    id = sa.Column(sa.Integer(), primary_key=True, autoincrement=True)
    user_name = sa.Column(sa.String(), nullable=False)
    password = sa.Column(sa.String(), nullable=False)
    # 修正secondary参数，指向关联表对象
    enrolments = relationship("Enrolment", secondary=user_enrolment, backref="users")

class Enrolment(Model):
    __tablename__ = "enrolment"
    id = sa.Column(sa.Integer(), primary_key=True, autoincrement=True)
    role_id = sa.Column(sa.Integer())
    # 其他业务字段...

如果是一对多关系（一个用户对应多个 enrollment），则无需关联表，直接在Enrollment模型中定义外键：

class UserEnrollmentModel(Model):
    __tablename__ = "user_enrolment"
    id = sa.Column(sa.Integer(), primary_key=True, autoincrement=True)
    user_id = sa.Column(sa.Integer(), sa.ForeignKey('user.id'), nullable=False)
    role_id = sa.Column(sa.Integer(), nullable=False)
    # 建立关系映射
    user = relationship("User", backref="enrolments")

3. 使用SQLAlchemy Core方式实现原子操作 #

若使用Core的execute方法插入，同样可以在同一事务内完成：

try:
    with db.session.begin():  # 上下文管理器自动管理事务，异常时自动回滚
        # 插入User并返回ID
        user_result = db.session.execute(
            sa.insert(User).values(user_name='abc@myemail.com', password='abcd').returning(User.id)
        )
        user_id = user_result.scalar_one()
        
        # 插入Enrollment
        db.session.execute(
            sa.insert(UserEnrollmentModel).values(user_id=user_id, role_id=123)
        )
        # 无需手动commit，上下文块结束自动提交
except Exception as e:
    raise e  # 异常时自动触发回滚

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