如何在Python+SQLAlchemy+SQLite环境下结合ON CONFLICT DO NOTHING实现批量插入？

看起来你是要把批量金融时序数据导入SQLite数据库，同时要避免重复插入相同时间戳的记录对吧？我来给你一步步讲怎么用SQLAlchemy实现这个需求，结合ON CONFLICT DO NOTHING来处理冲突：

第一步：完善你的数据库模型 #

首先先把你没写完的模型补全，要实现冲突处理，我们需要把Timestamp设为主键（或者唯一索引），这样SQLite才能识别重复冲突：

import datetime
from typing import List
from sqlalchemy import String, Integer, Float
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class PriceData(Base):
    __tablename__ = "price_data"
    
    # 把Timestamp设为主键，确保唯一性
    timestamp: Mapped[int] = mapped_column(Integer, primary_key=True)
    open: Mapped[float] = mapped_column(Float)
    high: Mapped[float] = mapped_column(Float)
    low: Mapped[float] = mapped_column(Float)
    close: Mapped[float] = mapped_column(Float)

然后别忘了先创建数据库表：

from sqlalchemy import create_engine

engine = create_engine("sqlite:///financial_market.db")
# 创建所有表
Base.metadata.create_all(engine)

第二步：读取CSV数据 #

假设你用pandas来处理CSV（金融数据处理的常用工具），需要把原本作为索引的Timestamp转成普通列，方便后续插入：

import pandas as pd

# 读取CSV，指定Timestamp为索引
df = pd.read_csv("your_financial_data.csv", index_col="Timestamp")
# 把索引转为普通列
df.reset_index(inplace=True)
# 转成字典列表，方便SQLAlchemy处理
data_records = df.to_dict("records")

第三步：批量插入+冲突忽略（核心步骤） #

这里推荐用SQLAlchemy Core层的插入语句，因为它更高效，而且原生支持ON CONFLICT DO NOTHING，非常适合大数据量的批量导入：

from sqlalchemy import insert

with engine.connect() as conn:
    # 构建插入语句，指定当timestamp主键冲突时，不做任何操作
    insert_stmt = insert(PriceData).values(data_records).on_conflict_do_nothing(
        index_elements=["timestamp"]  # 指定用来判断冲突的索引列（这里是主键）
    )
    # 执行插入
    conn.execute(insert_stmt)
    # 提交事务
    conn.commit()

补充：如果想用ORM会话实现 #

如果你更习惯用ORM的Session，也可以用同样的插入语句，只是通过会话执行：

from sqlalchemy.orm import Session

with Session(engine) as session:
    insert_stmt = insert(PriceData).values(data_records).on_conflict_do_nothing(index_elements=["timestamp"])
    session.execute(insert_stmt)
    session.commit()

关键说明 #

• on_conflict_do_nothing会告诉SQLite：当插入的记录和已有记录在指定的索引列（这里是timestamp主键）上重复时，直接跳过这条记录，不会抛出错误，也不会修改已有数据。
• 这种方法比先查询所有已有时间戳再过滤数据要高效得多，尤其是数据量很大的时候，因为所有逻辑都在数据库层面完成，减少了Python和数据库之间的交互。

备注：内容来源于stack exchange，提问作者kamran nasirizad