Skip to content

flowmachine.core.server.query_schemas.flowmachine_query

Class FlowmachineQuerySchema

FlowmachineQuerySchema(*, only: Union[Sequence[str], Set[str], NoneType] = None, exclude: Union[Sequence[str], Set[str]] = (), many: bool = False, context: Union[Dict, NoneType] = None, load_only: Union[Sequence[str], Set[str]] = (), dump_only: Union[Sequence[str], Set[str]] = (), partial: Union[bool, Sequence[str], Set[str]] = False, unknown: Union[str, NoneType] = None)
Source: flowmachine/core/server/query_schemas/flowmachine_query.py

This is a special kind of schema that actually multiplexes other schemas based on object type. When serializing values, it uses get_obj_type() method to get object type name. Then it uses type_schemas name-to-Schema mapping to get schema for that particular object type, serializes object using that schema and adds an extra "type" field with name of object type. Deserialization is reverse. Example: class Foo(object): def init(self, foo): self.foo = foo class Bar(object): def init(self, bar): self.bar = bar class FooSchema(marshmallow.Schema): foo = marshmallow.fields.String(required=True) @marshmallow.post_load def make_foo(self, data, **kwargs): return Foo(**data) class BarSchema(marshmallow.Schema): bar = marshmallow.fields.Integer(required=True) @marshmallow.post_load def make_bar(self, data, **kwargs): return Bar(**data) class MyUberSchema(marshmallow.OneOfSchema): type_schemas = { 'foo': FooSchema, 'bar': BarSchema, } def get_obj_type(self, obj): if isinstance(obj, Foo): return 'foo' elif isinstance(obj, Bar): return 'bar' else: raise Exception('Unknown object type: %s' % repr(obj)) MyUberSchema().dump([Foo(foo='hello'), Bar(bar=123)], many=True) # => [{'type': 'foo', 'foo': 'hello'}, {'type': 'bar', 'bar': 123}] You can control type field name added to serialized object representation by setting type_field class property.

Attributes

Methods

dict_class

dict_class
Source: marshmallow/schema.py

set_class

set_class
Source: marshmallow/schema.py