"""Module for requests validation.""" import json from functools import wraps from typing import Any, Callable, Mapping from flask import Request, Response from jsonschema import ValidationError, validate from assets.constants import error as error_const, field_const from assets.validation.schema import header def validate_headers( request: Request, required: bool = True ) -> Callable[[Callable[..., Response]], Callable[..., Response]]: """A decorator to call when validating HTTP request headers. Args: request (Request): request object from the handler. required (bool) Returns: callable: the wrapped function. """ def wrap(function: Callable[..., Response]) -> Callable[..., Response]: @wraps(function) def wrapped_f(*args: Any, **kwargs: Any) -> Response: if required or ( not required and field_const.ORCHARD_USER_ID in request.headers ): try: validate(dict(request.headers), header.user_headers_schema) except ValidationError as error: return Response( response=json.dumps( { "code": error_const.ERROR_CODE_HEADER_VALIDATION, "message": str(error), } ), status=400, ) return function(*args, **kwargs) return wrapped_f return wrap def validate_body( request: Request, schema: Mapping[str, Any] ) -> Callable[[Callable[..., Response]], Callable[..., Response]]: """A decorator to call when validating HTTP request body. Args: request (Request): Request object from the handler. schema (Mapping): The JSON Draft4 Schema to validate against. Returns: callable: the wrapped function. """ def wrap(function: Callable[..., Response]) -> Callable[..., Response]: @wraps(function) def wrapped_f(*args: Any, **kwargs: Any) -> Response: try: request_body = request.get_json(silent=True) or {} validate(request_body, schema) except ValidationError as error: return Response( response=json.dumps( { "code": error_const.ERROR_CODE_BODY_VALIDATION, "message": str(error), } ), status=400, ) return function(*args, **kwargs) return wrapped_f return wrap def validate_query( request: Request, schema: Mapping[str, Any] ) -> Callable[[Callable[..., Response]], Callable[..., Response]]: """A decorator to call when validating query arguments. Args: request (Request): Request object from the handler. schema (Mapping): The JSON Draft4 Schema to validate against. Returns: callable: the wrapped function. """ def wrap(function: Callable[..., Response]) -> Callable[..., Response]: @wraps(function) def wrapped_f(*args: Any, **kwargs: Any) -> Response: try: source_query = dict(request.args) query_dict = dict(source_query.items()) validate(query_dict, schema) except ValidationError as error: return Response( response=json.dumps( { "code": error_const.ERROR_CODE_QUERY_VALIDATION, "message": str(error), } ), status=400, ) return function(*args, **kwargs) return wrapped_f return wrap