使用SQLAlchemy批量上传

Question

我正在使用SQLAlchemy 1.1.0b将大量数据批量上传到PostgreSQL中，并且遇到了重复的关键错误。

from sqlalchemy import *
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.automap import automap_base

import pg

engine = create_engine("postgresql+pygresql://" + uname + ":" + passw + "@" + url)

# reflectively load the database.
metadata = MetaData()
metadata.reflect(bind=engine)
session = sessionmaker(autocommit=True, autoflush=True)
session.configure(bind=engine)
session = session()
base = automap_base(metadata=metadata)
base.prepare(engine, reflect=True)

table_name = "arbitrary_table_name" # this will always be arbitrary
mapped_table = getattr(base.classses, table_name)
# col and col2 exist in the table.
chunks = [[{"col":"val"},{"col2":"val2"}],[{"col":"val"},{"col2":"val3"}]]

for chunk in chunks:
    session.bulk_insert_mappings(mapped_table, chunk)
    session.commit()

运行它时，我得到以下信息：

sqlalchemy.exc.IntegrityError: (pg.IntegrityError) ERROR:  duplicate key value violates unique constraint <constraint>

我似乎也无法正确实例mapped_table化为Table()对象。

我正在使用时间序列数据，因此我正在批量获取时间范围有些重叠的数据。我想进行批量更新，以确保数据一致性。

对大数据集进行批量上插的最佳方法是什么？我知道PostgreSQL现在支持upserts，但是我不确定如何在SQLAlchemy中做到这一点。

Answer 1

来自 /sf/answers/1821325411/

找到这个命令后，我能够执行更新插入，但值得一提的是，对于批量“更新插入”来说，此操作很慢。

另一种方法是获取您想要更新插入的主键列表，并在数据库中查询任何匹配的 id：