Python marshmallow.fields 模块，Integer() 实例源码

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_number_hash_key(dynamo_local, request):
    """Test a number hash key and ensure the dynamo type gets set correctly"""
    class Model(DynaModel):
        class Table:
            name = 'table'
            hash_key = 'foo'
            read = 1
            write = 1

        class Schema:
            foo = Number(required=True)
            baz = String(required=True)

    Model.Table.create()
    request.addfinalizer(Model.Table.delete)

    model = Model(foo=1, baz='foo')
    assert model.Table.attribute_definitions == [{'AttributeName': 'foo', 'AttributeType': 'N'}]

    model.save()

def test_schema_parents_mro():
    """Inner Schema classes should obey MRO (to test our schematics field pull up)"""
    class MixinTwo(object):
        bar = Number()

    class MixinOne(object):
        bar = String()

    class Model(DynaModel):
        class Table:
            name = 'table'
            hash_key = 'foo'
            read = 1
            write = 1

        class Schema(MixinOne, MixinTwo):
            foo = Number(required=True)
            baz = String(required=True)

    assert 'bar' in Model.Schema.dynamorm_fields()
    assert isinstance(Model.Schema.dynamorm_fields()['bar'], String)

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 simple_schema():
    class InstanceSchema(Schema):
        key = fields.String()
        value = fields.Integer(default=0)
    return InstanceSchema()

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_instance_serializer():
    serializer = InstanceSerializer()
    field = fields.Integer()
    assert 'result["foo"] = obj.foo' == str(serializer.serialize(
        'foo', 'bar', 'result["foo"] = {0}', field))

def test_dict_serializer_with_default():
    serializer = DictSerializer()
    field = fields.Integer(default=3)
    result = str(serializer.serialize('foo', 'bar', 'result["foo"] = {0}',
                                      field))
    assert 'result["foo"] = obj.get("foo", bar__default)' == result

def test_dict_serializer_no_default():
    serializer = DictSerializer()
    field = fields.Integer()
    result = str(serializer.serialize('foo', 'bar', 'result["foo"] = {0}',
                                      field))
    expected = ('if "foo" in obj:\n'
                '    result["foo"] = obj["foo"]')
    assert expected == result

def test_hybrid_serializer():
    serializer = HybridSerializer()
    field = fields.Integer()
    result = str(serializer.serialize('foo', 'bar', 'result["foo"] = {0}',
                                      field))
    expected = ('try:\n'
                '    value = obj["foo"]\n'
                'except (KeyError, AttributeError, IndexError, TypeError):\n'
                '    value = obj.foo\n'
                'result["foo"] = value')
    assert expected == result

def test_generate_marshall_method_bodies():
    class OneFieldSchema(Schema):
        foo = fields.Integer()
    context = JitContext()
    result = generate_method_bodies(OneFieldSchema(), context)
    expected = '''\
def InstanceSerializer(obj):
    res = {}
    value = obj.foo; value = value() if callable(value) else value; \
value = int(value); res["foo"] = value
    return res
def DictSerializer(obj):
    res = {}
    if "foo" in obj:
        value = obj["foo"]; value = value() if callable(value) else value; \
value = int(value); res["foo"] = value
    return res
def HybridSerializer(obj):
    res = {}
    try:
        value = obj["foo"]
    except (KeyError, AttributeError, IndexError, TypeError):
        value = obj.foo
    value = value; value = value() if callable(value) else value; \
value = int(value); res["foo"] = value
    return res'''
    assert expected == result

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 schema():
    class TestSchema(Schema):
        key = fields.String(default='world')
        value = fields.Integer(missing=42)
    return TestSchema()

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 test_explicit_schema_parents():
    """Inner Schema classes should be able to have explicit parents"""
    class SuperMixin(object):
        bbq = String()

    if 'marshmallow' in (os.getenv('SERIALIZATION_PKG') or ''):
        class Mixin(SuperMixin):
            is_mixin = True
            bar = String()

            @validates('bar')
            def validate_bar(self, value):
                if value != 'bar':
                    raise SchemaValidationError('bar must be bar')
    else:
        class Mixin(SuperMixin):
            is_mixin = True
            bar = String()

            def validate_bar(self, data, value):
                if value != 'bar':
                    raise SchemaValidationError('bar must be bar')

    class Model(DynaModel):
        class Table:
            name = 'table'
            hash_key = 'foo'
            read = 1
            write = 1

        class Schema(Mixin):
            foo = Number(required=True)
            baz = String(required=True)

    assert Model.Schema.is_mixin is True
    assert list(sorted(Model.Schema.dynamorm_fields().keys())) == ['bar', 'baz', 'bbq', 'foo']

    with pytest.raises(ValidationError):
        Model(foo='foo', baz='baz', bar='not bar')

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()