设置一个 sqlalchemy 模型和带有大嵌套的 Marshmallow JSON

Question

我正在使用 Flask 构建 API，但在为嵌套结构和 Marshmallow 构建模型时遇到多个错误。无论我做什么，似乎我都无法到达任何地方。

输入数据是嵌套的JSON。我需要将信息存储在数据库 (sqlite) 中并根据需要进行查询。我猜我创建的模型看起来还不错。我对模式有一些疑问。不确定它们是否正确或使用是否正确。关于使用诸如此类的嵌套结构的信息并不多。我看到了很多基本的东西，包括付费教程，但结构更复杂的东西并不多：

{
    "bessAssets": [{
        "designation": "bess3",
        "status": true,
        "vNom": 0,
        "eNom": 0,
        "maxSoc": 90,
        "minSoc": 10,       
        "testData": {
            "vNomD": [{
                "cRate": 0,
                "vAvg": 0,
                "trial": 1
            }],
            "vNomC": [{
                "cRate": 0,
                "vAvg": 0,
                "trial": 1
            }]
        }
    }]
}

我上次尝试做的工作是这样的：

from marshmallow import Schema, fields
from marshmallow_sqlalchemy import SQLAlchemySchema, auto_field, SQLAlchemyAutoSchema
from marshmallow import EXCLUDE
from flask_sqlalchemy import SQLAlchemy
from flask import Flask

import os
import json


app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///test.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)


class BessAsset(db.Model):

    __tablename__ = 'bess_asset'

    designation = db.Column(db.String(70), primary_key=True)
    status = db.Column(db.Boolean)
    vNom = db.Column(db.Float)
    eNom = db.Column(db.Float)
    maxSoc = db.Column(db.Float)
    minSoc = db.Column(db.Float)


class TestData(db.Model):

    __tablename__ = 'test_data'

    test_data_id = db.Column(db.Integer, primary_key=True)
    bess_asset_designation = db.Column(db.String, db.ForeignKey('bess_asset.designation'))
    bess_asset = db.relationship('BessAsset', backref=db.backref("testData"), cascade='all')


class VnomD(db.Model):

    __tablename__ = 'vnom_d'

    vnom_id = db.Column(db.Integer, primary_key=True)
    cRate = db.Column(db.Float)
    vAvg = db.Column(db.Float)
    trial = db.Column(db.Integer)
    test_data_id = db.Column(db.Integer, db.ForeignKey('test_data.test_data_id'))
    test_data = db.relationship('TestData', backref=db.backref("vNomD"), cascade='all')


class VnomC(db.Model):

    __tablename__ = 'vnom_c'

    vnom_id = db.Column(db.Integer, primary_key=True)
    cRate = db.Column(db.Float)
    vAvg = db.Column(db.Float)
    trial = db.Column(db.Integer)
    test_data_id = db.Column(db.Integer, db.ForeignKey('test_data.test_data_id'))
    test_data = db.relationship('TestData', backref=db.backref("vNomC"), cascade='all')


class VnomDSchema(SQLAlchemyAutoSchema):

    class Meta:
        model = VnomD
        exclude = ('vnom_id',)
        include_fk = True
        # load_instance = True


class VnomCSchema(SQLAlchemyAutoSchema):

    class Meta:
        model = VnomC
        exclude = ('vnom_id',)
        include_fk = True
        # load_instance = True


class TestDataSchema(SQLAlchemyAutoSchema):

    vnom_c = fields.Nested(VnomCSchema, many=True)
    vnom_d = fields.Nested(VnomDSchema, many=True)

    class Meta:
        model = TestData
        unknown = EXCLUDE
        include_relationships = True


class BessAssetSchema(SQLAlchemyAutoSchema):

    testData = fields.Nested(TestDataSchema, )

    class Meta:
        model = BessAsset
        unknown = EXCLUDE
        include_relationships = True
        # load_instance = True


db.create_all()

with open(os.path.join('examples', 'structures', 'assets.json')) as f:
    json_file = json.load(f)

bess_asset_schema = BessAssetSchema()
vnomd_schema = VnomDSchema()
vnomc_schema = VnomCSchema()

for b_asset in json_file['bessAssets']:

    bess_structure = (bess_asset_schema.load(b_asset))

    bess_asset = BessAsset(**bess_structure)
    test_data = TestData(bess_asset=bess_asset)

    db.session.add(bess_asset)

    for value in b_asset['testData']['vNomD']:
        vnom_d_structure = (vnomd_schema.load(value))
        vnom_d = VnomD(**vnom_d_structure, test_data=test_data)
        db.session.add(vnom_d)

    for value in b_asset['testData']['vNomC']:
        vnom_c_structure = (vnomc_schema.load(value))
        vnom_c = VnomC(**vnom_c_structure, test_data=test_data)
        db.session.add(vnom_c)

    db.session.commit()

bess_assets = BessAsset.query.all()
dump_data = bess_asset_schema.dump(bess_assets)

print(dump_data)

错误似乎来自这一行 (AttributeError: 'NoneType' object has no attribute 'query'):

for b_asset in json_file['bessAssets']:

    bess_structure = (bess_asset_schema.load(b_asset))

你能帮我弄清楚如何正确存储该结构并将其输出给用户吗？

谢谢！

Answer 1

好的，我明白了。不确定这个问题是否会引起其他可能的人的注意，但我相信它对某人有用。这是我的解决方案（SQLAlchemy + Marshmallow）：

from sqlalchemy.orm import relationship, sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, ForeignKey, String, Float, Boolean
from marshmallow_sqlalchemy import SQLAlchemyAutoSchema
from marshmallow import fields
import json

from sqlalchemy import create_engine

engine = create_engine('sqlite:///test.db', echo=True)
Session = sessionmaker(bind=engine)
session = Session()

Base = declarative_base()


# parent
class BessAsset(Base):

    __tablename__ = 'bess_asset'

    designation = Column(String, primary_key=True)
    status = Column(Boolean)
    vNom = Column(Float)
    eNom = Column(Float)
    maxSoc = Column(Float)
    minSoc = Column(Float)
    maxCCh = Column(Float)
    maxCDch = Column(Float)
    minPCh = Column(Float)
    minPDch = Column(Float)
    chEff = Column(Float)
    dischEff = Column(Float)
    eolCriterion = Column(Float)
    lifetime = Column(Integer)
    invPNom = Column(Float)
    invVNom = Column(Float)
    # one to one mapping
    testData = relationship('TestData', backref='bessAsset', lazy='select', uselist=False)


# add other testData elements relationships here
# one to one relationship
# one bessAsset will have one testdata
class TestData(Base):

    __tablename__ = 'test_data'

    test_data_id = Column(Integer, primary_key=True)
    designation = Column(String, ForeignKey('bess_asset.designation'))
    vNomD = relationship('Vnomd', backref='testData', lazy='select')
    vNomC = relationship('Vnomc', backref='testData', lazy='select')
    dLim = relationship('Dlim', backref='testData', lazy='select')
    cLim = relationship('Clim', backref='testData', lazy='select')
    effD = relationship('Effd', backref='testData', lazy='select')
    effC = relationship('Effc', backref='testData', lazy='select')
    roundEff = relationship('RoundEff', backref='testData', lazy='select')


class RoundEff(Base):

    __tablename__ = 'round_eff'
    roundeff_id = Column(Integer, primary_key=True)
    cRate = Column(Float)
    roundEffAvg = Column(Float)
    trial = Column(Integer)
    test_data_id = Column(Integer, ForeignKey('test_data.test_data_id'))


class Effc(Base):

    __tablename__ = 'eff_c'
    effc_id = Column(Integer, primary_key=True)
    cRate = Column(Float)
    effChAvg = Column(Float)
    trial = Column(Integer)
    test_data_id = Column(Integer, ForeignKey('test_data.test_data_id'))


class Effd(Base):

    __tablename__ = 'eff_d'
    effd_id = Column(Integer, primary_key=True)
    cRate = Column(Float)
    effDchAvg = Column(Float)
    trial = Column(Integer)
    test_data_id = Column(Integer, ForeignKey('test_data.test_data_id'))


class Clim(Base):

    __tablename__ = 'clim'

    clim_id = Column(Integer, primary_key=True)
    cRate = Column(Float)
    eRemain = Column(Float)
    trial = Column(Integer)
    test_data_id = Column(Integer, ForeignKey('test_data.test_data_id'))


class Dlim(Base):

    __tablename__ = 'dlim'

    dlim_id = Column(Integer, primary_key=True)
    cRate = Column(Float)
    eRemain = Column(Float)
    trial = Column(Integer)
    test_data_id = Column(Integer, ForeignKey('test_data.test_data_id'))


class Vnomd(Base):

    __tablename__ = 'vnomd'

    vnomd_id = Column(Integer, primary_key=True)
    cRate = Column(Float)
    vAvg = Column(Float)
    trial = Column(Integer)
    test_data_id = Column(Integer, ForeignKey('test_data.test_data_id'))


class Vnomc(Base):

    __tablename__ = 'vnomc'

    vnomc_id = Column(Integer, primary_key=True)
    cRate = Column(Float)
    vAvg = Column(Float)
    trial = Column(Integer)
    test_data_id = Column(Integer, ForeignKey('test_data.test_data_id'))


class VnomDSchema(SQLAlchemyAutoSchema):

    class Meta:
        model = Vnomd
        exclude = ['vnomd_id']


class VnomCSchema(SQLAlchemyAutoSchema):

    class Meta:
        model = Vnomc
        exclude = ['vnomc_id']


class RoundEffSchema(SQLAlchemyAutoSchema):

    class Meta:
        model = RoundEff
        exclude = ['roundeff_id']


class DlimSchema(SQLAlchemyAutoSchema):

    class Meta:
        model = Dlim
        exclude = ['dlim_id']


class ClimSchema(SQLAlchemyAutoSchema):

    class Meta:
        model = Clim
        exclude = ['clim_id']


# include test_data nested objects here
# will not be a list
class TestDataSchema(SQLAlchemyAutoSchema):

    vNomD = fields.Nested(VnomDSchema, many=True)
    vNomC = fields.Nested(VnomCSchema, many=True)
    dLim = fields.Nested(DlimSchema, many=True)
    cLim = fields.Nested(ClimSchema, many=True)
    roundEff = fields.Nested(RoundEffSchema, many=True)

    class Meta:
        model = TestData
        exclude = ['test_data_id', 'bessAsset']
        include_relationships = True


class BessAssetSchema(SQLAlchemyAutoSchema):

    testData = fields.Nested(TestDataSchema)

    class Meta:
        model = BessAsset
        ordered = True  # returns a ordered dict
        # include_fk = True
        # exclude = ['id', 'director_id']


Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)

vnomc = Vnomc(cRate=0.2, vAvg=0.34, trial=3)
vnomd = Vnomd(cRate=0.2, vAvg=0.34, trial=3)
roundEff = RoundEff(cRate=0.3, roundEffAvg=2.1, trial=2)
dlim = Dlim(cRate=2.5, eRemain=3.4, trial=100)
clim = Clim(cRate=2.5, eRemain=3.4, trial=200)

test_data = TestData(vNomC=[vnomc],
                     vNomD=[vnomd],
                     roundEff=[roundEff],
                     dLim=[dlim],
                     cLim=[clim])

bess_asset = BessAsset(designation='Teste 1',
                       status=True,
                       vNom=0.2,
                       eNom=0.3,
                       minPCh=2.1,
                       maxCCh=20.3,
                       minSoc=20.1,
                       maxSoc=1.3,
                       testData=test_data)

session.add(bess_asset)
session.commit()

bess_assets = session.query(BessAsset).all()
my_schema = BessAssetSchema()
my_data = my_schema.dump(bess_assets, many=True)
print(my_data)

设置一个 sqlalchemy 模型和带有大嵌套的 Marshmallow JSON

Set a sqlalchemy model and Marshmallow with large nested JSON

python

marshalling

flask

flask-sqlalchemy

marshmallow