如何从带有连接的数据库查询中生成嵌套 JSON？使用 Python / SQLAlchemy

Question

我有一个特定的用例，但我的问题通常与执行此操作的最佳方式有关。

我有三张桌子

订单 - 主键 order_id

OrderLine - 链接表 与 order_id、product_id 和数量。一个订单有 1 个或多个订单行

产品-主键product_id，每个订单行有一个产品

在 sqlachemy / python 中我如何生成嵌套的 JSON：

{
    "orders": [
        {
            "order_id": 1
            "some_order_level_detail": "Kansas"
            "order_lines": [
                {
                    "product_id": 1,
                    "product_name": "Clawhammer",
                    "quantity": 5
                },
                ...
            ]
        },
        ...
    ]
}

可能的想法

破解连续查询

如果可能的话，我想摆脱的第一个想法是使用列表理解和蛮力方法。

def get_json():
    answer = {
        "orders": [
            {
                "order_id": o.order_id,
                "some_order_level_detail": o.some_order_level_detail,
                "order_lines": [
                    {
                        "product_id": 1,
                        "product_name": Product.query.get(o_line.product_id).product_name,
                        "quantity": 5
                    }
                    for o_line in OrderLine.query.filter(order_id=o.order_id).all()
                ]
            }
            for o in Order.query.all()
        ]
    }

这很难保持将查询与 json 混合。理想情况下，我想先进行查询...

先获取连接结果，然后以某种方式操作

第二个想法是做一个连接查询来连接OrderLine中每行显示的三个表的订单和产品详细信息。

我对 pythonista 的问题是有没有一种很好的方法可以将它转换为嵌套的 json。

另一种方式？

这似乎真的是一个普遍的要求我真的想知道这种事情是否有书籍方法？ 有没有this的SQLAchemy版本@

Answer 1

查看marshmallow-sqlalchemy，因为它完全符合您的要求。

我强烈建议不要将序列化直接烘焙到模型中，因为您最终将有两个服务请求相同的数据，但以不同的方式进行序列化（例如，包括更少或更多的嵌套关系以提高性能），并且您将要么最终导致（1）你的测试套件会错过很多错误，除非你正在检查每个字段，或者（2）序列化的数据比你需要的多，你会遇到性能问题，因为你的复杂性应用规模。

使用 marshmallow-sqlalchemy，您需要为要序列化的每个模型定义一个模式。是的，这有点额外的样板，但相信我 - 你最终会更快乐。

我们像这样使用 flask-sqlalchemy 和 marshmallow-sqlalchemy 构建应用程序（也强烈推荐 factory_boy 以便您可以模拟您的服务并编写单元测试来代替需要接触数据库的集成测试）：

# models

class Parent(Base):
    __tablename__ = 'parent'
    id = Column(Integer, primary_key=True)
    children = relationship("Child", back_populates="parent")

class Child(Base):
    __tablename__ = 'child'
    id = Column(Integer, primary_key=True)
    parent_id = Column(Integer, ForeignKey('parent.id'))
    parent = relationship('Parent', back_populates='children',
                          foreign_keys=[parent_id])

# schemas. Don't put these in your models. Avoid tight coupling here

from marshmallow_sqlalchemy import ModelSchema
import marshmallow as ma


class ParentSchema(ModelSchema):
    children = ma.fields.Nested(
        'myapp.schemas.child.Child', exclude=('parent',), many=True)
    class Meta(ModelSchema.Meta):
        model = Parent
        strict = True
        dump_only = ('id',)


class ChildSchema(ModelSchema):
    parent = ma.fields.Nested(
        'myapp.schemas.parent.Parent', exclude=('children',))
    class Meta(ModelSchema.Meta):
        model = Child
        strict = True
        dump_only = ('id',)

# services

class ParentService:
    '''
    This service intended for use exclusively by /api/parent
    '''
    def __init__(self, params, _session=None):
        # your unit tests can pass in _session=MagicMock()
        self.session = _session or db.session
        self.params = params

    def _parents(self) -> typing.List[Parent]:
        return self.session.query(Parent).options(
            joinedload(Parent.children)
        ).all()

    def get(self):
        schema = ParentSchema(only=(
            # highly recommend specifying every field explicitly
            # rather than implicit
            'id',
            'children.id',
        ))
        return schema.dump(self._parents()).data

# views

@app.route('/api/parent')
def get_parents():
    service = ParentService(params=request.get_json())
    return jsonify(data=service.get())


# test factories
class ModelFactory(SQLAlchemyModelFactory):
    class Meta:
        abstract = True
        sqlalchemy_session = db.session

class ParentFactory(ModelFactory):
    id = factory.Sequence(lambda n: n + 1)
    children = factory.SubFactory('tests.factory.children.ChildFactory')

class ChildFactory(ModelFactory):
    id = factory.Sequence(lambda n: n + 1)
    parent = factory.SubFactory('tests.factory.parent.ParentFactory')

# tests
from unittest.mock import MagicMock, patch

def test_can_serialize_parents():
    parents = ParentFactory.build_batch(4)
    session = MagicMock()
    service = ParentService(params={}, _session=session)
    assert service.session is session
    with patch.object(service, '_parents') as _parents:
        _parents.return_value = parents
        assert service.get()[0]['id'] == parents[0].id
        assert service.get()[1]['id'] == parents[1].id
        assert service.get()[2]['id'] == parents[2].id
        assert service.get()[3]['id'] == parents[3].id

Answer 2

我会为每个模型添加一个.json() 方法，以便它们相互调用。它本质上是您的“被黑”解决方案，但更具可读性/可维护性。你的Order 模型可能有：

def json(self):
    return {
        "id": self.id,
        "order_lines": [line.json() for line in self.order_lines]
    }

您的OrderLine 模型可能有：

def json(self):
    return {
        "product_id": self.product_id,
        "product_name": self.product.name,
        "quantity": self.quantity
    }

您的顶级资源（您在其中提出订单请求）可以执行以下操作：

...
orders = Order.query.all()
return {"orders": [order.json() for order in orders]}
...

这就是我通常如何构建此 JSON 要求的方式。

Answer 3

在此线程 Flask Sqlalchmey - Marshmallow Nested Schema fails for joins with filter ( where ) conditions 中检查我的答案，并使用您在架构中包含的 Marshmallow 包，如下所示：

name = fields.Nested(Schema, many=True)