python - 如何通过连接从数据库查询生成嵌套 JSON？使用 Python/SQLAlchemy

Question

我有一个特定的用例，但我的问题与一般情况下执行此操作的最佳方法有关。

我有三张 table

Order - primary key order_id

OrderLine - Linking table with order_id, product_id and quantity. An order has 1 or more order lines

Product - primary key product_id, each order line has one product

在 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