SQLAlchemy 从 JSON 列表插入数据库

Question

我有一个这样的 JSON 列表：

print(type(listed)) # <class 'list'>
print (listed)

[
   {
       "email": "x@gmail.com",
       "fullname": "xg gf",
       "points": 5,
       "image_url", "https://imgur.com/random.pmg"
   },
   {
        ... similar json for the next user and so on
   }
]

我正在尝试将它们插入到我的 postgres 数据库中，该数据库具有这样的模型：

class Users(db.Model):
    __tablename__ = 'users'

    email = db.Column(db.String(), primary_key=True)
    displayName = db.Column(db.String())
    image = db.Column(db.String())
    points = db.Column(db.Integer())

但是我很困惑，我尝试了几种方法，但都没有奏效，任何人都可以通过示例指导我如何正确地做到这一点？

Answer 1

这是一个没有 pandas 的解决方案，使用 SQLAlchemy Core

创建引擎

engine = sqlalchemy.create_engine('...')

使用引擎作为绑定参数加载元数据

metadata = sqalchemy.Metadata(bind=engine)

引用表格

users_table = sqlalchemy.Table('users', metadata, autoload = True)

然后您可以开始插入

for user in json:
    query = users_table.insert()
    query.values(**user)

    my_session = Session(engine)
    my_session.execute(query)
    my_session.close()

这会在 json 中为每个用户创建一个会话，但我想你可能会喜欢它。它非常灵活，适用于任何桌子，您甚至不需要模型。只需确保 json 不包含 db 中不存在的任何列（这意味着您需要在 json 键和 db 列名中使用“img_url”（列名））

Answer 2

这是一个示例 json 列表，就像您提供的那样。

json = [
   {
       "email": "x@gmail.com",
       "fullname": "xg gf",
       "points": 5,
       "image_url": "https://imgur.com/random.pmg"
   },
    {
       "email": "onur@gmail.com",
       "fullname": "o g",
       "points": 7,
       "image_url": "https://imgur.com/random_x.pmg"
   }
]

现在创建一个空数据框all_df 并在您的 json 列表中运行迭代。 每次迭代都会使用列表中字典中的数据创建一个数据框，将其转置并附加到all_df。

import pandas as pd

all_df = pd.DataFrame()
for i in json:
    df = pd.DataFrame.from_dict(data=i, orient='index').T
    all_df = all_df.append(df)

输出：

现在您可以继续创建与数据库的会话并推送all_df

all_df.to_sql(con=your_session.bind, name='your_table_name', if_exists='your_preferred_method', index=False)

Answer 3

使用marshmallow-sqlalchemy

1. 验证传入的 JSON
2. 创建用于加载和转储数据的通用实用程序

定义架构

schema.py

from marshmallow import EXCLUDE
from marshmallow_sqlalchemy import ModelSchema
from app import db

class UserSchema(ModelSchema):
    class Meta(ModelSchema.Meta):
        model = Users
        sqla_session = db.session

user_schema_full = UserSchema(only=(
    'email',
    'displayName',
    'image',
    'points'
))

utils.py

下面的具体细节无关紧要，但可以创建从 JSON 到 ORM 对象和从 ORM 对象到 JSON 的通用实用程序。 schema_partial 用于自动生成的主键。

def loadData(data, schema_partial, many=False,
             schema_full=None, instance=None):
    try:
        if instance is not None:
            answer = schema_full.load(data, instance=instance, many=many)
        else:
            answer = schema_partial.load(data, many=many)
    except ValidationError as errors:
        raise InvalidData(errors, status_code=400)
    return answer


def loadUser(data, instance=None, many=False):
    return loadData(data=data,
                    schema_partial=user_schema_full,
                    many=many,
                    schema_full=user_schema_full,
                    instance=instance)

def dumpData(load_object, schema, many=False):
    try:
        answer = schema.dump(load_object, many=many)
    except ValidationError as errors:
        raise InvalidDump(errors, status_code=400)
    return answer


def dumpUser(load_object, many=False):
    return dumpData(load_object, schema=user_schema_full, many=many)

在 api 中使用 loadUser 和 dumpUser 生成干净的扁平代码。 api.py

@app.route('/users/', methods=['POST'])
def post_users():
    """Post many users"""
    users_data = request.get_json()
    users =  loadUser(users_data, many=True)
    for user in users:
        db.session.add(user)

    object_dump = dumpUser(users, many=True)

    db.session.commit()

    return jsonify(object_dump), 201