如何通过带有联接的数据库查询生成嵌套的JSON?使用Python/SQLAlchemy [英] How do I produce nested JSON from database query with joins? Using Python / SQLAlchemy
问题描述
我有一个指定的用例，但我的问题通常与执行此操作的最佳方法有关.
I have a specify use case but my question pertains to the best way of doing this in general.
我有三个桌子
订单-主键order_id
Order - primary key order_id
OrderLine-链接表，其中包含order_id，product_id和数量.一条订单有1条或更多条订单行
OrderLine - Linking table with order_id, product_id and quantity. An order has 1 or more order lines
产品-主键product_id，每个订单行都有一个产品
Product - primary key product_id, each order line has one product
在sqlachemy/python中，如何按照以下方式生成嵌套的JSON:
In sqlachemy / python how do I generate nested JSON along the lines of:
{
"orders": [
{
"order_id": 1
"some_order_level_detail": "Kansas"
"order_lines": [
{
"product_id": 1,
"product_name": "Clawhammer",
"quantity": 5
},
...
]
},
...
]
}
潜在想法
请继续进行连续的查询
如果可能的话，我想摆脱的第一个想法是使用列表混合和蛮力方法.
Potential Ideas
Hack away doing successive queries
First idea which I want to get away from if possible is using list comprehesion and a brute force approach.
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混合在一起.理想情况下，我想先进行查询...
This gets hard to maintain mixing the queries with json. Ideally I'd like to do a query first...
第二个想法是进行联接查询，以联接按OrderLine顺序和产品详细信息每行显示的三个表.
The second idea is to do a join query to join the three tables showing per row in OrderLine the order and product details.
我对pythonista的问题是有一种将其转换为嵌套json的好方法.
My question to pythonista out there is is there a nice way to convert this to nested json.
这真的看起来像是一个普遍的要求，我真的很想知道对于这种事情是否有预定方法? 是否有 this
This really seems like such a common requirement I'm really wondering whether there is a book method for this sort of thing? Is there an SQLAchemy version of this
推荐答案
查看 marshmallow-sqlalchemy ，因为它确实可以满足您的需求.
Look into marshmallow-sqlalchemy, as it does exactly what you're looking for.
我强烈建议您不要将序列化直接烘焙到模型中，因为您最终将有两个服务请求相同的数据，但是以不同的方式进行序列化(例如，包括更少或更多的嵌套关系以提高性能)，并且您将要么以(1)测试套件将遗漏的许多错误(除非您正在逐字检查每个字段)结束，要么以(2)序列化的数据量超过所需数量而结束，并且由于性能的复杂性，您将遇到性能问题应用规模.
I strongly advise against baking your serialization directly into your model, as you will eventually have two services requesting the same data, but serialized in a different way (including fewer or more nested relationships for performance, for instance), and you will either end up with either (1) a lot of bugs that your test suite will miss unless you're checking for literally every field or (2) more data serialized than you need and you'll run into performance issues as the complexity of your application scales.
使用marshmallow-sqlalchemy，您需要为要序列化的每个模型定义一个架构.是的，这有点多余，但是请相信我-最终您会更加快乐.
With marshmallow-sqlalchemy, you'll need to define a schema for each model you'd like to serialize. Yes, it's a bit of extra boilerplate, but believe me - you will be much happier in the end.
我们使用像这样的flask-sqlalchemy和棉花糖-sqlalchemy构建应用程序(强烈建议 factory_boy ，以便您可以模拟服务并编写单元测试来代替需要接触数据库的集成测试):
We build applications using flask-sqlalchemy and marshmallow-sqlalchemy like this (also highly recommend factory_boy so that you can mock your service and write unit tests in place of of integration tests that need to touch the database):
# 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
这篇关于如何通过带有联接的数据库查询生成嵌套的JSON?使用Python/SQLAlchemy的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
