"""Handler Utils. These are general purpose functions for use with all route handlers. """ from datetime import date from datetime import datetime from functools import wraps import json from flask import request from deliveryhistory import response from deliveryhistory.validation import json_schema def validate_header(validator): """Validate HTTP request headers decorator. Args: request (werkzeug.local.LocalProxy): the request object from the handler. validator (Draft3Validator): the JSON Draft3 Schema to validate against. Returns: callable: the wrapped function. """ def decorator(f): @wraps(f) def _validate_header(*args, **kwargs): """Validate the request headers against a validator. Args: request (werkzeug.local.LocalProxy): the request object from the handler. validator (Draft3Validator): the JSON Draft3 Schema to validate against. Returns: Response: error HTTP response, or continue to the calling function if successful. """ validation_response = json_schema.validate( dict(request.headers), validator) if validation_response.status != 200: return response.flaskify(validation_response) return f(*args, **kwargs) return _validate_header return decorator def validate_body(validator): """Validate HTTP request body decorator. Args: request (werkzeug.local.LocalProxy): the request object from the handler. validator (Draft3Validator): the JSON Draft3 Schema to validate against. Returns: callable: the wrapped function. """ def decorator(f): @wraps(f) def _validate_body(*args, **kwargs): """Validate the request body against a validator. Args: request (werkzeug.local.LocalProxy): the request object from the handler. validator (Draft3Validator): the JSON Draft3 Schema to validate against. Returns: Response: error HTTP response, or continue to the calling function if successful. """ validation_response = json_schema.validate( request.get_json(), validator) if validation_response.status != 200: return response.flaskify(validation_response) return f(*args, **kwargs) return _validate_body return decorator def validate_request_args(validator): """Validate HTTP request args decorator. Args: request (werkzeug.local.LocalProxy): the request object from the handler. validator (Draft3Validator): the JSON Draft3 Schema to validate against. Returns: callable: the wrapped function. """ def decorator(f): @wraps(f) def _validate_request_args(*args, **kwargs): """Validate the request body against a validator. Args: request (werkzeug.local.LocalProxy): the request object from the handler. validator (Draft3Validator): the JSON Draft3 Schema to validate against. Returns: Response: error HTTP response, or continue to the calling function if successful. """ validation_response = json_schema.validate( request.args.to_dict(), validator) if validation_response.status != 200: return response.flaskify(validation_response) return f(*args, **kwargs) return _validate_request_args return decorator def add_pagination(records, page_offset=0, page_limit=1000): """Wrap pagination information around records. Args: records (list): list of report dictionaries. page_offset (int): page offset. page_limit (int): total number of records per page. Return: dict: dictionary containing a list of records and pagination information. """ return { 'items': records[ page_offset * page_limit: page_offset * page_limit + page_limit ], 'pagination': { 'type': 'standard', 'offset': page_offset, 'limit': page_limit, 'total_records': len(records) } } class DatetimeEncoder(json.JSONEncoder): """JSON encoder with correct date object serialization.""" DEFAULT_DATETIME_FORMAT = '%Y-%m-%d %H:%M:%S.%f' DEFAULT_DATE_FORMAT = '%Y-%m-%d' def __init__( self, datetime_format=DEFAULT_DATETIME_FORMAT, date_format=DEFAULT_DATE_FORMAT, **kwargs): """Create new DatetimeEncoder instance. Args: datetime_format (str): datetime format string date_format (str): date format string """ super().__init__(**kwargs) self.datetime_format = datetime_format self.date_format = date_format def default(self, obj): """Serialize JSON object to string. Args: obj: JSON object Returns: str: serialized JSON object """ if isinstance(obj, datetime): return obj.strftime(self.datetime_format) elif isinstance(obj, date): return obj.strftime(self.date_format) # Let the base class default method raise the TypeError return json.JSONEncoder.default(self, obj)