python-3.x - 用 SQLAlchemy 一次性替换大量数据？

Question

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

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

这是我的泡菜/数据库代码：

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()

（在阅读保存和加载pickle文件中的多个对象后制作？）

标签： python-3.xsqlalchemypickle

Answer 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()