如何通过带有联接的数据库查询生成嵌套的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的好方法。

其他方式？

这真的看起来像是一个常见的要求，我真的想知道对于这种事情是否有预定方法？是否有一个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