"""Module for requests validation.""" from functools import wraps import json import os import jsonschema from oto import response from oto.adaptors.flask import flaskify as oto_flaskify from availability import config from availability.constants import error def validate_headers(request, schema): """Validate HTTP request headers. Args: request (werkzeug.local.LocalProxy): request object from the handler. schema (Draft4Validator): the JSON Draft4 Schema to validate against. Returns: callable: the wrapped function. """ def wrap(function): @wraps(function) def wrapped_f(*args, **kwargs): try: jsonschema.validate(dict(request.headers), schema) except jsonschema.ValidationError as err: return oto_flaskify( response.create_error_response( code=error.HEADER_VALIDATION_ERROR, message=str(err))) return function(*args, **kwargs) return wrapped_f return wrap def validate_body(request, schema): """Validate HTTP request body. Args: request (werkzeug.local.LocalProxy): request object from the handler. schema (Draft4Validator): the JSON Draft4 Schema to validate against. Returns: callable: the wrapped function. """ def wrap(function): @wraps(function) def wrapped_f(*args, **kwargs): try: jsonschema.validate( request.get_json(), schema, format_checker=jsonschema.draft4_format_checker) except jsonschema.ValidationError as err: return oto_flaskify( response.create_error_response( code=error.BODY_VALIDATION_ERROR, message=str(err))) return function(*args, **kwargs) return wrapped_f return wrap def load_schema(name): """Load schema from json file. Args: name (str): name of file with schema. Returns: dict: JSON schema from given file """ schema_path = os.path.join( config.BASE_DIR, '..', 'spec', name) with open(schema_path) as schema_file: return json.load(schema_file)