我在解析 ASCII 数据文件以将数据导入表时也遇到了这个问题。问题是我本能地和直觉地希望 SQLAlchemy 在允许唯一数据的同时跳过重复的行。或者,由于当前的 SQL 引擎,例如不允许使用 unicode 字符串,可能会引发随机错误。
但是,这种行为超出了 SQL 接口的定义范围。SQL API,因此 SQLAlchemy 只理解事务和提交,而不考虑这种选择性行为。此外,依赖自动提交功能听起来很危险,因为插入会在异常发生后停止,留下其余数据。
我的解决方案(我不确定它是否是最优雅的)是处理循环中的每一行,捕获并记录异常,并在最后提交更改。
假设您以某种方式获取列表列表中的数据,即作为列值列表的行列表。然后你在一个循环中读取每一行:
# Python 3.5
from sqlalchemy import Table, create_engine
import logging
# Create the engine
# Create the table
# Parse the data file and save data in `rows`
conn = engine.connect()
trans = conn.begin() # Disables autocommit
exceptions = {}
totalRows = 0
importedRows = 0
ins = table.insert()
for currentRowIdx, cols in enumerate(rows):
try:
conn.execute(ins.values(cols)) # try to insert the column values
importedRows += 1
except Exception as e:
exc_name = type(e).__name__ # save the exception name
if not exc_name in exceptions:
exceptions[exc_name] = []
exceptions[exc_name].append(currentRowIdx)
totalRows += 1
for key, val in exceptions.items():
logging.warning("%d out of %d lines were not imported due to %s."%(len(val), totalRows, key))
logging.info("%d rows were imported."%(importedRows))
trans.commit() # Commit at the very end
conn.close()
为了最大限度地提高此操作的速度,您应该禁用自动提交。我将此代码与 SQLite 一起使用,它仍然比我使用 only 的旧版本慢 3-5 倍sqlite3
,即使禁用了自动提交。(我移植到 SQLAlchemy 的原因是能够将它与 MySQL 一起使用。)
它不是最优雅的解决方案,因为它不如 SQLite 的直接接口快。如果我分析代码并在不久的将来找到瓶颈,我将用解决方案更新这个答案。