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
python类Integer()的实例源码
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()