def dump_schema(schema_obj):
json_schema = {
"type": "object",
"properties": {},
"required": [],
}
mapping = {v: k for k, v in schema_obj.TYPE_MAPPING.items()}
mapping[fields.Email] = text_type
mapping[fields.Dict] = dict
mapping[fields.List] = list
mapping[fields.Url] = text_type
mapping[fields.LocalDateTime] = datetime.datetime
for field_name, field in sorted(schema_obj.fields.items()):
schema = None
if field.__class__ in mapping:
pytype = mapping[field.__class__]
schema = _from_python_type(field, pytype)
elif isinstance(field, fields.Nested):
schema = _from_nested_schema(field)
elif issubclass(field.__class__, fields.Field):
for cls in mapping.keys():
if issubclass(field.__class__, cls):
pytype = mapping[cls]
schema = _from_python_type(field, pytype)
break
if schema is None:
raise ValueError('unsupported field type %s' % field)
field_name = field.dump_to or field.name
json_schema['properties'][field_name] = schema
if field.required:
json_schema['required'].append(field.name)
return json_schema
python类Email()的实例源码
def __repr__(self):
return 'ID: {} Email: {}'.format(self.id, self.email)
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 __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 error_handler(cls, func):
"""Decorator that registers an error handler function for the schema.
The function receives the :class:`Schema` instance, a dictionary of errors,
and the serialized object (if serializing data) or data dictionary (if
deserializing data) as arguments.
Example: ::
class UserSchema(Schema):
email = fields.Email()
@UserSchema.error_handler
def handle_errors(schema, errors, obj):
raise ValueError('An error occurred while marshalling {}'.format(obj))
user = User(email='invalid')
UserSchema().dump(user) # => raises ValueError
UserSchema().load({'email': 'bademail'}) # raises ValueError
.. versionadded:: 0.7.0
.. deprecated:: 2.0.0
Set the ``error_handler`` class Meta option instead.
"""
warnings.warn(
'Schema.error_handler is deprecated. Set the error_handler class Meta option '
'instead.', category=DeprecationWarning
)
cls.__error_handler__ = func
return func
