sqlalchemy 中的 Postgresql ON CONFLICT

Question

我已经阅读了很多资源（ao.1、2），但我无法让 Postgresql 的 ON CONFLICT IGNORE 行为在 sqlalchemy 中起作用。

我使用this accepted answer 作为基础，但它给出了

SAWarning: Can't validate argument 'append_string'; can't locate any SQLAlchemy dialect named 'append'

我尝试将 postgresql 方言添加到 @compile 子句，重命名我的对象，但它不起作用。 我也尝试使用str(insert())+ " ON CONFILCT IGNORE" 没有结果。 （顺便说一句，这并不奇怪）

如何将On CONFLICT IGNORE 添加到我的插入中？我喜欢建议的解决方案，因为我可以看到自己不希望每个 INSERT 上的 IGNORE 行为

ps。使用 python 2.7（不介意升级到 3.4/3.5），最新的 sqlalchemy (1.x)

Answer 1

你不需要这个，使用存在条件来防止插入重复。

例如：

INSERT INTO table (unique_name) 
SELECT 'some_string'
WHERE NOT EXISTS(SELECT 1 FROM table WHERE unique_name = 'some_string')

你也可以

INSERT INTO table (unique_name)
VALUES('some_string')
ON CONFLICT (unique_name) DO NOTHING

但如果您需要insert 或在单个查询中更新，那么这是您的示例：

INSERT INTO distributors (did, dname)
VALUES (5, 'Gizmo Transglobal'), (6, 'Associated Computing, Inc')
ON CONFLICT (did) DO UPDATE SET dname = EXCLUDED.dname;

这是来自 PostgreSQL 文档的示例。

Answer 2

这适用于 Postgresql 9.5：

from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql.expression import Insert

@compiles(Insert)
def prefix_inserts(insert, compiler, **kw):
    return compiler.visit_insert(insert, **kw) + " ON CONFLICT DO NOTHING"

我将它用于bulk_insert_mappings。但是，它不会使 ON CONFLICT DO NOTHING 成为可选

Answer 3

使用 Postgres 9.6.1、sqlachemy 1.1.4 和 psycopg2 2.6.2：

1. 将您的数据结构转换为字典。来自 Pandas 的

   import pandas
   from sqlalchemy import MetaData
   from sqlalchemy.dialects.postgresql import insert
   import psycopg2
   
   # The dictionary should include all the values including index values
   insrt_vals = df.to_dict(orient='records')
   

2. 通过 sqlalchemy 连接到数据库。而是尝试下面的 psycog2 驱动程序和本机 COPY 函数，它绕过所有 postgres 索引。

   csv_data = os.path.realpath('test.csv')
   con = psycopg2.connect(database = 'db01', user = 'postgres')
   cur = con.cursor()
   cur.execute("\copy stamm_data from '%s' DELIMITER ';' csv header" % csv_data)
   con.commit()
   

3. 执行

   results = engine.execute(do_nothing_stmt)
   # Get number of rows inserted
   rowcount = results.rowcount
   

警告：

此方法不适用于开箱即用的NaTs。

一切都在一起

tst_df = pd.DataFrame({'colA':['a','b','c','a','z', 'q'],
              'colB': pd.date_range(end=datetime.datetime.now() , periods=6),
              'colC' : ['a1','b2','c3','a4','z5', 'q6']})


insrt_vals = tst_df.to_dict(orient='records')
engine =      sqlalchemy.create_engine("postgresql://user:password@localhost/postgres")
connect = engine.connect()
meta = MetaData(bind=engine)
meta.reflect(bind=engine)
table = meta.tables['tstbl']
insrt_stmnt = insert(table).values(insrt_vals)

do_nothing_stmt  = insrt_stmnt.on_conflict_do_nothing(index_elements=['colA','colB'])
results = engine.execute(do_nothing_stmt)

代替第 2 步和第 3 步，在 postgres 中使用带有复制命令的 psycog2 驱动程序对于较大的文件（接近演出）更快，因为它会关闭所有表索引。

csv_data = os.path.realpath('test.csv')

Answer 4

这适用于 Postgresql 10.5 和 Sqlalchemy 1.3.6：

from sqlalchemy.dialects.postgresql import insert


table_info = {
'tableTime': '',
'deploymentID': '',
'tableData': ''
}
insert_table = insert(Table).values(table_info)
insert_table_sql = insert_table.on_conflict_do_nothing(
  index_elements=['tableTime', 'deploymentID']
)
db.session.execute(insert_table_sql)
db.session.commit()

Answer 5

这是 Niklas 答案的延伸。

基本上，使用线程本地状态和上下文管理器来附加ON CONFLICT DO NOTHING 可选。不过，这仍然是一个大黑客。

它也只挂钩 postgres 特定的语句，而不是从文本手动构造 sql 查询。

import threading
from contextlib import contextmanager

from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql.expression import Insert
from sqlalchemy.dialects.postgresql.dml import OnConflictDoNothing

state = threading.local()

@contextmanager
def on_conflict_do_nothing():
  state.active = True
  yield
  del state.active

@compiles(Insert, 'postgresql')
def prefix_inserts(insert, compiler, **kw):
  if getattr(state, "active", False):
    insert._post_values_clause = OnConflictDoNothing()
  return compiler.visit_insert(insert, **kw)