def inline(self, field, context):
# type: (fields.Field, JitContext) -> Optional[str]
"""Generates a template for inlining string serialization.
For example, generates "unicode(value) if value is not None else None"
to serialize a string in Python 2.7
"""
if is_overridden(field._serialize, fields.String._serialize):
return None
result = text_type.__name__ + '({0})'
result += ' if {0} is not None else None'
if not context.is_serializing:
string_type_strings = ','.join([x.__name__ for x in string_types])
result = ('(' + result + ') if '
'(isinstance({0}, (' + string_type_strings +
')) or {0} is None) else dict()["error"]')
return result
python类String()的实例源码
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_missing_field_validation():
class Model(DynaModel):
class Table:
name = 'table'
hash_key = 'foo'
read = 1
write = 1
class Schema:
foo = String(required=True)
baz = String(required=True)
model = Model(foo='foo', partial=True)
with pytest.raises(ValidationError):
model.validate()
try:
model.validate()
except ValidationError as exc:
assert str(exc).startswith("Validation failed for schema ModelSchema. Errors: {'baz'")
def test_model_prepared():
def receiver(model):
receiver.calls.append(model)
receiver.calls = []
model_prepared.connect(receiver)
assert len(receiver.calls) == 0
class SillyModel(DynaModel):
class Table:
name = 'silly'
hash_key = 'silly'
read = 1
write = 1
class Schema:
silly = String(required=True)
assert receiver.calls == [SillyModel]
def test_make_instance():
"""Ensure that the schema's make instance works properly."""
peewee = pytest.importorskip('peewee')
class User(peewee.Model):
name = peewee.CharField(max_length=255)
class UserSchema(Schema):
name = fields.String()
class Meta:
model = User
data = {'name': 'Bob Blah'}
user = UserSchema.make_instance(data)
assert isinstance(user, User)
assert user.name == data['name']
def schema_class(self):
class Schema(marshmallow.Schema):
uuid_field = fields.UUID(required=True)
string_field = fields.String(required=False)
return Schema
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 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 schema():
class TestSchema(Schema):
key = fields.String(default='world')
value = fields.Integer(missing=42)
return TestSchema()
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 get_max_length_error_message(field_name, max_chars):
"""
Returns default max_length message
:param field_name: String
:type field_name: str
:param max_chars: maximum number of chars for the field
:type max_chars: int
:return:
"""
err_msg = 'El campo "{}" no puede contener mas ' \
'de {} caracteres'.format(field_name, str(max_chars))
return err_msg
def TestModelTwo():
"""Provides a test model without a range key"""
if 'marshmallow' in (os.getenv('SERIALIZATION_PKG') or ''):
from marshmallow import fields
class TestModelTwo(DynaModel):
class Table:
name = 'peanut-butter'
hash_key = 'foo'
read = 5
write = 5
class Schema:
foo = fields.String(required=True)
bar = fields.String()
baz = fields.String()
else:
from schematics import types
class TestModelTwo(DynaModel):
class Table:
name = 'peanut-butter'
hash_key = 'foo'
read = 5
write = 5
class Schema:
foo = types.StringType(required=True)
bar = types.StringType()
baz = types.StringType()
return TestModelTwo
def test_parent_inner_classes():
class Parent(DynaModel):
class Table:
name = 'table'
hash_key = 'foo'
read = 1
write = 1
class Schema:
foo = String(required=True)
class Child(Parent):
pass
assert Child.Table is Parent.Table
def test_table_validation():
"""Defining a model with missing table attributes should raise exceptions"""
with pytest.raises(MissingTableAttribute):
class Model(DynaModel):
class Table:
name = 'table'
class Schema:
foo = String(required=True)
def test_table_create_validation():
"""You cannot create a table that is missing read/write attrs"""
with pytest.raises(MissingTableAttribute):
class Model(DynaModel):
class Table:
name = 'table'
hash_key = 'foo'
read = 5
class Schema:
foo = String(required=True)
Model.Table.create_table()
with pytest.raises(MissingTableAttribute):
class Model(DynaModel):
class Table:
name = 'table'
hash_key = 'foo'
write = 5
class Schema:
foo = String(required=True)
Model.Table.create_table()
with pytest.raises(MissingTableAttribute):
class Model(DynaModel):
class Table:
name = 'table'
hash_key = 'foo'
class Schema:
foo = String(required=True)
Model.Table.create_table()
def test_invalid_hash_key():
"""Defining a model where ``hash_key`` in Table points to an invalid field should raise InvalidSchemaField"""
with pytest.raises(InvalidSchemaField):
class Model(DynaModel):
class Table:
name = 'table'
hash_key = 'foo'
read = 1
write = 1
class Schema:
bar = String(required=True)
def test_index_setup():
"""Ensure our index objects are setup & transformed correctly by our meta class"""
class Model(DynaModel):
class Table:
name = 'table'
hash_key = 'foo'
range_key = 'bar'
read = 1
write = 1
class Index(GlobalIndex):
name = 'test-idx'
hash_key = 'foo'
range_key = 'bar'
projection = ProjectAll()
class Schema:
foo = String(required=True)
bar = String(required=True)
model = Model(foo='hi', bar='there')
assert 'test-idx' in model.Table.indexes
assert model.Index.index is model.Table.indexes['test-idx']
assert model.Index.index.table is model.Table
assert model.Index.index.schema is model.Schema
# this gets automatically set during initialization, since read is an optional parameter
assert model.Index.index.read is None
def test_sparse_indexes(dynamo_local):
class MyModel(DynaModel):
class Table:
name = 'mymodel'
hash_key = 'foo'
read = 10
write = 10
class Index1(GlobalIndex):
name = 'index1'
hash_key = 'bar'
read = 10
write = 10
projection = ProjectInclude('foo', 'bar')
class Schema:
foo = String(required=True)
bar = String(required=True)
baz = String(required=True)
bbq = String(required=True)
MyModel.Table.create_table()
MyModel.put_batch(
{'foo': '1', 'bar': '1', 'baz': '1', 'bbq': '1'},
{'foo': '2', 'bar': '2', 'baz': '2', 'bbq': '2'},
)
items = list(MyModel.Index1.query(bar='2'))
assert len(items) == 1
assert items[0].foo == '2'
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 auth_login(email: fields.Email(), password: fields.String()):
user = UserService.instance().login(email, password)
if not user:
raise HTTPBadRequest("login", "failed")
token_data = dict(id=user.id)
return {"token": token_generate(token_data)}
def convert_PrimaryKeyField(self, field, required=False, **params):
dump_only = self.opts.dump_only_pk
return fields.String(dump_only=dump_only, required=False, **params)
def convert_CharField(self, field, validate=None, **params):
validate = ma_validate.Length(max=field.max_length)
return fields.String(validate=validate, **params)
