113 lines
4.5 KiB
Python
113 lines
4.5 KiB
Python
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 app.db_sql import redis_lookup_id_random
|
|
from app.lib_general import log, logging, Response, status
|
|
|
|
from app.config import settings
|
|
|
|
from app.models.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())
|
|
|
|
# test_prop: Optional[str] = Field(
|
|
# alias = 'test_prop_alias'
|
|
# )
|
|
|
|
data: Union[list, dict]
|
|
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: Response = None
|
|
):
|
|
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
|
log.debug(locals())
|
|
|
|
if data is None: data_out = { 'result': data }
|
|
elif data == False: data_out = { 'result': data }
|
|
elif data == True: data_out = { 'result': data }
|
|
elif isinstance(data, dict):
|
|
log.info('Data type is a dict')
|
|
data_out = data
|
|
elif isinstance(data, list):
|
|
log.info('Data type is a list')
|
|
data_out = data
|
|
else: # Assuming it is still and object. This should be improved. Example model type: "<class 'app.models.account_models.Account_Base'>"
|
|
log.info('Data type is other')
|
|
data_out = data.dict(by_alias=by_alias, exclude_unset=exclude_unset)
|
|
# log.debug(data_out)
|
|
|
|
resp_body = {}
|
|
resp_body['data'] = data_out
|
|
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:
|
|
log.debug(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.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
|
# log.debug(resp_body)
|
|
# log.debug(type(resp_body['data']))
|
|
|
|
# import json
|
|
# with open('data.txt', 'w') as outfile:
|
|
# json.dump(resp_body['data'], outfile)
|
|
|
|
resp_body_obj = Resp_Body_Base(**resp_body)
|
|
log.debug(resp_body_obj)
|
|
resp_body_dict = resp_body_obj.dict(by_alias=by_alias, exclude_unset=exclude_unset)
|
|
log.debug(resp_body_dict)
|
|
|
|
return resp_body_dict
|
|
# ### END ### API Response Model ### mk_resp() ### |