如何用SQLAlchemy 2.0批量插入并维护ORM关联关系？

我正在开发一个网页爬虫项目，需使用SQLAlchemy ORM将爬取的数据存入数据库，以https://quotes.toscrape.com/作为示例。

模型定义（models.py） #

# models.py
from sqlalchemy import Column, Date, ForeignKey, Integer, String, Table, Text
from sqlalchemy.orm import declarative_base, relationship


Base = declarative_base()


class Author(Base):
    __tablename__ = 'author'

    id = Column(Integer, primary_key=True)
    name = Column(String(50), unique=True, nullable=False)
    birthday = Column(Date, nullable=False)
    bio = Column(Text, nullable=False)


class Tag(Base):
    __tablename__ = 'tag'

    id = Column(Integer, primary_key=True)
    name = Column(String(31), unique=True, nullable=False)


class Quote(Base):
    __tablename__ = 'quote'

    id = Column(Integer, primary_key=True)
    author_id = Column(ForeignKey('author.id'), nullable=False)
    quote = Column(Text, nullable=False, unique=True)

    author = relationship('Author')
    tags = relationship('Tag', secondary='quote_tag')


t_quote_tag = Table(
    'quote_tag', Base.metadata,
    Column('quote_id', ForeignKey('quote.id'), primary_key=True),
    Column('tag_id',   ForeignKey('tag.id'),   primary_key=True)
)

工作单元模式示例（unit_of_work.py） #

使用ORM工作单元范式时，只需将Quote实例添加到session并调用session.commit()，即可自动填充全部4张关联表：

# unit_of_work.py 

from datetime import datetime
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from models import *

engine = create_engine('sqlite:///quotes.db')

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

einstein_dict = {
    'name': 'Albert Einstein',
    'birthday': datetime(day=14, month=3, year=1879),
    'bio': 'Won the 1921 Nobel Prize in Physics.'
}

change_dict = {'name': 'change'}
deep_thoughts_dict = {'name': 'deep-thoughts'}
thinking_dict = {'name': 'thinking'}
world_dict = {'name': 'world'}

quote_dict = {
    'quote': (
        "The world as we have created it is a process of our thinking. "
        "It cannot be changed without changing our thinking."
    )
}

quote_instance = Quote(**quote_dict)
quote_instance.author = Author(**einstein_dict)
quote_instance.tags = [
    Tag(**change_dict), Tag(**deep_thoughts_dict), Tag(**thinking_dict), Tag(**world_dict),
]


# 自动维护所有关联关系
session.add(quote_instance)
session.commit()

现有批量插入实现（bulk_insert.py） #

根据SQLAlchemy ORM批量插入文档，可以脱离工作单元模式，使用字典引用和子查询进行插入/更新，但需分步操作：

# bulk_insert.py

# Insert Authors
session.execute(
    insert(Author),
    [
        einstein_dict
    ]
)

# Insert Tags
session.execute(
    insert(Tag),
    [
        change_dict,
        deep_thoughts_dict,
        thinking_dict,
        world_dict,
    ]
)

# Insert Quotes
session.execute(
    insert(Quote).values([
        {
            'author_id': select(Author.id).where(Author.name == quote_instance.author.name),
            'quote': quote_instance.quote
        }
    ])
)

# Insert quote_tag association_table
values = []
for tag in quote_instance.tags:
    values.append({
        'quote_id': select(Quote.id).where(Quote.quote == quote_instance.quote),
        'tag_id': select(Tag.id).where(Tag.name == tag.name)
    })
session.execute(
    insert(t_quote_tag).values(values)
)

session.commit()

理想实现诉求 #

我希望有更简便的方式，能在SQLAlchemy 2.0批量插入时自动维护模型间的关联关系，比如类似以下代码的实现：

# ideal.py

session.execute(
    insert(Quote).instances([quote_instance])
)
session.commit()

在SQLAlchemy 2.0中，确实存在更简洁的方式实现批量插入时自动维护关联关系，以下是两种实用方案：

方案1：使用bulk_save_objects批量保存ORM实例 #

bulk_save_objects支持批量处理ORM实例，能自动维护关联关系（包括多对多的中间表），同时比逐个调用session.add更高效。

示例代码：

# bulk_with_bulk_save_objects.py
from datetime import datetime
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from models import *

engine = create_engine('sqlite:///quotes.db')
Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)
session = Session()

# 构建关联实例
einstein = Author(
    name='Albert Einstein',
    birthday=datetime(1879, 3, 14),
    bio='Won the 1921 Nobel Prize in Physics.'
)
tags = [
    Tag(name='change'),
    Tag(name='deep-thoughts'),
    Tag(name='thinking'),
    Tag(name='world')
]
quote = Quote(
    quote="The world as we have created it is a process of our thinking. It cannot be changed without changing our thinking."
)
quote.author = einstein
quote.tags = tags

# 批量保存所有关联实例
session.bulk_save_objects([einstein] + tags + [quote])
session.commit()

注意：如果关联对象（如Author、Tag）可能已存在，需先检查并避免重复插入，否则会触发唯一性约束错误。

方案2：结合get_or_create处理重复数据，批量插入 #

针对爬虫场景中可能出现的重复数据（同一作者/标签多次爬取），可以封装get_or_create方法先获取或创建关联对象，再批量插入Quote，既自动维护关联，又避免重复。

示例代码：

# bulk_with_get_or_create.py
from datetime import datetime
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from models import *

engine = create_engine('sqlite:///quotes.db')
Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)
session = Session()

def get_or_create_author(name, birthday, bio):
    """获取或创建作者实例，避免重复"""
    author = session.query(Author).filter_by(name=name).first()
    if not author:
        author = Author(name=name, birthday=birthday, bio=bio)
        session.add(author)
    return author

def get_or_create_tag(name):
    """获取或创建标签实例，避免重复"""
    tag = session.query(Tag).filter_by(name=name).first()
    if not tag:
        tag = Tag(name=name)
        session.add(tag)
    return tag

# 准备批量爬取的数据
quotes_batch = [
    {
        'quote_text': "The world as we have created it is a process of our thinking. It cannot be changed without changing our thinking.",
        'author': {
            'name': 'Albert Einstein',
            'birthday': datetime(1879, 3, 14),
            'bio': 'Won the 1921 Nobel Prize in Physics.'
        },
        'tags': ['change', 'deep-thoughts', 'thinking', 'world']
    },
    # 可添加更多爬取的quote数据
]

# 处理批量数据，构建Quote实例
quote_instances = []
for q_item in quotes_batch:
    author = get_or_create_author(**q_item['author'])
    tags = [get_or_create_tag(tag_name) for tag_name in q_item['tags']]
    quote = Quote(quote=q_item['quote_text'], author=author)
    quote.tags = tags
    quote_instances.append(quote)

# 批量插入Quote
session.bulk_save_objects(quote_instances)
session.commit()

关键注意点 #

• 唯一性约束：确保Author.name和Tag.name的唯一约束生效，这是避免重复数据的基础；
• 性能平衡：bulk_save_objects比ORM工作单元的逐个add高效，但纯SQL批量插入（如session.execute(insert(...))）性能更高，不过后者需要手动处理关联；
• 事务控制：批量操作建议放在事务中，避免部分插入失败导致数据不一致；
• 多对多关联：使用bulk_save_objects时，只要关联的两端实例已被添加到session或存在于数据库，中间关联表会自动插入数据。

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