用 SQLAlchemy 一次性替换大量数据？答案

【问题标题】：Replace a large amount of data at once with SQLAlchemy?用 SQLAlchemy 一次性替换大量数据？
【发布时间】：2021-05-20 11:32:48
【问题描述】：

我每天需要更新大量数据（大意味着>3MB）。我想将它存储为 JSON，但据我所知，SQLAlchemy 不支持从 JSON 转换。所以现在我正在尝试用 Pickle 来做这件事。目前，我将我拥有的每个产品都存储在一个巨大的 Pickle 文件中，以便稍后将其重新加载并提交。但是，我不断收到错误消息，说我的产品类未映射，我不确定这意味着什么或如何修复它。我在谷歌搜索时遇到的所有内容都与我的代码一点也不相似。这是我的产品类别：

class Product:
    id = ""
    name = ""
    store_name = ""
    brand = ""
    price = ""
    amount = ""
    info = ""
    image = ""

这是我的 Pickle / 数据库代码：

def loadall():
    with open('products.txt', mode='rb') as products_txt:
        while True:
            try:
                yield pickle.load(products_txt)
            except EOFError:
                break

Session = sessionmaker(bind=db)
    session = Session()
    products = loadall()

    with db.connect() as conn:
        session.add_all(products)
        session.commit()

（阅读Saving and loading multiple objects in pickle file?后制作）

【问题讨论】：

我明白这不是你的问题，但是SqlAlchemy 显然支持JSON as a data type 现在，sqlalchemy 的目标不是从JSON 或任何其他数据格式转换；但这可以使用 python 标准库json module 轻松完成，我更喜欢使用它而不是 pickle。
如果它可以帮助我解决我的问题，我很乐意尝试一下！我一直在寻找和摆弄您链接的 JSON 实现，但我无法让它工作。他们使用 table.insert()，但我在本地没有表，我只是从我的桌面连接到它。你可能知道我该如何解决这个问题吗？
如果您使用 ORM，您可以通过访问Product.__table__ 来获取表格。但你不需要。但是如果你想在数据库上使用 JSON 列，你的表的设计会有所不同：你将拥有 JSON 列，它将一些属性存储为 JSON。我不建议你对像你这样定义明确的属性这样做。我只想指出 sqlalchemy 可以处理 JSON。

标签： python-3.x sqlalchemy pickle

【解决方案1】：

下面应该会给你一个想法（我将测试数据限制为 2 列）：

test.py :

#!/usr/bin/env python3

import json

from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import Session, as_declarative, registry

## configuration
engine = create_engine("sqlite+pysqlite:///:memory:", echo=True, future=True)
mapper_registry = registry()


@as_declarative()
class Base(object):
    pass


class Product(Base):
    __tablename__ = "product"
    id = Column("id", Integer, primary_key=True)
    name = Column(String)
    info = Column(String)


def _main():
    with Session(engine) as session:
        Base.metadata.drop_all(engine)
        Base.metadata.create_all(engine)

        ## test data
        objects = [
            Product(name="no-info"),
            Product(name="red-color", info="large RED"),
        ]

        session.add_all(objects)
        session.commit()
        session.expunge_all()

        ## test data: from JSON
        # product_list = load_all_test()  # load from test code
        product_list = load_all_file()  # load from the file
        print(product_list)

        # create Product() instances from JSON
        products = [Product(**kw) for kw in product_list]
        session.add_all(products)
        session.commit()


def load_all_test():
    test_json_content = """
[
    { "name": "json-product-1", "info": "some info from json-1" },
    { "name": "json-product-2", "info": "some info from json-2" }
]
        """

    product_list = json.loads(test_json_content)
    return product_list


def load_all_file():
    with open("myjsonfile.json") as fh:
        product_list = json.load(fh)
    return product_list


_main()

【讨论】：

我最初是通过使用文本文件来存储整个查询来解决这个问题的，但是这更加优雅。谢谢！