研究一下marshmallow-sqlalchemy,它确实可以满足您的需求。
我强烈建议您不要将序列化直接烘焙到模型中,因为您最终将有两个服务请求相同的数据,但是以不同的方式进行序列化(例如,包括更少或更多的嵌套关系以提高性能),您要么最终选择要么(1)测试套件会遗漏许多错误,除非您要逐字检查每个字段,或者(2)序列化的数据量超过所需数量,并且随着应用程序的复杂性扩展,您将遇到性能问题。
使用marshmallow-sqlalchemy,您需要为要序列化的每个模型定义一个架构。是的,这有点多余,但是请相信我-最终您会更加快乐。
我们像这样使用flask-sqlalchemy和marshmallow-
sqlalchemy构建应用程序(也强烈建议使用factory_boy,以便您可以模拟服务并编写单元测试来代替需要接触数据库的集成测试):
# modelsclass 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 herefrom marshmallow_sqlalchemy import ModelSchemaimport marshmallow as maclass 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',)# servicesclass 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 factoriesclass ModelFactory(SQLAlchemyModelFactory): class meta: abstract = True sqlalchemy_session = db.sessionclass 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')# testsfrom unittest.mock import MagicMock, patchdef 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
欢迎分享,转载请注明来源:内存溢出
评论列表(0条)