如何加速SQLAlchemy中带多关系数据的session.add操作

这种嵌套循环+频繁调用session.flush()的写法确实是批量插入的性能杀手——每一次flush都会触发和数据库的网络往返，加上PostgreSQL处理Identity列的ID生成开销，100条用户数据耗时8秒太正常了。下面给你几个针对性的优化方案，从ORM用法到批量操作，一步步把速度提上来：

1. 彻底减少session.flush()的调用次数 #

你当前的代码每插入一个对象就调用一次flush()，这完全没必要。flush()的核心作用是将内存中的对象同步到数据库并获取自动生成的ID（比如你的Identity主键），但我们可以批量处理完一层数据后再统一flush，而不是每个对象都单独触发。

比如，原来的代码可以改成：

for u in user_list:
    new_user = User(user_name=u.get('user_name'), email=u.get('email'))
    session.add(new_user)
    
    items = []
    for item in u.get('items'):
        new_item = Item(name=item.get('name'), note=item.get('note'))
        session.add(new_item)
        items.append((new_user, new_item))
        
        props = []
        for prop in item.get('properties'):
            new_prop = Properties(name=prop.get('name'), value=prop.get('value'))
            session.add(new_prop)
            props.append((new_item, new_prop))
    
    # 一次性flush所有新增对象，获取所有ID
    session.flush()
    
    # 现在可以用获取到的ID创建关联表记录
    for user, item in items:
        session.add(UserItemLink(user_id=user.id, item_id=item.id))
    for item, prop in props:
        session.add(ItemPropLink(item_id=item.id, prop_id=prop.id))

# 最后统一commit
session.commit()

这样flush次数从数百次降到了100次（每个用户一次），性能会有明显提升。

2. 利用ORM多对多关系自动管理关联表 #

你的模型已经定义了关联表，但还在手动创建UserItemLink和ItemPropLink对象——这不仅麻烦，还增加了额外的数据库操作。我们可以直接通过ORM的relationship定义多对多关系，让SQLAlchemy自动处理中间表。

修改模型定义： #

class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, Identity(always=True, start=1, increment=1, minvalue=1, maxvalue=2147483647, cycle=False, cache=1), primary_key=True)
    name = Column(String(20))
    email = Column(String(50))
    # 定义与Item的多对多关系，关联到中间表user_item_link
    items = relationship('Item', secondary='user_item_link', back_populates='users')

class Item(Base):
    __tablename__ = 'item'
    id = Column(Integer, Identity(always=True, start=1, increment=1, minvalue=1, maxvalue=2147483647, cycle=False, cache=1), primary_key=True)
    name = Column(String(50))
    note = Column(String(50))
    users = relationship('User', secondary='user_item_link', back_populates='items')
    # 定义与Properties的多对多关系，关联到中间表item_prop_link
    properties = relationship('Properties', secondary='item_prop_link', back_populates='items')

class Properties(Base):
    __tablename__ = 'properties'
    id = Column(Integer, Identity(always=True, start=1, increment=1, minvalue=1, maxvalue=2147483647, cycle=False, cache=1), primary_key=True)
    name = Column(String(50))
    value = Column(String(50))
    items = relationship('Item', secondary='item_prop_link', back_populates='properties')

# 关联表模型保持不变，无需修改
class UserItemLink(Base):
    __tablename__ = 'user_item_link'
    id = Column(Integer, Identity(always=True, start=1, increment=1, minvalue=1, maxvalue=2147483647, cycle=False, cache=1), primary_key=True)
    user_id = Column(ForeignKey('user.id'), nullable=False)
    item_id = Column(ForeignKey('item.id'), nullable=False)

class ItemPropLink(Base):
    __tablename__ = 'item_prop_link'
    id = Column(Integer, Identity(always=True, start=1, increment=1, minvalue=1, maxvalue=2147483647, cycle=False, cache=1), primary_key=True)
    item_id = Column(ForeignKey('item.id'), nullable=False)
    prop_id = Column(ForeignKey('properties.id'), nullable=False)

优化后的入库逻辑： #

# 先创建所有对象并建立ORM关系
users = []
for u_data in user_list:
    user = User(name=u_data['user_name'], email=u_data['email'])
    items = []
    for item_data in u_data['items']:
        item = Item(name=item_data['name'], note=item_data.get('note', ''))
        # 直接给item.properties赋值属性对象
        item.properties = [
            Properties(name=p['name'], value=p['value'])
            for p in item_data['properties']
        ]
        items.append(item)
    # 直接给user.items赋值物品对象
    user.items = items
    users.append(user)

# 批量添加所有用户（连带物品和属性，ORM会自动处理关联表）
session.add_all(users)
# 只需要一次flush（如果后续不需要ID的话，甚至可以直接commit）
session.commit()

这种写法不仅代码简洁，SQLAlchemy还会优化插入顺序和SQL语句，大幅减少数据库交互次数。

3. 使用批量插入API进一步提速 #

如果数据量极大（比如上万条用户），可以使用SQLAlchemy的批量插入API：bulk_insert_mappings或bulk_save_objects，它们会生成单条批量INSERT语句，比逐个add快得多。

比如用bulk_insert_mappings处理用户数据（需要先整理成字典列表）：

# 整理用户数据字典
user_dicts = [
    {'name': u['user_name'], 'email': u['email']}
    for u in user_list
]
# 批量插入用户，return_defaults=True会返回自动生成的ID
session.bulk_insert_mappings(User, user_dicts, return_defaults=True)

# 然后用返回的ID处理物品和关联表，逻辑类似...

注意：bulk_insert_mappings是更底层的操作，不会触发ORM的事件和关系自动处理，所以适合纯批量插入场景；如果需要维护ORM关系，还是推荐用add_all+自动关联的方式。

4. 会话配置优化 #

在创建session时，可以关闭自动flush，避免不必要的数据库交互：

from sqlalchemy.orm import sessionmaker

Session = sessionmaker(bind=engine, autoflush=False)
session = Session()

这样只有当你手动调用flush()或commit()时，才会同步数据到数据库，减少额外的网络开销。

总结 #

按优先级排序，最有效的优化是：

• 移除不必要的flush()调用，批量处理后统一同步
• 利用ORM多对多关系自动管理中间表，减少手动操作
• 尝试批量插入API处理超大数据集

按这些方案优化后，100条用户数据的插入时间应该能降到几百毫秒以内。

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