from __future__ import annotations import datetime, hashlib, logging, os, pytz, redis, secrets from typing import Dict, List, Optional, Set, Union from pydantic import BaseModel, EmailStr, Field, Json, PrivateAttr, ValidationError, validator from ..lib_general import * from app.config import settings from .common_field_schema import base_fields, default_num_bytes # ### BEGIN ### API Response Model ### Resp_Body_Base() ### # The pydantic BaseModel to help make consistent REST responses - STI 2021-03-05 class Resp_Body_Base(BaseModel): log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(locals()) data: Union[dict, list] meta: Optional[dict] # ### END ### API Response Model ### Resp_Body_Base() ### # ### BEGIN ### API Response Model ### mk_resp() ### # The make response function for REST - STI 2021-03-17 def mk_resp( data:None|bool|dict|list, dict_to_json:bool=False, status_code:int=200, status_message:str='', status_name:str='', success:bool=True, details:str='', by_alias:bool=True, exclude_unset:bool=True, response=None ): log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(locals()) if data is None: data = { 'result': data } elif data == False: data = { 'result': data } elif data == True: data = { 'result': data } resp_body = {} resp_body['data'] = data resp_body['meta'] = {} resp_body['meta']['details'] = details resp_body['meta']['status_code'] = status_code if status_message: resp_body['meta']['status_message'] = status_message else: resp_body['meta']['status_message'] = settings.HTTP_STATUS_LI[status_code]['message'] resp_body['meta']['status_name'] = settings.HTTP_STATUS_LI[status_code]['name'] resp_body['meta']['success'] = success if isinstance(data, bool): resp_body['meta']['data_type'] = 'bool' elif isinstance(data, int): resp_body['meta']['data_type'] = 'int' elif isinstance(data, str): resp_body['meta']['data_type'] = 'str' elif isinstance(data, dict): resp_body['meta']['data_type'] = 'dict' elif isinstance(data, list): resp_body['meta']['data_type'] = 'list' resp_body['meta']['data_list_count'] = len(data) if response: if status_code == 400: response.status_code = status.HTTP_400_BAD_REQUEST elif status_code == 401: response.status_code = status.HTTP_401_UNAUTHORIZED elif status_code == 403: response.status_code = status.HTTP_403_FORBIDDEN elif status_code == 404: response.status_code = status.HTTP_404_NOT_FOUND elif status_code == 408: response.status_code = status.HTTP_408_REQUEST_TIMEOUT elif status_code == 429: response.status_code = status.HTTP_429_TOO_MANY_REQUESTS elif status_code == 500: response.status_code = status.HTTP_500_INTERNAL_SERVER_ERROR elif status_code == 501: response.status_code = status.HTTP_501_NOT_IMPLEMENTED elif status_code == 502: response.status_code = status.HTTP_502_BAD_GATEWAY elif status_code == 503: response.status_code = status.HTTP_503_SERVICE_UNAVAILABLE elif status_code == 504: response.status_code = status.HTTP_504_GATEWAY_TIMEOUT log.debug(type(resp_body['data'])) resp_body = Resp_Body_Base(**resp_body).dict(by_alias=by_alias, exclude_unset=exclude_unset) return resp_body # ### END ### API Response Model ### mk_resp() ###