如何使用 postgres\sqlalchemy 以最快的方式一次提交多条记录？

Question

我有 1,000,000 条记录要输入到数据库中，遗憾的是其中一些记录不支持 db 模式。在记录失败的那一刻我正在做：

1. 回滚到数据库
2. 观察异常
3. 解决问题
4. 再次运行。 我希望构建一个脚本来保存所有“坏”记录，但会提交所有正确的记录。

当然，我可以一个一个地提交，然后当提交失败时回滚并提交下一个，但我会支付“运行时价格”，因为代码会运行很长时间。

在下面的示例中，我有两个模型：文件和客户端。一个客户端有很多文件的关系。 在 commit.py 文件中，我希望一次或分批（1k）提交 1M 个文件对象。目前我只知道当我最后提交失败时，有没有办法知道之前哪个对象是“坏的”（以外键为例的完整性错误），即停放一边（在另一个列表中）但是承诺所有的“好”

非常感谢您的帮助

#model.py

from sqlalchemy import Column, DateTime, String, func, Integer, ForeignKey

from . import base
class Client(base):
    __tablename__ = 'clients'
    id = Column(String, primary_key=True)

class File(base):
    __tablename__ = 'files'
    id = Column(Integer, primary_key=True, autoincrement=True)
    client_id = Column(String, ForeignKey('clients.id'))
    path = Column(String)


#init.py
import os
from dotenv import load_dotenv
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base

load_dotenv()
db_name = os.getenv("DB_NAME")
db_host = os.getenv("DB_HOST")
db_port = os.getenv("DB_PORT")
db_user = os.getenv("DB_USER")
db_password = os.getenv("DB_PASSWORD")

db_uri = 'postgresql://' + db_user + ':' + db_password + '@' + db_host + ':' + db_port + '/' + db_name
print(f"product_domain: {db_uri}")


base = declarative_base()
engine = create_engine(db_uri)
base.metadata.bind = engine
Session = sessionmaker(bind=engine)
session = Session()
conn = engine.connect()

#commit.py
from . import session

def commit(list_of_1m_File_objects_records):
    #I wish to for loop over the rows and if a specific row rasie excaption to insert it to a list and handle after wards
    for file in list_of_1m_File_objects_records:
        session.add(file)
    session.commit()
    
# client:
# id
# "a"
# "b"
# "c"
# file:
# id|client_id|path
# --|---------|-------------
# 1   "a"      "path1.txt"
# 2   "aa"     "path2.txt"
# 3   "c"      "path143.txt"
# 4   "a"      "pat.txt"
# 5   "b"      "patb.txt"
# wish the file data would enter the database although it has one record "aa" which will raise integrity error

Answer 1

由于我无法发表评论，我建议使用 psycopg2 和 sqlAlchemy 生成与数据库的连接，然后在查询末尾使用带有“发生冲突”的查询来添加和提交您的数据

Answer 2

当然，我可以一个一个地提交，然后当提交失败时回滚并提交下一个，但我会支付“运行时价格”，因为代码会运行很长时间。

这个价格的来源是什么？如果是 fsync 速度，您可以通过在本地连接上将 synchronous_commit 设置为 off 来消除大部分成本。如果您在中途发生了崩溃，那么您需要确定在它恢复后实际记录了哪些内容，这样您就知道从哪里重新开始，但我认为这并不难做到。这种方法应该让您以最少的工作获得最大的收益。

目前我只知道当我最后提交时失败了

听起来您正在使用延迟约束。有什么原因吗？

有没有办法知道哪个对象是“坏的”（以外键为例的完整性错误）

对于该示例，请在开始之前将所有客户端 ID 读入字典（假设它们适合 RAM），然后在 python 端测试文件，以便在尝试插入它们之前拒绝孤儿。