如何在 Python 和 JSON 之间序列化和反序列化复杂的 POCO 数据结构

Question

我们已经研究了几个小时了，没有运气，有很多方法可以在 Python 中序列化和反序列化对象，但我们需要一个简单而标准的尊重类型的方法，例如：

from typings import List, NamedTuple

class Address(object):
    city:str
    postcode:str

class Person(NamedTuple):
    name:str
    addresses:List[Address]

我的问题非常简单，我正在寻找一种标准的方式来转换为 JSON，而无需为每个类编写序列化/反序列化代码，例如：

json = '{ "name": "John", "addresses": [{ "postcode": "EC2 2FA", "city": "London" }, { "city": "Paris", "postcode": "545887", "extra_attribute": "" }]}'

我需要一种方法：

p= magic(json, Person) # or something similar
print(type(p)) # should print Person
for a in p.addresses:
    print(type(a)) # prints Address
    print(a.city) # should print London then Paris
json2 = unmagic(p)
print(json2 == json) # prints true (probably there will be difference in spacing, but just to clarify the idea)

我从事编程工作 15 年，并且已经使用 Python 一年了，即使经过广泛研究，仍然不确定非常简单地序列化/反序列化 POCO 对象结构的最佳方法是什么，我觉得很愚蠢。

编辑

迄今为止探索的选项具有以下一项或多项限制：

• 取决于 JSON / 类定义中元素的顺序，而不是属性的名称（前面的示例会失败，因为城市和邮政编码混淆了）。
• 如果 JSON 中有额外的详细信息，则失败（前面的示例将失败，因为有一个 extra_attribute）。
• 返回字典而不是类型化对象或 SimpleNamespace，而不是预期类型的​​对象。
• 需要为每个不同的类编写序列化/反序列化代码，这非常容易出错。

Answer 1

我通常使用Marshmallow 项目来处理 JSON 序列化、反序列化和验证。当与marshmallow-dataclass结合使用时，或者当使用 SQLAlchemy 数据库模型marshmallow-sqlalchemy时，您可以直接从现有对象定义生成 Marshmallow 模式。您使用模型本身的实例，即数据类定义的类实例或 SQLAlchemy ORM 模型实例。

Marshmallow 模式还允许您定义 JSON 文档中的额外值会发生什么情况；您可以忽略这些，或者为它们抛出异常，并根据模型进行更改（模型可以根据需要嵌套）。您也可以将模式重用于字段的子集。

您的小样本模型，使用marshmallow-dataclass，可以定义为：

import marshmallow
from marshmallow_dataclass import dataclass
from typing import List

class BaseSchema(marshmallow.Schema):
    class Meta:
        unknown = marshmallow.EXCLUDE

@dataclass(base_schema=BaseSchema)
class Address:
    city: str
    postcode: str

@dataclass(base_schema=BaseSchema)
class Person:
    name: str
    addresses: List[Address]

除了pip install marshmallow-dataclass尝试运行上述内容之前，就是这样。此示例使用显式基本架构将unknown配置设置为EXCLUDE，这意味着：加载时忽略 JSON 中的额外属性。

要从 JSON 数据反序列化或序列化为JSON，请创建模式的实例；每个dataclass类都有一个Schema引用相应（生成的）Marshmallow 架构对象的属性：

>>> schema = Person.Schema()
>>> json = '{ "name": "John", "addresses": [{ "postcode": "EC2 2FA", "city": "London" }, { "city": "Paris", "postcode": "545887", "extra_attribute": "" }]}'
>>> p = schema.loads(json)
>>> p
Person(name='John', addresses=[Address(city='London', postcode='EC2 2FA'), Address(city='Paris', postcode='545887')])
>>> print(type(p)) # should print Person
<class '__main__.Person'>
>>> for a in p.addresses:
...     print(type(a)) # prints Address
...     print(a.city) # should print London then Paris
...
<class '__main__.Address'>
London
<class '__main__.Address'>
Paris
>>> schema.dumps(p)
'{"name": "John", "addresses": [{"postcode": "EC2 2FA", "city": "London"}, {"postcode": "545887", "city": "Paris"}]}'

和Schema.loads()方法Schema.dumps()接受并生成 JSON 字符串。您还可以通过和来使用普通 Python 字典和列表（可以使用标准库模块序列化为 JSON 的类型json）。Schema.load()Schema.dump()

对于更复杂的设置，您可能需要为字段配置精确的验证规则，或从序列化中排除某些字段。您可以使用标准dataclasses.field()函数来执行此操作，通过参数传入 Marshmallow 字段选项metadata。marshmallow-dataclass可以计算出要使用的确切Marshmallow 字段类型，但您始终可以覆盖它。您可以使用该类为此定义可重用的定义NewType()；允许您像在项目中SomeType = NewType("SomeType", python_type, field=MarshmallowField, **field_args)一样标记数据类字段。field_name: SomeType

至少对我来说，Marshmallow 是序列化和反序列化的瑞士军刀项目，并且有很多资源与 Marshmallow 集成。例如，我目前正在考虑为客户构建多个 RESTFul API，并且我肯定会使用Flask-Smorest来定义 API 端点并同时生成 OpenAPI 文档。我所要做的就是为此创建 SQLAlchemy 模型，真的。

下面是一个基于您的人员和地址架构的 Flask RESTful API 示例，但作为 SQLALchemy 模型，充当 RESTful API：

# pip install Flask flask-marshmallow flask-smorest flask-sqlalchemy marshmallow-sqlalchemy 

import marshmallow
from flask import Flask
from flask.views import MethodView
from flask_marshmallow import Marshmallow
from flask_smorest import Api, Blueprint, abort
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['API_TITLE'] = 'ContactBook'
app.config['API_VERSION'] = 'v1'
app.config['OPENAPI_VERSION'] = '3.0.3'
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///:memory:'
api = Api(app)
db = SQLAlchemy(app)
ma = Marshmallow(app)

class Address(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    city = db.Column(db.String)
    postcode = db.Column(db.String)
    person_id = db.Column(db.Integer, db.ForeignKey('person.id'), nullable=False)

class Person(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String)
    addresses = db.relationship('Address', backref='person', lazy=True)

# create tables in the (in-memory, temporary) database
db.create_all()

class BaseSQLAlchemyAutoSchema(ma.SQLAlchemyAutoSchema):
    def update(self, instance, **data):
        for fname in self.fields:
            if fname not in data:
                continue
            setattr(instance, fname, data.get(fname))

class AddressSchema(BaseSQLAlchemyAutoSchema):
    class Meta:
        table = Address.__table__

class PersonSchema(BaseSQLAlchemyAutoSchema):
    class Meta:
        table = Person.__table__

    addresses = ma.List(ma.Nested(AddressSchema(unknown=marshmallow.EXCLUDE)))

class PersonQueryArgsSchema(ma.Schema):
    name = ma.String()
    city = ma.String()

blp = Blueprint(
    "people", "people", url_prefix="/people", description="Operations on people"
)

@blp.route("/")
class People(MethodView):
    @blp.arguments(PersonQueryArgsSchema, location="query")
    @blp.response(200, PersonSchema(many=True))
    def get(self, args):
        """List people"""
        query = Person.query
        if args.get("name"):
            query = query.filter(Person.name == args["name"])
        if args.get("city"):
            query = query.filter(Person.addresses.any(Address.city == args["city"]))
        return query

    @blp.arguments(PersonSchema(unknown=marshmallow.EXCLUDE))
    @blp.response(201, PersonSchema)
    def post(self, new_person):
        """Add a new person"""
        addresses = new_person.pop("addresses", ())
        person = Person(**new_person)
        for address in addresses:
            person.addresses.append(Address(**address))
        db.session.add(person)
        db.session.commit()
        return person

@blp.route("/<person_id>")
class PersonById(MethodView):
    @blp.response(200, PersonSchema)
    def get(self, person_id):
        """Get person by ID"""
        return Person.query.get_or_404(person_id)

    @blp.arguments(PersonSchema(unknown=marshmallow.EXCLUDE, exclude=('addresses',)))
    @blp.response(200, PersonSchema)
    def put(self, updated_person_data, person_id):
        """Update existing person"""
        person = Person.query.get_or_404(person_id)
        PersonSchema().update(person, **updated_person_data)
        db.session.commit()
        return person

    @blp.response(204)
    def delete(self, person_id):
        """Delete person"""
        db.session.delete(Person.query.get_or_404(person_id))

api.register_blueprint(blp)

瞧，功能齐全的 REST API，让我们可以列出、更新、创建和删除Person条目。