LukeParke/connexion
1
0
Fork 0
mirror of https://github.com/LukeHagar/connexion.git synced 2025-12-10 20:37:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

https://github.com/zalando/connexion/issues/32 Reduce code repetition

Browse Source
This commit is contained in:
João Santos
2015-07-13 15:46:34 +02:00
parent 6d5297fbe0
commit f5ef93bd2f
2 changed files with 47 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

98
connexion/decorators/validation.py
View File

@@ -29,57 +29,6 @@ TYPE_MAP = {'integer': int,
'boolean': bool} # map of swagger types to python types
def validate_schema(data, schema) -> flask.Response:
schema_type = schema.get('type')
log_extra = {'url': flask.request.url, 'schema_type': schema_type}
if schema_type == 'array':
if not isinstance(data, list):
actual_type_name = type(data).__name__
logger.error("Wrong data type, expected 'list' got '%s'", actual_type_name, extra=log_extra)
return problem(400, 'Bad Request', "Wrong type, expected 'array' got '{}'".format(actual_type_name))
for item in data:
error = validate_schema(item, schema.get('items'))
if error:
return error
elif schema_type == 'object':
if not isinstance(data, dict):
actual_type_name = type(data).__name__
logger.error("Wrong data type, expected 'dict' got '%s'", actual_type_name, extra=log_extra)
return problem(400, 'Bad Request', "Wrong type, expected 'object' got '{}'".format(actual_type_name))
# verify if required keys are present
required_keys = schema.get('required', [])
logger.debug('... required keys: %s', required_keys)
log_extra['required_keys'] = required_keys
for required_key in schema.get('required', required_keys):
if required_key not in data:
logger.error("Missing parameter '%s'", required_key, extra=log_extra)
return problem(400, 'Bad Request', "Missing parameter '{}'".format(required_key))
# verify if value types are correct
for key in data.keys():
key_properties = schema['properties'].get(key)
if key_properties:
expected_type = TYPE_MAP.get(key_properties['type']) # type: type
if expected_type and not isinstance(data[key], expected_type):
expected_type_name = expected_type.__name__
actual_type_name = type(data[key]).__name__
logger.error("'%s' is not a '%s'", key, expected_type_name)
return problem(400, 'Bad Request',
"Wrong type, expected '{}' got '{}'".format(expected_type_name, actual_type_name))
else:
expected_type = TYPE_MAP.get(schema_type) # type: type
actual_type = type(data) # type: type
if expected_type and not isinstance(data, expected_type):
expected_type_name = expected_type.__name__
actual_type_name = actual_type.__name__
logger.error("'%s' is not a '%s'", data, expected_type_name)
return problem(400, 'Bad Request',
"Wrong type, expected '{}' got '{}'".format(schema_type, actual_type_name))
class RequestBodyValidator:
def __init__(self, schema):
self.schema = schema
@@ -90,7 +39,7 @@ class RequestBodyValidator:
data = flask.request.json
logger.debug("%s validating schema...", flask.request.url)
error = validate_schema(data, self.schema)
error = self.validate_schema(data, self.schema)
if error:
return error
@@ -98,3 +47,48 @@ class RequestBodyValidator:
return response
return wrapper
def validate_schema(self, data, schema) -> flask.Response:
schema_type = schema.get('type')
log_extra = {'url': flask.request.url, 'schema_type': schema_type}
if schema_type == 'array':
if not isinstance(data, list):
actual_type_name = type(data).__name__
logger.error("Wrong data type, expected 'list' got '%s'", actual_type_name, extra=log_extra)
return problem(400, 'Bad Request', "Wrong type, expected 'array' got '{}'".format(actual_type_name))
for item in data:
error = self.validate_schema(item, schema.get('items'))
if error:
return error
elif schema_type == 'object':
if not isinstance(data, dict):
actual_type_name = type(data).__name__
logger.error("Wrong data type, expected 'dict' got '%s'", actual_type_name, extra=log_extra)
return problem(400, 'Bad Request', "Wrong type, expected 'object' got '{}'".format(actual_type_name))
# verify if required keys are present
required_keys = schema.get('required', [])
logger.debug('... required keys: %s', required_keys)
log_extra['required_keys'] = required_keys
for required_key in schema.get('required', required_keys):
if required_key not in data:
logger.error("Missing parameter '%s'", required_key, extra=log_extra)
return problem(400, 'Bad Request', "Missing parameter '{}'".format(required_key))
# verify if value types are correct
for key in data.keys():
key_properties = schema['properties'].get(key)
if key_properties:
error = self.validate_schema(data[key], key_properties)
if error:
return error
else:
expected_type = TYPE_MAP.get(schema_type) # type: type
actual_type = type(data) # type: type
if expected_type and not isinstance(data, expected_type):
expected_type_name = expected_type.__name__
actual_type_name = actual_type.__name__
logger.error("'%s' is not a '%s'", data, expected_type_name)
return problem(400, 'Bad Request',
"Wrong type, expected '{}' got '{}'".format(schema_type, actual_type_name))

2
tests/test_app.py
View File

@@ -247,7 +247,7 @@ def test_schema(app):
assert bad_type.content_type == 'application/problem+json'
bad_type_response = json.loads(bad_type.data.decode()) # type: dict
assert bad_type_response['title'] == 'Bad Request'
assert bad_type_response['detail'] == "Wrong type, expected 'str' got 'int'"
assert bad_type_response['detail'] == "Wrong type, expected 'string' got 'int'"
good_request = app_client.post('/v1.0/test_schema', headers=headers,
data=json.dumps({'image_version': 'version'})) # type: flask.Response