Files
OSIT-AE-API-FastAPI/app/routers/cont_edu_cert_person.py
2021-10-01 17:15:31 -04:00

473 lines
20 KiB
Python

import datetime
from fastapi import APIRouter, Body, Depends, Header, HTTPException, Query, Response, status
from pydantic import BaseModel, EmailStr, Field
from typing import Dict, List, Optional, Set, Union
from app.lib_general import log, logging
from app.config import settings
from app.db_sql import sql_insert, sql_update, sql_insert_or_update, sql_select, sql_delete, redis_lookup_id_random
from app.routers.api_crud import delete_obj_template, get_obj_template, get_obj_li_template, patch_obj_template, post_obj_template
from app.methods.cont_edu_cert_person_methods import get_cont_edu_cert_person_rec_list, load_cont_edu_cert_person_obj
from app.models.cont_edu_cert_person_models import Cont_Edu_Cert_Person_Base
from app.models.response_models import *
router = APIRouter()
@router.post('/cont_edu/cert/person', response_model=Resp_Body_Base)
async def post_cont_edu_cert_person_obj(
obj: Cont_Edu_Cert_Person_Base,
x_account_id: str = Header(...),
return_obj: Optional[bool] = True,
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
response: Response = Response,
):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
obj_type = 'cont_edu_cert_person'
obj_data_dict = obj.dict(by_alias=False, exclude_unset=True)
result = post_obj_template(
obj_type=obj_type,
data=obj_data_dict,
return_obj=True,
by_alias=True,
exclude_unset=True,
)
return result
@router.patch('/cont_edu/cert/person/{obj_id}', response_model=Resp_Body_Base)
async def patch_cont_edu_cert_person_obj(
obj: Cont_Edu_Cert_Person_Base,
obj_id: str = Query(..., min_length=1, max_length=22),
x_account_id: Optional[str] = Header(..., ),
return_obj: Optional[bool] = True,
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
response: Response = Response,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
obj_type = 'cont_edu_cert_person'
obj_data_dict = obj.dict(by_alias=False, exclude_unset=True)
obj_data_dict['id'] = redis_lookup_id_random(record_id_random=obj_id, table_name=obj_type)
obj_data_dict['id_random'] = obj_id
result = patch_obj_template(
obj_type=obj_type,
data=obj_data_dict,
obj_id=obj_id,
return_obj=True,
by_alias=True,
exclude_unset=True,
)
return result
# ### BEGIN ### API Cont Edu Cert Person ### patch_cont_edu_cert_person_json() ###
@router.patch('/cont_edu/cert/person/{cont_edu_cert_person_id}/json', response_model=Resp_Body_Base)
async def patch_cont_edu_cert_person_json(
cont_edu_cert_person_obj: Cont_Edu_Cert_Person_Base,
cont_edu_cert_person_id: str = Query(..., min_length=1, max_length=22),
create_sub_obj: bool = False,
x_account_id: Optional[str] = Header(..., ),
return_obj: Optional[bool] = True,
by_alias: Optional[bool] = True,
include: Optional[list] = [],
exclude: Optional[list] = [],
exclude_unset: Optional[bool] = True,
exclude_none: Optional[bool] = True,
response: Response = Response,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if cont_edu_cert_person_id := redis_lookup_id_random(record_id_random=cont_edu_cert_person_id, table_name='cont_edu_cert_person'): pass
else:
return mk_resp(data=None, status_code=404, response=response)
if cont_edu_cert_person_obj_up_result := update_cont_edu_cert_person_obj(
cont_edu_cert_person_id=cont_edu_cert_person_id,
cont_edu_cert_person_obj_up=cont_edu_cert_person_obj,
create_sub_obj=create_sub_obj,
):
log.debug(cont_edu_cert_person_obj_up_result)
if return_obj:
cont_edu_cert_person_obj = load_cont_edu_cert_person_obj(cont_edu_cert_person_id=cont_edu_cert_person_id)
cont_edu_cert_person_dict = cont_edu_cert_person_obj.dict(by_alias=by_alias, exclude_unset=exclude_unset)
return mk_resp(data=cont_edu_cert_person_dict, response=response)
else:
return mk_resp(data=cont_edu_cert_person_obj_up_result, response=response)
else:
return mk_resp(data=False, status_code=400, response=response) # Bad Request
# ### END ### API Cont Edu Cert Person ### patch_cont_edu_cert_person_json() ###
@router.get('/cont_edu/cert/person/list', response_model=Resp_Body_Base)
async def get_cont_edu_cert_person_obj_li(
for_obj_type: str = Query(None, min_length=2, max_length=50),
for_obj_id: str = Query(None, min_length=1, max_length=22),
group: str = Query(None, min_length=2, max_length=50),
x_account_id: str = Header(...),
by_alias: bool = True,
exclude_unset: bool = True,
response: Response = Response,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
obj_type = 'cont_edu_cert_person'
if for_obj_type == 'event_exhibit' and for_obj_id:
#base_name = obj_type_li[obj_type]['base_name']
base_name = Cont_Edu_Cert_Person_Base
for_obj_id_random = for_obj_id
for_obj_id = redis_lookup_id_random(record_id_random=for_obj_id_random, table_name=for_obj_type)
data = {}
data['for_obj_type'] = for_obj_type
data['for_obj_id'] = for_obj_id
data['for_obj_id_random'] = for_obj_id_random
data['group'] = group
if group:
sql = """
SELECT *
FROM `cont_edu_cert_person` AS cont_edu_cert_person
WHERE cont_edu_cert_person.for_type = :for_obj_type AND cont_edu_cert_person.for_id = :for_obj_id
AND cont_edu_cert_person.group = :group
"""
else:
sql = """
SELECT *
FROM `cont_edu_cert_person` AS cont_edu_cert_person
WHERE cont_edu_cert_person.for_type = :for_obj_type AND cont_edu_cert_person.for_id = :for_obj_id
"""
if sql_result := sql_select(data=data, sql=sql, as_list=True):
resp_data_li = []
for record in sql_result:
resp_data = base_name(**record).dict(by_alias=by_alias, exclude_unset=exclude_unset)
resp_data_li.append(resp_data)
return mk_resp(data=resp_data_li, response=response)
else:
log.debug(sql_result)
return mk_resp(data=False, status_code=404, response=response)
result = get_obj_li_template(
obj_type=obj_type,
for_obj_type=for_obj_type,
for_obj_id=for_obj_id,
by_alias=True,
exclude_unset=True,
)
return result
# ### BEGIN ### API Cont Edu Cert Person ### search_cont_edu_cert_person_obj_li() ###
# Updated 2021-07-28
@router.get('/cont_edu/cert/person/search', response_model=Resp_Body_Base)
async def search_cont_edu_cert_person_obj_li(
account_id: str = Query(None, min_length=11, max_length=22),
cont_edu_cert_id_list: List[str] = Query(['']),
external_id: str = Query('%', max_length=50),
given_name: str = Query('%', max_length=25),
family_name: str = Query('%', max_length=25),
email: str = Query('%', max_length=50),
from_datetime: Optional[datetime.datetime] = None, # based on created_on
to_datetime: Optional[datetime.datetime] = None, # based on created_on
limit: int = 50,
enabled: str = 'enabled',
x_account_id: str = Header(...),
by_alias: bool = True,
exclude_unset: bool = True,
response: Response = Response,
):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
log.debug(cont_edu_cert_id_list)
print('---- XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX ----')
print(cont_edu_cert_id_list)
print('---- XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX ----')
if cont_edu_cert_id_list:
# Do something to change from id_random to id
cont_edu_cert_id_list_new = []
for cont_edu_cert_id in cont_edu_cert_id_list:
cont_edu_cert_id = redis_lookup_id_random(record_id_random=cont_edu_cert_id, table_name='cont_edu_cert')
cont_edu_cert_id_list_new.append(cont_edu_cert_id)
cont_edu_cert_id_list = cont_edu_cert_id_list_new
log.debug(cont_edu_cert_id_list)
# cont_edu_cert_id_list = [1, 2]
# cont_edu_cert_id_list = [3, 4]
# cont_edu_cert_id_list = ['3vqsbpWjd9s', 'bCv8WBzdRJQ']
data = {}
data['account_id_random'] = account_id
data['cont_edu_cert_id_list'] = cont_edu_cert_id_list
# data['external_id'] = external_id
# data['given_name'] = given_name
# data['family_name'] = family_name
data['email'] = '%'+email+'%' # Adding the % symbol here because it turns certain combinations into special characters
# data['from_datetime'] = from_datetime
# data['to_datetime'] = to_datetime
log.debug(data)
if enabled in ['enabled', 'disabled', 'all']:
if enabled == 'enabled':
data['enable'] = True
sql_enabled = f'AND `cont_edu_cert_person`.enable = :enable'
elif enabled == 'disabled':
data['enable'] = False
sql_enabled = f'AND `cont_edu_cert_person`.enable = :enable'
elif enabled == 'all':
sql_enabled = ''
if limit:
data['limit'] = limit
sql_limit = f'LIMIT :limit'
else:
sql_limit = ''
sql = f"""
SELECT *
FROM `v_cont_edu_cert_person` AS cont_edu_cert_person
WHERE cont_edu_cert_person.account_id = :account_id
AND cont_edu_cert_person.cont_edu_cert_id IN :cont_edu_cert_id_list
AND (cont_edu_cert_person.external_id LIKE :external_id OR cont_edu_cert_person.external_id IS NULL)
AND (cont_edu_cert_person.given_name LIKE :given_name
AND cont_edu_cert_person.family_name LIKE :family_name)
AND cont_edu_cert_person.email LIKE :email
{sql_enabled}
ORDER BY cont_edu_cert_person.given_name ASC, cont_edu_cert_person.family_name ASC, cont_edu_cert_person.created_on DESC, cont_edu_cert_person.updated_on DESC
{sql_limit};
"""
sql = f"""
SELECT *
FROM `v_cont_edu_cert_person` AS cont_edu_cert_person
WHERE cont_edu_cert_person.account_id = :account_id
AND cont_edu_cert_person.cont_edu_cert_id IN :cont_edu_cert_id_list
AND cont_edu_cert_person.email LIKE :email
{sql_enabled}
ORDER BY cont_edu_cert_person.given_name ASC, cont_edu_cert_person.family_name ASC, cont_edu_cert_person.created_on DESC, cont_edu_cert_person.updated_on DESC
{sql_limit};
"""
log.debug(sql)
if sql_result := sql_select(data=data, sql=sql, as_list=True, rm_id_random=True):
resp_data_li = []
for record in sql_result:
resp_data = Cont_Edu_Cert_Person_Base(**record).dict(by_alias=by_alias, exclude_unset=exclude_unset)
resp_data_li.append(resp_data)
return mk_resp(data=resp_data_li, response=response)
else:
log.debug(sql_result)
return mk_resp(data=False, status_code=404, response=response)
# ### END ### API Cont Edu Cert Person ### search_cont_edu_cert_person_obj_li() ###
# ### BEGIN ### API Cont Edu Cert Person ### get_cont_edu_cert_person_obj() ###
# Updated 2021-08-05
@router.get('/cont_edu/cert/person/{cont_edu_cert_person_id}', response_model=Resp_Body_Base)
async def get_cont_edu_cert_person_obj(
cont_edu_cert_person_id: str = Query(..., min_length=1, max_length=22),
limit: int = 500, # For now this covers any included objects or object lists
enabled: str = 'enabled', # For now this covers any included objects or object lists
inc_cont_edu_cert: bool = False,
x_account_id: str = Header(...),
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
response: Response = Response,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if cont_edu_cert_person_id := redis_lookup_id_random(record_id_random=cont_edu_cert_person_id, table_name='cont_edu_cert_person'): pass
else: return mk_resp(data=None, status_code=404, response=response)
cont_edu_cert_person_obj = load_cont_edu_cert_person_obj(
cont_edu_cert_person_id = cont_edu_cert_person_id,
limit = limit,
by_alias = by_alias,
exclude_unset = exclude_unset,
# model_as_dict = model_as_dict,
enabled = enabled,
inc_cont_edu_cert = inc_cont_edu_cert,
)
cont_edu_cert_person_dict = cont_edu_cert_person_obj.dict(by_alias=by_alias, exclude_unset=exclude_unset)
return mk_resp(data=cont_edu_cert_person_dict, response=response)
# ### END ### API Cont Edu Cert Person ### get_cont_edu_cert_person_obj() ###
# ### BEGIN ### API Cont Edu Cert Person ### get_account_obj_cont_edu_cert_person_list() ###
# Updated 2021-07-28
@router.get('/account/{account_id}/cont_edu/cert/person/list', response_model=Resp_Body_Base)
async def get_account_obj_cont_edu_cert_person_list(
account_id: str = Query(..., min_length=1, max_length=22),
limit: int = 500, # For now this covers any included objects or object lists
enabled: str = 'enabled', # For now this covers any included objects or object lists
inc_cont_edu_cert: bool = False,
x_account_id: str = Header(...),
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
response: Response = Response,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if account_id := redis_lookup_id_random(record_id_random=account_id, table_name='account'): pass
else: return mk_resp(data=None, status_code=404, response=response)
response_data = None
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
# Updated 2021-07-28
if cont_edu_cert_person_rec_list_result := get_cont_edu_cert_person_rec_list(
account_id = account_id,
limit = limit,
enabled = enabled,
):
cont_edu_cert_person_result_list = []
for cont_edu_cert_person_rec in cont_edu_cert_person_rec_list_result:
if load_cont_edu_cert_person_result := load_cont_edu_cert_person_obj(
cont_edu_cert_person_id = cont_edu_cert_person_rec.get('cont_edu_cert_person_id', None),
limit = limit,
by_alias = by_alias,
exclude_unset = exclude_unset,
# model_as_dict = model_as_dict,
enabled = enabled,
inc_cont_edu_cert = inc_cont_edu_cert,
):
cont_edu_cert_person_result_list.append(load_cont_edu_cert_person_result)
else:
cont_edu_cert_person_result_list.append(None)
response_data = cont_edu_cert_person_result_list
else:
return mk_resp(data=False, status_code=400, response=response) # Bad Request
return mk_resp(data=response_data, response=response)
# ### END ### API Cont Edu Cert Person ### get_account_obj_cont_edu_cert_person_list() ###
# ### BEGIN ### API Cont Edu Cert Person ### get_person_obj_cont_edu_cert_person_list() ###
# Updated 2021-07-28
@router.get('/person/{person_id}/cont_edu/cert/person/list', response_model=Resp_Body_Base)
async def get_person_obj_cont_edu_cert_person_list(
person_id: str = Query(..., min_length=11, max_length=22),
limit: int = 500, # For now this covers any included objects or object lists
enabled: str = 'enabled', # For now this covers any included objects or object lists
x_account_id: str = Header(...),
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
response: Response = Response,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if person_id := redis_lookup_id_random(record_id_random=person_id, table_name='person'): pass
else: return mk_resp(data=None, status_code=404, response=response)
# Updated 2021-10-01
if cont_edu_cert_person_rec_list_result := get_cont_edu_cert_person_rec_list(
person_id = person_id,
enabled = enabled,
limit = limit,
):
cont_edu_cert_person_result_list = []
for cont_edu_cert_person_rec in cont_edu_cert_person_rec_list_result:
if load_cont_edu_cert_person_result := load_cont_edu_cert_person_obj(
cont_edu_cert_person_id = cont_edu_cert_person_rec.get('cont_edu_cert_person_id', None),
limit = limit,
by_alias = by_alias,
exclude_unset = exclude_unset,
# model_as_dict = model_as_dict,
enabled = enabled,
):
cont_edu_cert_person_result_list.append(load_cont_edu_cert_person_result)
else:
cont_edu_cert_person_result_list.append(None)
response_data = cont_edu_cert_person_result_list
else:
return mk_resp(data=False, status_code=400, response=response) # Bad Request
return mk_resp(data=response_data, response=response)
# ### END ### API Cont Edu Cert Person ### get_account_obj_cont_edu_cert_person_list() ###
# ### BEGIN ### API Cont Edu Cert Person ### get_cont_edu_cert_obj_cont_edu_cert_person_list() ###
# Updated 2021-07-28
@router.get('/cont_edu/cert/{cont_edu_cert_id}/cont_edu/cert/person/list', response_model=Resp_Body_Base)
async def get_cont_edu_cert_obj_cont_edu_cert_person_list(
cont_edu_cert_id: str = Query(..., min_length=1, max_length=22),
limit: int = 500, # For now this covers any included objects or object lists
enabled: str = 'enabled', # For now this covers any included objects or object lists
inc_cont_edu_cert: bool = False,
x_account_id: str = Header(...),
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
response: Response = Response,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if cont_edu_cert_id := redis_lookup_id_random(record_id_random=cont_edu_cert_id, table_name='cont_edu_cert'): pass
else: return mk_resp(data=None, status_code=404, response=response)
response_data = None
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
# Updated 2021-07-28
if cont_edu_cert_person_rec_list_result := get_cont_edu_cert_person_rec_list(
cont_edu_cert_id = cont_edu_cert_id,
limit = limit,
enabled = enabled,
):
cont_edu_cert_person_result_list = []
for cont_edu_cert_person_rec in cont_edu_cert_person_rec_list_result:
if load_cont_edu_cert_person_result := load_cont_edu_cert_person_obj(
cont_edu_cert_person_id = cont_edu_cert_person_rec.get('cont_edu_cert_person_id', None),
limit = limit,
by_alias = by_alias,
exclude_unset = exclude_unset,
# model_as_dict = model_as_dict,
enabled = enabled,
inc_cont_edu_cert = inc_cont_edu_cert,
):
cont_edu_cert_person_result_list.append(load_cont_edu_cert_person_result)
else:
cont_edu_cert_person_result_list.append(None)
response_data = cont_edu_cert_person_result_list
else:
return mk_resp(data=False, status_code=400, response=response) # Bad Request
return mk_resp(data=response_data, response=response)
# ### END ### API Cont Edu Cert Person ### get_cont_edu_cert_obj_cont_edu_cert_person_list() ###
@router.delete('/cont_edu/cert/person/{obj_id}', response_model=Resp_Body_Base)
async def delete_cont_edu_cert_person_obj(
obj_id: str = Query(..., min_length=1, max_length=22),
x_account_id: str = Header(...),
response: Response = Response,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
obj_type = 'cont_edu_cert_person'
result = delete_obj_template(
obj_type=obj_type,
obj_id=obj_id,
)
return result