"""JSON Draft3 Schema Validator Wrapper.""" from functools import wraps from flask import request from oto import response from oto.adaptors import flask as oto_flask from product.constants import error as error_constants from product.validation.schema import json_validators def _filter(data): """Sanitize a dictionary of request input data (e.g. headers or body). This is done before applying JSON schema validations to the data. Sanitization includes: 1.) Stripping leading or trailing space from strings 2.) Replacing empty strings with None Args: data (dict): the data to be sanitized Returns: dict: a sanitized version of the given data """ for key, value in data.items(): if isinstance(value, str): data[key] = value.strip() if data[key] == '': data[key] = None return data def validate(data, validator, failure_status=400): """Generic JSON Draft3 Schema Vaidation. Args: data (dict): the JSON to be validated. schema (Draft3Validator): the JSON Draft3 Schema to validate against Returns: Response: the result of the validation. """ errors = {} data = _filter(data) for error in sorted(validator.iter_errors(data), key=str): if error.relative_path: errors[error.relative_path.pop()] = error.message elif error.absolute_schema_path: errors[error.absolute_schema_path.pop()] = error.message if errors: return response.create_error_response( code=error_constants.ERROR_CODE_BAD_REQUEST, message=errors, status=failure_status) return response.Response(message={'status': 'ok'}, status=200) def validate_request_args(resource, method): """Validate the query args in the given request against the RAML spec. Args: resource (str): name of the API resource in the RAML spec. method (str): name of the HTTP method in the RAML spec. Returns: Response: the result of the validation. """ return validate( request.args.to_dict(), json_validators.query_args_validator(resource, method)) def validate_request_body(resource, method): """Validate the JSON payload of the given request against the RAML spec. The method will generally be "post" or "put". Args: resource (str): name of the API resource in the RAML spec. method (str): name of the HTTP method in the RAML spec. Returns: Response: the result of the validation. """ return validate( request.get_json(), json_validators.body_validator(resource, method)) def validate_request_headers(resource, method): """Validate the headers of the given request against the RAML spec. Args: resource (str): name of the API resource in the RAML spec. method (str): name of the HTTP method in the RAML spec. Returns: Response: the result of the validation. """ return validate( dict(request.headers), json_validators.headers_validator(resource, method)) def wrap_request_validation(resource, method, request_validation): """Decorate a route handler with the given validation function. If validation fails, the route will return an error response and the handler will not be executed. If validation succeeds, the hander will be executed. Sample usage: @app.route('/resource', methods=['POST']) @json_schema.wrap_request_validation( request, '/resource', 'post', json_schema.validate_request_headers) @json_schema.wrap_request_validation( request, '/resource', 'post', json_schema.validate_request_body) def handle_resource_request(): # handler code here... Args: resource (str): name of the API resource in the RAML spec. method (str): name of the HTTP method in the RAML spec. request_validation (function): the validation function to execute. Returns: function: decorator that wraps the given route handler with validation. """ def decorator(function): @wraps(function) def validate_request(*args, **kwargs): validation_response = request_validation(resource, method) if validation_response.status != 200: return oto_flask.flaskify(validation_response) return function(*args, **kwargs) return validate_request return decorator