我有一个指定的用例,但我的问题通常与执行此操作的最佳方法有关。
我有三张桌子
订单-主键order_id OrderLine- 具有order_id,product_id和数量的 链接表 。一个订单有1个或更多订单行 产品-主键product_id,每个订单行都有一个产品
订单-主键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订单和产品详细信息中每行显示的三个表。
OrderLine
我对pythonista的问题是有一种将其转换为嵌套json的好方法。
这真的看起来像是一个常见的要求,我真的想知道对于这种事情是否有预定方法?是否有一个SQLAchemy版本本
研究一下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