Python marshmallow 模块，Schema() 实例源码

def test_if_encode_raises_exception_with_invalid_data_and_strict_schema():
    class StrictSchema(marshmallow.Schema):
        uuid_field = fields.UUID(required=True)

        class Meta:
            strict = True

    class Event(structures.Model):
        class Meta:
            schema = StrictSchema
            type_name = 'Event'

    data = Event(uuid_field='not an uuid')
    with pytest.raises(exceptions.EncodingError) as excinfo:
        encoding.encode(data)
    assert str(excinfo.value) == (
        "({'uuid_field': ['Not a valid UUID.']}, '')")

def test_if_raises_exception_with_invalid_data_and_strict_schema(self):
        class StrictSchema(marshmallow.Schema):
            uuid_field = fields.UUID(required=True)

            class Meta:
                strict = True

        class Event(structures.Model):
            class Meta:
                schema = StrictSchema
                type_name = 'Event'

        data = '{"uuid_field": "not an uuid"}'
        with pytest.raises(exceptions.DecodingError) as excinfo:
            encoding.decode(type=Event, encoded_data=data)
        assert str(excinfo.value) == (
            "({'uuid_field': ['Not a valid UUID.']}, '')")

def test_generate_unmarshall_method_bodies_with_load_from():
    class OneFieldSchema(Schema):
        foo = fields.Integer(load_from='bar', allow_none=True)
    context = JitContext(is_serializing=False, use_inliners=False)
    result = str(generate_transform_method_body(OneFieldSchema(),
                                                DictSerializer(context),
                                                context))
    expected = '''\
def DictSerializer(obj):
    res = {}
    __res_get = res.get
    if "foo" in obj:
        res["foo"] = _field_foo__deserialize(obj["foo"], "bar", obj)
    if "foo" not in res:
        if "bar" in obj:
            res["foo"] = _field_foo__deserialize(obj["bar"], "bar", obj)
    return res'''
    assert expected == result

def test_generate_unmarshall_method_bodies_required():
    class OneFieldSchema(Schema):
        foo = fields.Integer(required=True)
    context = JitContext(is_serializing=False, use_inliners=False)
    result = str(generate_transform_method_body(OneFieldSchema(),
                                                DictSerializer(context),
                                                context))
    expected = '''\
def DictSerializer(obj):
    res = {}
    __res_get = res.get
    res["foo"] = _field_foo__deserialize(obj["foo"], "foo", obj)
    if "foo" not in res:
        raise ValueError()
    if __res_get("foo", res) is None:
        raise ValueError()
    return res'''
    assert expected == result

def test_secure_field(app):
    class SecureSchema(Schema):
        token = SecureField()

    schema = SecureSchema()

    # case 1: plaintext
    data = {'token': 'abc'}
    result = schema.load(data)
    assert result.data['token'] == 'abc'

    # case 2: valid secure token
    data = {'token': {'secure': SecureToken.encrypt('def')}}
    result = schema.load(data)
    assert result.data['token'] == 'def'

    # case 3: invalid secure token
    data = {'token': {'secure': 'gAAAAABYmoldCp-EQGUKCppiqmVOu2jLrAKUz6E2e4aOMMD8Vu0VKswmJexHX6vUEoxVYKFUlSonPb91QKXZBEZdBezHzJMCHg=='}}  # NOQA
    result = schema.load(data)
    assert result.data['token'] == ''

def __call__(self, info):
        """
        If a schema is present, replace value with output from schema.dump(..).
        """
        original_render = super().__call__(info)

        def schema_render(value, system):
            request = system.get('request')
            if (request is not None and isinstance(getattr(request, 'render_schema', None), Schema)):
                try:
                    value, errors = request.render_schema.dump(value)
                except Exception:
                    errors = True

                if errors:
                    raise HTTPInternalServerError(body="Serialization failed.")

            return original_render(value, system)

        return schema_render

def jsonify(self, obj, many=sentinel, *args, **kwargs):
        """Return a JSON response containing the serialized data.


        :param obj: Object to serialize.
        :param bool many: Whether `obj` should be serialized as an instance
            or as a collection. If unset, defaults to the value of the
            `many` attribute on this Schema.
        :param kwargs: Additional keyword arguments passed to `flask.jsonify`.

        .. versionchanged:: 0.6.0
            Takes the same arguments as `marshmallow.Schema.dump`. Additional
            keyword arguments are passed to `flask.jsonify`.

        .. versionchanged:: 0.6.3
            The `many` argument for this method defaults to the value of
            the `many` attribute on the Schema. Previously, the `many`
            argument of this method defaulted to False, regardless of the
            value of `Schema.many`.
        """
        if many is sentinel:
            many = self.many
        data = self.dump(obj, many=many).data
        return flask.jsonify(data, *args, **kwargs)

def _register_deduced_schemas(Base):
    def setup_schema_fn():
        # Generate missing schemas
        for class_ in Base._decl_class_registry.values():
            if hasattr(class_, '__tablename__') and not hasattr(class_, '__marshmallow__'):
                if class_.__name__.endswith('Schema'):
                    raise ModelConversionError(
                        "For safety, setup_schema can not be used when a"
                        "Model class ends with 'Schema'"
                    )

                class Meta(BaseSchema.Meta):
                    model = class_

                schema_class_name = '%sSchema' % class_.__name__

                schema_class = type(
                    schema_class_name,
                    (BaseSchema,),
                    {'Meta': Meta}
                )

                setattr(class_, '__marshmallow__', schema_class)

    return setup_schema_fn

def test_custom_base_schema(self):

        class MyBaseSchema(marshmallow.Schema):
            name = marshmallow.fields.Int()
            age = marshmallow.fields.Int()

        ma_schema_cls = self.User.schema.as_marshmallow_schema(base_schema_cls=MyBaseSchema)
        assert issubclass(ma_schema_cls, MyBaseSchema)

        schema = ma_schema_cls()
        ret = schema.dump({'name': "42", 'age': 42, 'dummy': False})
        assert not ret.errors
        assert ret.data == {'name': "42", 'age': 42}
        ret = schema.load({'name': "42", 'age': 42, 'dummy': False})
        assert ret.errors == {'_schema': ['Unknown field name dummy.']}
        ret = schema.load({'name': "42", 'age': 42})
        assert not ret.errors
        assert ret.data == {'name': "42", 'age': 42}

def generate_json_schema(cls, schema, context=DEFAULT_DICT):
        """Generate a JSON Schema from a Marshmallow schema.

        Args:
            schema (marshmallow.Schema|str): The Marshmallow schema, or the
                Python path to one, to create the JSON schema for.

        Keyword Args:
            file_pointer (file, optional): The path or pointer to the file
                to write this schema to. If not provided, the schema will be
                dumped to ``sys.stdout``.

        Returns:
            dict: The JSON schema in dictionary form.
        """
        schema = cls._get_schema(schema)

        # Generate the JSON Schema
        return cls(context=context).dump(schema).data

def _get_schema(cls, schema):
        """Method that will fetch a Marshmallow schema flexibly.

        Args:
            schema (marshmallow.Schema|str): Either the schema class, an
                instance of a schema, or a Python path to a schema.

        Returns:
            marshmallow.Schema: The desired schema.

        Raises:
            TypeError: This is raised if the provided object isn't
                a Marshmallow schema.
        """
        if isinstance(schema, string_types):
            schema = cls._get_object_from_python_path(schema)

        if isclass(schema):
            schema = schema()

        if not isinstance(schema, Schema):
            raise TypeError("The schema must be a path to a Marshmallow "
                            "schema or a Marshmallow schema.")

        return schema

def __init__(self, cls):
        """Initialize the resource."""
        self._collection = None
        super(MongoOptions, self).__init__(cls)
        self.name = self.meta and getattr(self.meta, 'name', None)
        if not self.collection:
            return

        self.name = self.name or str(self.collection.name)

        if not cls.Schema:
            meta = type('Meta', (object,), self.schema_meta)
            cls.Schema = type(
                self.name.title() + 'Schema', (MongoSchema,), dict({'Meta': meta}, **self.schema))

def get_schema(self, resource=None, **kwargs):
        """Create the resource schema."""
        return self.Schema(instance=resource)  # noqa

def test_if_encode_raises_exception_with_invalid_data_and_not_strict_schema():
    class NotStrictSchema(marshmallow.Schema):
        uuid_field = fields.UUID(required=True)

    class Event(structures.Model):
        class Meta:
            schema = NotStrictSchema
            type_name = 'Event'

    data = Event(uuid_field='not an uuid')
    with pytest.raises(exceptions.EncodingError) as excinfo:
        encoding.encode(data)
    assert str(excinfo.value) == (
        "({'uuid_field': ['Not a valid UUID.']}, '')")

def schema_class(self):
        class Schema(marshmallow.Schema):
            uuid_field = fields.UUID(required=True)
            string_field = fields.String(required=False)
        return Schema

def test_if_raises_exception_with_invalid_data_and_not_strict_schema(self):
        class NotStrictSchema(marshmallow.Schema):
            uuid_field = fields.UUID(required=True)

        class Event(structures.Model):
            class Meta:
                schema = NotStrictSchema
                type_name = 'Event'

        data = '{"uuid_field": "not an uuid"}'
        with pytest.raises(exceptions.DecodingError) as excinfo:
            encoding.decode(type=Event, encoded_data=data)
        assert str(excinfo.value) == (
            "({'uuid_field': ['Not a valid UUID.']}, '')")

def configure_retrieve(self, ns, definition):
        """
        Register a retrieve endpoint.

        The definition's func should be a retrieve function, which must:
        - accept kwargs for path data
        - return an item or falsey

        :param ns: the namespace
        :param definition: the endpoint definition

        """
        request_schema = definition.request_schema or Schema()

        @self.add_route(ns.instance_path, Operation.Retrieve, ns)
        @qs(request_schema)
        @response(definition.response_schema)
        @wraps(definition.func)
        def retrieve(**path_data):
            headers = dict()
            request_data = load_query_string_data(request_schema)
            response_data = require_response_data(definition.func(**merge_data(path_data, request_data)))
            definition.header_func(headers)
            response_format = self.negotiate_response_content(definition.response_formats)
            return dump_response_data(
                definition.response_schema,
                response_data,
                headers=headers,
                response_format=response_format,
            )

        retrieve.__doc__ = "Retrieve a {} by id".format(ns.subject_name)

def configure_retrievefor(self, ns, definition):
        """
        Register a relation endpoint.

        The definition's func should be a retrieve function, which must:
        - accept kwargs for path data and optional request data
        - return an item

        The definition's request_schema will be used to process query string arguments, if any.

        :param ns: the namespace
        :param definition: the endpoint definition

        """
        request_schema = definition.request_schema or Schema()

        @self.add_route(ns.relation_path, Operation.RetrieveFor, ns)
        @qs(request_schema)
        @response(definition.response_schema)
        @wraps(definition.func)
        def retrieve(**path_data):
            headers = dict()
            request_data = load_query_string_data(request_schema)
            response_data = require_response_data(definition.func(**merge_data(path_data, request_data)))
            definition.header_func(headers)
            response_format = self.negotiate_response_content(definition.response_formats)
            return dump_response_data(
                definition.response_schema,
                response_data,
                headers=headers,
                response_format=response_format,
            )

        retrieve.__doc__ = "Retrieve {} relative to a {}".format(pluralize(ns.object_name), ns.subject_name)

def create_upload_func(self, ns, definition, path, operation):
        request_schema = definition.request_schema or Schema()
        response_schema = definition.response_schema or Schema()

        @self.add_route(path, operation, ns)
        @wraps(definition.func)
        def upload(**path_data):
            request_data = load_query_string_data(request_schema)

            if not request.files:
                raise BadRequest("No files were uploaded")

            uploads = [
                temporary_upload(name, fileobj)
                for name, fileobj
                in request.files.items()
                if not self.exclude_func(name, fileobj)
            ]
            with nested(*uploads) as files:
                response_data = definition.func(files, **merge_data(path_data, request_data))
                if response_data is None:
                    return "", 204

            return dump_response_data(response_schema, response_data, operation.value.default_code)

        if definition.request_schema:
            upload = qs(definition.request_schema)(upload)
        if definition.response_schema:
            upload = response(definition.response_schema)(upload)
        return upload

def make_paginated_list_schema_class(cls, ns, item_schema):
        """
        Generate a schema class that represents a paginted list of items.

        """
        class PaginatedListSchema(Schema):
            __alias__ = "{}_list".format(ns.subject_name)
            items = fields.List(fields.Nested(item_schema), required=True)
            _links = fields.Raw()

        return PaginatedListSchema

def make_paginated_list_schema_class(cls, ns, item_schema):
        class PaginatedListSchema(Schema):
            __alias__ = "{}_list".format(ns.subject_name)

            offset = fields.Integer(required=True)
            limit = fields.Integer(required=True)
            count = fields.Integer(required=True)
            items = fields.List(fields.Nested(item_schema), required=True)
            _links = fields.Raw()

            @property
            def csv_column_order(self):
                return getattr(item_schema, "csv_column_order", None)

        return PaginatedListSchema

def test_offset_limit_page_to_paginated_list():
    graph = create_object_graph(name="example", testing=True)

    ns = Namespace("foo")

    @graph.flask.route("/", methods=["GET"], endpoint="foo.search.v1")
    def search():
        pass

    with graph.flask.test_request_context():
        page = OffsetLimitPage(
            offset=10,
            limit=10,
            foo="bar",
        )
        result = [], 0
        paginated_list, headers = page.to_paginated_list(result, _ns=ns, _operation=Operation.Search)

        schema_cls = page.make_paginated_list_schema_class(ns, Schema())
        data = schema_cls().dump(paginated_list).data
        assert_that(
            data,
            is_(equal_to(dict(
                offset=10,
                limit=10,
                count=0,
                items=[],
                _links=dict(
                    self=dict(
                        href="http://localhost/?offset=10&limit=10&foo=bar",
                    ),
                    prev=dict(
                        href="http://localhost/?offset=0&limit=10&foo=bar",
                    ),
                ),
            ))))

def _to_jsonschema(type_):
    if isinstance(type_, marshmallow.Schema):
        return _jsonschema.dump_schema(type_)
    elif type_ in six.integer_types:
        return {'type': 'number', 'format': 'integer'}
    elif type_ == float:
        return {'type': 'number', 'format': 'float'}
    elif type_ == decimal.Decimal:
        return {'type': 'string', 'format': 'decimal'}
    elif type_ == uuid.UUID:
        return {'type': 'string', 'format': 'uuid'}
    elif type_ == datetime.datetime:
        return {'type': 'string', 'format': 'date-time'}
    elif type_ == datetime.date:
        return {'type': 'string', 'format': 'date'}
    elif type_ == datetime.time:
        return {'type': 'string', 'format': 'time'}
    elif type_ == dict:
        return {'type': 'object'}
    elif type_ == six.text_type or type_ == six.binary_type:
        return {'type': 'string'}
    elif type_ is None:
        return {'type': 'null'}
    elif type_ == list:
        return {'type': 'array'}
    elif type_ == bool:
        return {'type': 'boolean'}
    elif issubclass(type_, typing.MutableSequence[typing.T]):
        items_type = type_.__parameters__[0]
        if issubclass(items_type, marshmallow.Schema):
            items_type = items_type()
        return {
            'type': 'array',
            'items': _to_jsonschema(items_type),
        }
    else:
        raise ValueError('unsupported return type: %s' % type_)

def load_data(self, data):
        """
        Deserialize data to an object defined by its Schema and raises a
        ValidationError if there are eny errors.
        :param data:
        :return:
        """
        data, errors = self.load(data)
        if errors:
            raise ApiValidationError(errors)
        return data

def simple_schema():
    class InstanceSchema(Schema):
        key = fields.String()
        value = fields.Integer(default=0)
    return InstanceSchema()

def nested_circular_ref_schema():
    class NestedStringSchema(Schema):
        key = fields.String()
        me = fields.Nested('NestedStringSchema')
    return NestedStringSchema()

def nested_schema():
    class GrandChildSchema(Schema):
        bar = fields.String()
        raz = fields.String()

    class SubSchema(Schema):
        name = fields.String()
        value = fields.Nested(GrandChildSchema)

    class NestedSchema(Schema):
        key = fields.String()
        value = fields.Nested(SubSchema, only=('name', 'value.bar'))
        values = fields.Nested(SubSchema, exclude=('value', ), many=True)
    return NestedSchema()

def optimized_schema():
    class OptimizedSchema(Schema):
        class Meta:
            jit_options = {
                'no_callable_fields': True,
                'expected_marshal_type': 'object'
            }
        key = fields.String()
        value = fields.Integer(default=0, as_string=True)
    return OptimizedSchema()

def schema():
    class BasicSchema(Schema):
        class Meta:
            ordered = True
        foo = fields.Integer(attribute='@#')
        bar = fields.String()
        raz = fields.Method('raz_')
        meh = fields.String(load_only=True)
        blargh = fields.Boolean()

        def raz_(self, obj):
            return 'Hello!'
    return BasicSchema()

def test_generate_unmarshall_method_bodies():
    class OneFieldSchema(Schema):
        foo = fields.Integer()
    context = JitContext(is_serializing=False, use_inliners=False)
    result = generate_method_bodies(OneFieldSchema(), context)
    expected = '''\
def InstanceSerializer(obj):
    res = {}
    __res_get = res.get
    res["foo"] = _field_foo__deserialize(obj.foo, "foo", obj)
    if __res_get("foo", res) is None:
        raise ValueError()
    return res
def DictSerializer(obj):
    res = {}
    __res_get = res.get
    if "foo" in obj:
        res["foo"] = _field_foo__deserialize(obj["foo"], "foo", obj)
    if __res_get("foo", res) is None:
        raise ValueError()
    return res
def HybridSerializer(obj):
    res = {}
    __res_get = res.get
    try:
        value = obj["foo"]
    except (KeyError, AttributeError, IndexError, TypeError):
        value = obj.foo
    res["foo"] = _field_foo__deserialize(value, "foo", obj)
    if __res_get("foo", res) is None:
        raise ValueError()
    return res'''
    assert expected == result

def test_jit_bails_with_get_attribute():
    class DynamicSchema(Schema):
        def get_attribute(self, obj, attr, default):
            pass
    marshal_method = generate_marshall_method(DynamicSchema())
    assert marshal_method is None

def test_jit_bails_nested_attribute():
    class DynamicSchema(Schema):
        foo = fields.String(attribute='foo.bar')

    marshal_method = generate_marshall_method(DynamicSchema())
    assert marshal_method is None

def test_list_field(app):
    class ListSchema(Schema):
        services = ListField(fields.String())

    schema = ListSchema()

    # case 1: scalar as list
    data = {'services': 'redis-server'}
    result = schema.load(data)
    assert result.data['services'] == ['redis-server']

    # case 2: list
    data = {'services': ['redis-server']}
    result = schema.load(data)
    assert result.data['services'] == ['redis-server']

def paginated(basequery, schema_type, offset=None, limit=None):
    """
    Paginate a sqlalchemy query

    :param basequery: The base query to be iterated upon
    :param schema_type: The ``Marshmallow`` schema to dump data with
    :param offset: (Optional) The offset into the data. If omitted it will 
                  be read from the query string in the ``?offset=`` argument. If
                  not query string, defaults to 0.
    :param limit: (Optional) The maximum results per page. If omitted it will 
                  be read from the query string in the ``?limit=`` argument. If
                  not query string, defaults to 20.

    :returns: The page's data in a namedtuple form ``(data=, errors=)``
    """

    if offset is None or limit is None:
        args = parser.parse(pagination_args, request)
        if offset is None:
            offset = args['offset']

        if limit is None:
            limit = args['limit']

    data = {
        'offset': offset,
        'limit': limit,
        'items': basequery.limit(limit).offset(offset),
        'totalItems': basequery.count()
    }

    class _Pagination(Schema):
        offset = fields.Integer()
        limit = fields.Integer()
        totalItems = fields.Integer()
        items = fields.Nested(schema_type, many=True)

    return _Pagination().dump(data)

def setUpClass(cls) -> None:
        """
        Add a service to the DB
        """
        AcceptanceTestCase.setUpClass()
        cls.service_name = 'Testing Service'
        cls.description = 'Description for the Testing Service'

        cls.job_registration_schema = JSONSchema(
            title='Job Registration Schema',
            description='Must be fulfilled for an experiment'
        ).dump(cls.JobRegistrationSchema())

        cls.job_result_schema = JSONSchema(
            title='Job Result Schema',
            description='Must be fulfilled to post results'
        ).dump(cls.JobRegistrationSchema())

        session = Session(bind=APP_FACTORY.engine, expire_on_commit=False)

        service_list = ServiceList(session)
        cls.service = service_list.new(
            cls.service_name, cls.description, cls.job_registration_schema,
            cls.job_result_schema
        )
        session.commit()

def dump(self, schema: Schema, many=False) -> JSON:
        result = self._json_schema_serializer.dump(schema, many=many).data
        self._add_title(result)
        self._add_description(result)
        self._add_schema(result)

        return result

def dumps(self, schema: Schema, many=False) -> str:
        return str(self.dump(schema, many=many))

def __init__(self, req_key='json', resp_key='result', force_json=True,
                 json_module=json):
        # type: (str, str, bool, type(json)) -> None
        """Instantiate the middleware object

        :param req_key: (default ``'json'``) the key on the
            ``req.context`` object where the parsed request body
            will be stored
        :param resp_key: (default ``'result'``) the key on the
            ``req.context`` object where the response parser will
            look to find data to serialize into the response body
        :param force_json: (default ``True``) whether requests
            and responses for resources *without* any defined
            Marshmallow schemas should be parsed as json anyway.
        :param json_module: (default ``simplejson``) the json module to
            use for  (de)serialization if no schema is available on a
            resource and ``force_json`` is ``True`` - if you would like
            to use an alternative serializer to the stdlib ``json``
            module for your Marshmallow schemas, you will have to
            specify using a schema metaclass, as defined in the
            `Marshmallow documentation`_

            .. _marshmallow documentation: http://marshmallow.readthedocs.io/
                en/latest/api_reference.html#marshmallow.Schema.Meta

        """
        log.debug(
            'Marshmallow.__init__(%s, %s, %s, %s)',
            req_key, resp_key, force_json, json_module
        )
        self._req_key = req_key
        self._resp_key = resp_key
        self._force_json = force_json
        self._json = json_module

def _get_specific_schema(resource, method, msg_type):
        # type: (object, str, str) -> Optional[Schema]
        """Return a specific schema or None

        If the provided resource has defined method-specific schemas
        or method-request/response-specific schemas, return that
        schema. If there are multiple schemas defined, the more
        specific ones will take precedence.

        Examples:
            - 'get_schema' for a 'GET' request & response
            - `post_schema' for a 'POST' request & response
            - 'post_request_schema' for a 'POST' request
            - 'post_response_schema' for a 'POST' response

        Return ``None`` if no matching schema exists

        :param resource: the resource object passed to
            ``process_response`` or ``process_resource``
        :param method: the (case-insensitive) HTTP method used
            for the request, e.g. 'GET' or 'POST'
        :param msg_type: a string 'request' or 'response'
            representing whether this was called from
            ``process_response`` or ``process_resource``
        """
        log.debug(
            'Marshmallow._get_specific_schema(%s, %s, %s)',
            resource, method, msg_type
        )

        sch_name = '%s_%s_schema' % (method.lower(), msg_type)
        specific_schema = getattr(resource, sch_name, None)
        if specific_schema is not None:
            return specific_schema

        sch_name = '%s_schema' % method.lower()
        specific_schema = getattr(resource, sch_name, None)
        return specific_schema

def _get_schema(cls, resource, method, msg_type):
        # type: (object, str, str) -> Optional[Schema]
        """Return a method-specific schema, a generic schema, or None

        If the provided resource has defined method-specific schemas
        or method-request/response-specific schemas, return that
        schema. If there are multiple schemas defined, the more
        specific ones will take precedence.

        Examples:
            - 'get_schema' for a 'GET' request & response
            - `post_schema' for a 'POST' request & response
            - 'post_request_schema' for a 'POST' request
            - 'post_response_schema' for a 'POST' response

        Otherwise, if the provided resource has defined a generic
        schema under ``resource.schema``, return that schema.

        Return ``None`` if neither of the above is found

        :param resource: the resource object passed to
            ``process_response`` or ``process_resource``
        :param method: the (case-insensitive) HTTP method used
            for the request, e.g. 'GET' or 'POST'
        :param msg_type: a string 'request' or 'response'
            representing whether this was called from
            ``process_response`` or ``process_resource``
        """
        log.debug(
            'Marshmallow._get_schema(%s, %s, %s)',
            resource, method, msg_type
        )
        specific_schema = cls._get_specific_schema(
            resource, method, msg_type
        )
        if specific_schema is not None:
            return specific_schema
        return getattr(resource, 'schema', None)

def schema_from_request(self, schema: Schema, partial=False):
        return schema.load(request.get_json() or request.form, partial=partial)

def respond_with_schema(self, schema: Schema, value, status: int=200) -> Response:
        result = schema.dump(value)
        if result.errors:
            return self.error('invalid schema supplied')
        return self.respond(result.data, status)

def argmap2schema(argmap):
    class Meta(object):
        strict = True

    attrs = dict(argmap, Meta=Meta)
    CLS = type(str(''), (Schema,), attrs)
    return CLS()

def _register_explicit_schemas():
    for name, obj in inspect.getmembers(sys.modules[__name__]):
        if hasattr(obj, '__name__') and obj.__name__.endswith('Schema') \
                and issubclass(obj, BaseSchema) and obj != BaseSchema:
            if not hasattr(obj.Meta, '__marshmallow__'):
                # FIXME
                if obj != PostEnvironmentSchema:
                    obj.Meta.model.__marshmallow__ = obj


# adapted from
# http://marshmallow-sqlalchemy.readthedocs.io/en/latest/recipes.html#automatically-generating-schemas-for-sqlalchemy-models

def register_schemas(Base):
    """Sets the __marshmallow__ attribute on all model classes.

    Model classes are all the models under the provided declarative Base.
    The __marshmallow__ attribute specifies which schema to use for general serialization of this model
    (esp. in the API).

    Such a schema can be explicitely defined in this module (needs to inherit from ModelSchema and
    have a class name ending in "Schema"). If not, it is inferred from the model.
    """
    _register_explicit_schemas()
    _register_deduced_schemas(Base)()
    # If the mapper configuration is not complete yet, models will not be stored
    # in the declarative Base, hence this trigger:
    sa.event.listen(orm.mapper, 'after_configured', _register_deduced_schemas(Base))

def __init__(self, allow_extra):
        class LocationSchema(Schema):
            latitude = fields.Float(allow_none=True)
            longitude = fields.Float(allow_none=True)

        class SkillSchema(Schema):
            subject = fields.Str(required=True)
            subject_id = fields.Integer(required=True)
            category = fields.Str(required=True)
            qual_level = fields.Str(required=True)
            qual_level_id = fields.Integer(required=True)
            qual_level_ranking = fields.Float(default=0)

        class Model(Schema):
            id = fields.Integer(required=True)
            client_name = fields.Str(validate=validate.Length(max=255), required=True)
            sort_index = fields.Float(required=True)
            #client_email = fields.Email()
            client_phone = fields.Str(validate=validate.Length(max=255), allow_none=True)

            location = LocationSchema()

            contractor = fields.Integer(validate=validate.Range(min=0), allow_none=True)
            upstream_http_referrer = fields.Str(validate=validate.Length(max=1023), allow_none=True)
            grecaptcha_response = fields.Str(validate=validate.Length(min=20, max=1000), required=True)
            last_updated = fields.DateTime(allow_none=True)
            skills = fields.Nested(SkillSchema(many=True))

        self.allow_extra = allow_extra  # unused
        self.schema = Model()

def sanitize_params(self):
        title_class = self.class_name.title().replace('/', '').replace('-', '')
        full_class = '{0}Schema'.format(title_class)
        try:
            schema_class = getattr(Schemas, full_class)()
        except Exception as e:
            raise ChaliceViewError(e)
        sanitized_params = schema_class.load(self.parameters)
        if sanitized_params.errors:
            raise BadRequestError(sanitized_params.errors)
        return sanitized_params.data

def test_by_schema(self):
        ma_schema_cls = self.User.schema.as_marshmallow_schema()
        assert issubclass(ma_schema_cls, marshmallow.Schema)
        assert not issubclass(ma_schema_cls, BaseSchema)

def test_marshmallow_access_custom_attributes(self):

        @self.instance.register
        class Doc(EmbeddedDocument):
            a = fields.IntField()

            attribute_foo = 'foo'

            @property
            def str_prop(self):
                return "I'm a property !"

            @property
            def none_prop(self):
                return None

            @property
            def missing_prop(self):
                return marshmallow.missing

            def func_get_42(self):
                return 42

        class Schema(Doc.schema.as_marshmallow_schema()):
            str_prop = marshmallow.fields.Str(dump_only=True)
            none_prop = marshmallow.fields.Str(allow_none=True, dump_only=True)
            missing_prop = marshmallow.fields.Str(dump_only=True)
            attribute_foo = marshmallow.fields.Str(dump_only=True)
            get_42 = marshmallow.fields.Int(dump_only=True, attribute="func_get_42")

        ret = Schema().dump(Doc(a=1))
        assert not ret.errors
        assert ret.data == {
            'a': 1,
            'str_prop': "I'm a property !",
            'none_prop': None,
            'attribute_foo': 'foo',
            'get_42': 42
        }

def __init__(self, schema, status=200):
        """

        :param schema: `Schema` class or instance
        :param status: associated http status code
        """
        if not isinstance(schema, Schema):
            self.schema = schema()
        else:
            self.schema = schema
        self.status = status