当前位置：首页 > article >正文

SQLAlchemy系列教程：批量插入数据

article 2025/3/16 3:00:51

高效地批量插入数据对于应用程序的性能至关重要。SQLAlchemy为批处理操作提供了几种机制，可以最大限度地减少开销并加快数据库事务时间。在本指南中，我们将探讨如何使用SQLAlchemy执行批量插入，包括从基础技术到高级技术。

在这里插入图片描述

搭建环境

在开始之前，请确保已经安装了Python和SQLAlchemy。如果您还没有安装SQLAlchemy，请使用pip Install SQLAlchemy。你还需要与数据库进行交互；在本指南中，我们将使用SQLite，因为它不需要额外的设置。

from sqlalchemy import create_engine, Column, Integer, String, MetaData, Table
from sqlalchemy.orm import sessionmaker

# Create an engine that stores data in the local directory's
# sqlalchemy_example.db file.
engine = create_engine('sqlite:///sqlalchemy_example.db')

# Define a table
metadata = MetaData()
users_table = Table('users', metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String),
    Column('age', Integer))

# Create the table
metadata.create_all(engine)

# Create a Session class
Session = sessionmaker(bind=engine)

基本批量插入

要执行基本的批量插入，可以使用Session.bulk_insert_mappings()方法。此方法允许您一次插入许多对象，而无需实例化模型实例。

session = Session()

users = [
    {'name': 'John Doe', 'age': 28},
    {'name': 'Jane Doe', 'age': 25},
    # ... more user dicts
]

# Bulk insert mappings
session.bulk_insert_mappings(users_table, users)
session.commit()

使用插入构造函数处理大型批处理

对于更大的数据集，通过 SQLAlchemy 核心提供的 SQL 插入构造函数可能更高效。以下是使用方法：

insert_stmt = users_table.insert().values(
    [{'name': 'Alice', 'age': 30}, {'name': 'Bob', 'age': 22}]  # list of dicts
)

conn = engine.connect()
conn.execute(insert_stmt)
conn.close()

在批量插入期间处理重复项

通过使用数据库机制（如PostgreSQL的ON CONFLICT）来处理批量插入操作中的重复项。下面是一个使用SQLAlchemy核心的例子：

insert_stmt = users_table.insert().values(
    [{'name': 'Charlie', 'age': 35}, {'name': 'Dana', 'age': 19}]
).on_conflict_do_update(
    index_elements=['name'],
    set_=dict(age=users_table.c.age + 1)
)

conn = engine.connect()
conn.execute(insert_stmt)
conn.close()

性能考虑

在执行批量操作时，可以遵循以下最佳实践以获得更好的性能：

在批量操作期间禁用ORM的自动刷新，以防止不必要的刷新。
只有在所有数据插入后才使用事务并提交。
了解数据库的批量插入功能和限制。

高级批处理操作

SQLAlchemy version 1.3引入了bulk_save_objects()函数，用于可能涉及主键和外键处理的复杂批量操作。这里有一个简单的例子：

session = Session()

users = [
    {'name': 'Eva', 'age': 28},
    # ... more user dicts
]

# Handle bulk save objects
session.bulk_save_objects([
    users_table(**data) for data in users
])
session.commit()

使用SQLAlchemy ORM进行批量插入

如果需要ORM的特性来进行批量插入，例如，触发事件或处理关系，你可以使用session.add_all()方法：

session = Session()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    age = Column(Integer)

# Define some user instances
users = [User(name='Frank', age=45), User(name='Grace', age=30)]

# Use add_all for bulk insertion
session.add_all(users)
session.commit()