Scott.Idem/OSIT-AE-API-FastAPI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

General code clean up. More work on event, event exhibits, and event tracking related

Browse Source
This commit is contained in:
Scott Idem
2022-02-15 17:47:30 -05:00
parent df6e860620
commit 2ddb549a75
7 changed files with 320 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
app/main.py
View File

@@ -134,7 +134,7 @@ app.include_router(
)
app.include_router(
event_exhibit.router,
prefix='/event/exhibit',
# prefix='/event/exhibit',
tags=['Event Exhibit'],
)
app.include_router(

264
app/methods/event_exhibit_methods.py Normal file
View File

@@ -0,0 +1,264 @@
import datetime
from typing import Dict, List, Optional, Set, Union
from pydantic import BaseModel, EmailStr, Field, PrivateAttr, ValidationError, validator
from app.db_sql import get_account_id_w_for_type_id, redis_lookup_id_random, sql_delete, sql_enable_part, sql_delete, sql_insert, sql_limit_offset_part, sql_select, sql_update
from app.lib_general import log, logging, logger_reset
from app.methods.event_exhibit_tracking_methods import get_event_exhibit_tracking_rec_list, load_event_exhibit_tracking_obj
from app.models.common_field_schema import default_num_bytes
from app.models.event_exhibit_models import Event_Exhibit_Base
# ### BEGIN ### API Event Exhibit Methods ### create_event_exhibit_obj() ###
# Updated 2022-02-15
@logger_reset
def create_event_exhibit_obj(
event_id: int,
event_exhibit_dict_obj: Event_Exhibit_Base,
create_sub_obj: bool = False,
fail_any: bool = False, # Fail if any thing goes wrong for sub objects
log_lvl: int = logging.WARNING, # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
) -> int|bool:
log.setLevel(log_lvl)
# ### SECTION ### Secondary data validation
log.info('Create dictionary or Pydantic object')
log.debug(type(event_exhibit_dict_obj))
if isinstance(event_exhibit_dict_obj, dict):
event_exhibit_dict = event_exhibit_dict_obj
event_exhibit_dict['event_id'] = event_id
try:
event_exhibit_obj = Event_Exhibit_Base(**event_exhibit_dict)
except ValidationError as e:
log.error(e.json())
return False
else:
event_exhibit_obj = event_exhibit_dict_obj
event_exhibit_obj.event_id = event_id
log.debug(event_exhibit_obj)
event_exhibit_dict = event_exhibit_obj.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'event', 'event_id_random', 'created_on', 'updated_on'})
# ### SECTION ### Process data
if event_exhibit_dict_in_result := sql_insert(
data = event_exhibit_dict,
table_name = 'event_exhibit',
rm_id_random = True,
id_random_length = default_num_bytes
):
event_exhibit_id = event_exhibit_dict_in_result
log.info(f'Returning the new Event Exhibit ID: {event_exhibit_id}')
return event_exhibit_id
else:
log.warning(f'Event Exhibit not created.')
return event_exhibit_dict_in_result # False or None
# ### END ### API Event Exhibit Methods ### create_event_exhibit_obj() ###
# ### BEGIN ### API Event Exhibit Methods ### update_event_exhibit_obj() ###
# Updated 2022-02-15
@logger_reset
def update_event_exhibit_obj(
event_exhibit_id: int,
event_exhibit_dict_obj: Event_Exhibit_Base,
log_lvl: int = logging.WARNING, # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
) -> bool:
log.setLevel(log_lvl)
# ### SECTION ### Secondary data validation
log.info('Create dictionary or Pydantic object')
log.debug(type(event_exhibit_dict_obj))
if isinstance(event_exhibit_dict_obj, dict):
event_exhibit_dict = event_exhibit_dict_obj
event_exhibit_dict['id'] = event_exhibit_id
try:
event_exhibit_obj = Person_Base(**event_exhibit_dict)
except ValidationError as e:
log.error(e.json())
return False
else:
event_exhibit_obj = event_exhibit_dict_obj
event_exhibit_obj.id = event_exhibit_id
log.debug(event_exhibit_obj)
event_exhibit_dict = event_exhibit_obj.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'event', 'event_id_random', 'created_on', 'updated_on'})
# ### SECTION ### Process data
if event_exhibit_dict_up_result := sql_update(
data = event_exhibit_dict,
table_name = 'event_exhibit',
rm_id_random = True,
):
log.info(f'Updated the Event Exhibit ID: {event_exhibit_id}')
return event_exhibit_id
else:
log.warning(f'Event Exhibit not updated.')
return event_exhibit_dict_up_result # False or None
# ### END ### API Event Exhibit Methods ### update_event_exhibit_obj() ###
# ### BEGIN ### API Event Exhibit Methods ### delete_event_exhibit_obj() ###
# Updated 2022-02-15
@logger_reset
def remove_event_exhibit_obj(
event_exhibit_id: int,
log_lvl: int = logging.DEBUG, # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
) -> bool|None:
log.setLevel(log_lvl)
if event_exhibit_dict_del_result := sql_delete(
table_name = 'event_exhibit',
record_id = event_exhibit_id,
log_lvl = log_lvl,
):
log.info(f'Event Exhibit was deleted.')
return True
else:
log.warning(f'Event Exhibit not deleted.')
return event_exhibit_dict_del_result # False or None
# ### END ### API Event Exhibit Methods ### delete_event_exhibit_obj() ###
# ### BEGIN ### API Event Exhibit Methods ### load_event_exhibit_obj() ###
# Updated 2022-02-14
@logger_reset
def load_event_exhibit_obj(
event_exhibit_id: int,
limit: int = 100,
offset: int = 0,
by_alias: bool = True,
exclude_unset: bool = True,
model_as_dict: bool = False,
enabled: str = 'enabled', # enabled, disabled, all
inc_address: bool = False,
inc_contact: bool = False,
inc_event_badge: bool = False,
inc_event_exhibit_tracking_list: bool = False,
inc_event_person: bool = False,
) -> Event_Exhibit_Base|dict|bool:
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
# if event_exhibit_id := redis_lookup_id_random(record_id_random=event_exhibit_id, table_name='event_exhibit'): pass
# else: return False
if event_exhibit_rec := sql_select(table_name='v_event_exhibit', record_id=event_exhibit_id, log_lvl=logging.DEBUG): pass
else: return False
try:
event_exhibit_obj = Event_Exhibit_Base(**event_exhibit_rec)
log.debug(event_exhibit_obj)
except ValidationError as e:
log.error(e.json())
return False
# Updated 2022-02-15
if inc_event_exhibit_tracking_list:
log.info('Need to include Event Exhibit Tracking List data...')
if event_exhibit_tracking_rec_list_result := get_event_exhibit_tracking_rec_list(
event_exhibit_id = event_exhibit_id,
limit = limit,
enabled = enabled,
):
event_exhibit_tracking_result_list = []
for event_exhibit_tracking_rec in event_exhibit_tracking_rec_list_result:
event_exhibit_tracking_result_list.append(
load_event_exhibit_tracking_obj(
event_exhibit_tracking_id = event_exhibit_tracking_rec.get('event_exhibit_tracking_id'),
inc_event_person = inc_event_person,
)
)
event_exhibit_obj.event_exhibit_tracking_list = event_exhibit_tracking_result_list
else: event_exhibit_obj.event_exhibit_tracking_list = []
if model_as_dict:
return event_exhibit_obj.dict(by_alias=by_alias, exclude_unset=exclude_unset) # pylint: disable=no-member
else:
return event_exhibit_obj
# ### END ### API Event Exhibit Methods ### load_event_exhibit_obj() ###
# ### BEGIN ### API Event Exhibit Methods ### get_event_exhibit_rec_list() ###
# for_obj_type: account, event, event_person, event_exhibit
# Updated 2022-02-14
@logger_reset
def get_event_exhibit_rec_list(
event_id: int|str = None,
event_exhibit_id: int|str = None,
event_person_id: int|str = None,
enabled: str = 'enabled', # enabled, disabled, all
limit: int = 100,
offset: int = 0,
) -> list|bool:
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
data = {}
sql_select_event_id = ''
sql_select_event_exhibit_id = ''
sql_select_event_person_id = ''
if event_id:
if event_id := redis_lookup_id_random(record_id_random=event_id, table_name='event'): pass
else: return False
data['event_id'] = event_id
sql_select_event_id = f'`event_exhibit`.event_id = :event_id'
else:
sql_select_event_id = f'`event_exhibit`.event_id IS NOT NULL'
if event_exhibit_id:
if event_exhibit_id := redis_lookup_id_random(record_id_random=event_exhibit_id, table_name='event_exhibit'): pass
else: return False
data['event_exhibit_id'] = event_exhibit_id
sql_select_event_exhibit_id = f'AND `event_exhibit`.event_exhibit_id = :event_exhibit_id'
if event_person_id:
if event_person_id := redis_lookup_id_random(record_id_random=event_person_id, table_name='event_person'): pass
else: return False
data['event_person_id'] = event_person_id
sql_select_event_person_id = f'AND `event_exhibit`.event_person_id = :event_person_id'
sql_enabled, data['enable'] = sql_enable_part(table_name='event_exhibit', enabled=enabled) # Reasonably safe return str and bool
sql_limit = sql_limit_offset_part(limit=limit, offset=offset) # Reasonably safe return str
if event_id or event_exhibit_id or event_person_id: pass
else: return False
sql_enabled = ''
if limit:
data['limit'] = limit
sql_limit = f'LIMIT :limit'
else:
sql_limit = ''
sql = f"""
SELECT `event_exhibit`.id AS 'event_exhibit_id', `event_exhibit`.id_random AS 'event_exhibit_id_random'
FROM `event_exhibit` AS `event_exhibit`
WHERE
{sql_select_event_id}
{sql_select_event_exhibit_id}
{sql_select_event_person_id}
{sql_enabled}
ORDER BY `event_exhibit`.created_on DESC, `event_exhibit`.updated_on DESC
{sql_limit};
"""
if event_exhibit_rec_li_result := sql_select(data=data, sql=sql, as_list=True):
log.info('Got a list result')
event_exhibit_rec_li = event_exhibit_rec_li_result
else: # [] or False
log.info('No results or something went wrong')
event_exhibit_rec_li = event_exhibit_rec_li_result
log.debug(event_exhibit_rec_li_result)
return event_exhibit_rec_li
# ### END ### API Event Exhibit Methods ### get_event_exhibit_rec_list() ###

16
app/methods/event_exhibit_tracking_methods.py
View File

@@ -4,7 +4,7 @@ import datetime
from typing import Dict, List, Optional, Set, Union
from pydantic import BaseModel, EmailStr, Field, PrivateAttr, ValidationError, validator
from app.db_sql import get_account_id_w_for_type_id, redis_lookup_id_random, sql_delete, sql_insert, sql_select, sql_update
from app.db_sql import get_account_id_w_for_type_id, redis_lookup_id_random, sql_enable_part, sql_delete, sql_insert, sql_limit_offset_part, sql_select, sql_update
from app.lib_general import log, logging, logger_reset
from app.models.common_field_schema import default_num_bytes
@@ -15,7 +15,7 @@ from app.models.event_exhibit_tracking_models import Event_Exhibit_Tracking_Base
# Updated 2022-02-15
@logger_reset
def create_event_exhibit_tracking_obj(
event_exhibit_id: int|str,
event_exhibit_id: int,
event_exhibit_tracking_dict_obj: Event_Exhibit_Tracking_Base,
event_badge_id: int = None,
event_person_id: int = None,
@@ -131,7 +131,7 @@ def remove_event_exhibit_tracking_obj(
# Updated 2022-02-14
@logger_reset
def load_event_exhibit_tracking_obj(
event_exhibit_tracking_id: int|str,
event_exhibit_tracking_id: int,
limit: int = 100,
offset: int = 0,
by_alias: bool = True,
@@ -147,8 +147,8 @@ def load_event_exhibit_tracking_obj(
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if event_exhibit_tracking_id := redis_lookup_id_random(record_id_random=event_exhibit_tracking_id, table_name='event_exhibit_tracking'): pass
else: return False
# if event_exhibit_tracking_id := redis_lookup_id_random(record_id_random=event_exhibit_tracking_id, table_name='event_exhibit_tracking'): pass
# else: return False
if event_exhibit_tracking_rec := sql_select(table_name='v_event_exhibit_tracking', record_id=event_exhibit_tracking_id): pass
else: return False
@@ -175,7 +175,7 @@ def load_event_exhibit_tracking_obj(
else: event_exhibit_tracking_obj.event_exhibit = None
if inc_event_person:
log.info('Need to include Event Exhibit data...')
log.info('Need to include Event Person data...')
event_person_id = event_exhibit_tracking_rec.get('event_person_id', None)
log.debug(event_person_id)
from app.methods.event_person_methods import load_event_person_obj
@@ -255,7 +255,7 @@ def get_event_exhibit_tracking_rec_list(
sql = f"""
SELECT `event_exhibit_tracking`.id AS 'event_exhibit_tracking_id', `event_exhibit_tracking`.id_random AS 'event_exhibit_tracking_id_random'
FROM `event_exhibit_tracking` AS `event_exhibit_tracking`
FROM `v_event_exhibit_tracking` AS `event_exhibit_tracking`
WHERE
{sql_select_event_id}
{sql_select_event_exhibit_id}
@@ -265,7 +265,7 @@ def get_event_exhibit_tracking_rec_list(
{sql_limit};
"""
if event_exhibit_rec_li_result := sql_select(data=data, sql=sql, as_list=True):
if event_exhibit_rec_li_result := sql_select(data=data, sql=sql, as_list=True, log_lvl=logging.WARNING):
log.info('Got a list result')
event_exhibit_rec_li = event_exhibit_rec_li_result
else: # [] or False

1
app/methods/user_methods.py
View File

@@ -1,4 +1,3 @@
from __future__ import annotations
import datetime, random, secrets
import urllib.parse

14
app/models/event_exhibit_tracking_models.py
View File

@@ -26,8 +26,8 @@ class Event_Exhibit_Tracking_Base(BaseModel):
alias = 'event_exhibit_tracking_id'
)
# event_id_random: Optional[str]
# event_id: Optional[int]
event_id_random: Optional[str]
event_id: Optional[int]
event_exhibit_id_random: Optional[str]
event_exhibit_id: Optional[int]
@@ -93,6 +93,16 @@ class Event_Exhibit_Tracking_Base(BaseModel):
return redis_lookup_id_random(record_id_random=id_random, table_name='event_exhibit_tracking')
return None
@validator('event_id', always=True)
def event_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if isinstance(v, int) and v > 0: return v
elif id_random := values.get('event_id_random'):
return redis_lookup_id_random(record_id_random=id_random, table_name='event')
return None
@validator('event_exhibit_id', always=True)
def event_exhibit_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)

52
app/routers/event_exhibit.py
View File

@@ -9,6 +9,8 @@ from app.db_sql import sql_enable_part, sql_insert, sql_update, sql_insert_or_up
from .api_crud import delete_obj_template, get_obj_template, get_obj_li_template, patch_obj_template, post_obj_template
from app.methods.event_exhibit_methods import load_event_exhibit_obj
from app.models.event_exhibit_models import Event_Exhibit_Base
from app.models.response_models import Resp_Body_Base, mk_resp
@@ -16,7 +18,7 @@ from app.models.response_models import Resp_Body_Base, mk_resp
router = APIRouter()
@router.post('', response_model=Resp_Body_Base)
@router.post('/event/exhibit', response_model=Resp_Body_Base)
async def post_event_exhibit_obj(
obj: Event_Exhibit_Base,
x_account_id: str = Header(...),
@@ -40,7 +42,7 @@ async def post_event_exhibit_obj(
return result
@router.patch('/{obj_id}', response_model=Resp_Body_Base)
@router.patch('/event/exhibit/{obj_id}', response_model=Resp_Body_Base)
async def patch_event_exhibit_obj(
obj_id: str = Query(..., min_length=1, max_length=22),
obj: Event_Exhibit_Base = None,
@@ -68,7 +70,7 @@ async def patch_event_exhibit_obj(
return result
@router.get('/list', response_model=Resp_Body_Base)
@router.get('/event/exhibit/list', response_model=Resp_Body_Base)
async def get_event_exhibit_obj_li(
for_obj_type: Optional[str] = Query(None, min_length=2, max_length=50),
for_obj_id: Optional[str] = Query(None, min_length=1, max_length=22),
@@ -91,28 +93,42 @@ async def get_event_exhibit_obj_li(
return result
@router.get('/{obj_id}', response_model=Resp_Body_Base)
# ### BEGIN ### API Event Exhibit ### get_event_exhibit_obj() ###
# Updated 2022-02-15
@router.get('/event/exhibit/{event_exhibit_id}', response_model=Resp_Body_Base)
async def get_event_exhibit_obj(
obj_id: str = Query(..., min_length=1, max_length=22),
x_account_id: str = Header(...),
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
response: Response = Response,
event_exhibit_id: str = Query(..., min_length=11, max_length=22),
inc_event_exhibit_tracking_list: bool = False,
inc_event_person: bool = False,
commons: Common_Route_Params = Depends(common_route_params),
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
obj_type = 'event_exhibit'
result = get_obj_template(
obj_type=obj_type,
obj_id=obj_id,
by_alias=True,
exclude_unset=True,
)
return result
# ### SECTION ### Secondary data validation
event_exhibit_id_random = event_exhibit_id # This is used later for the response data
if event_exhibit_id := redis_lookup_id_random(record_id_random=event_exhibit_id, table_name='event_exhibit'): pass
else: return mk_resp(data=None, status_code=404, response=commons.response, status_message='The Event Exhibit ID was invalid or not found.')
if event_exhibit_obj_result := load_event_exhibit_obj(
event_exhibit_id = event_exhibit_id,
inc_event_exhibit_tracking_list = inc_event_exhibit_tracking_list,
inc_event_person = inc_event_person,
):
log.info('Loading successful. Returning result')
return mk_resp(data=event_exhibit_obj_result, response=commons.response) # Success
elif isinstance(event_exhibit_obj_result, list) or event_exhibit_obj_result is None: # Empty list or None
log.info('No results')
return mk_resp(data=None, status_code=404, response=commons.response) # Not Found
else:
log.warning('Likely bad request')
return mk_resp(data=False, status_code=400, response=commons.response) # Bad Request
# ### END ### API Event Exhibit ### get_event_exhibit_obj() ###
@router.delete('/{obj_id}', response_model=Resp_Body_Base)
@router.delete('/event/exhibit/{obj_id}', response_model=Resp_Body_Base)
async def delete_event_exhibit_obj(
obj_id: str = Query(..., min_length=1, max_length=22),
x_account_id: str = Header(...),

4
app/routers/event_exhibit_tracking.py
View File

@@ -1,12 +1,11 @@
import datetime
#from datetime import datetime, time, timedelta
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, common_route_params, Common_Route_Params
from app.config import settings
from app.db_sql import sql_insert, sql_update, sql_insert_or_update, sql_select, sql_delete, get_id_random,redis_lookup_id_random
from app.db_sql import sql_insert, sql_update, sql_insert_or_update, sql_select, sql_delete, get_id_random, 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
@@ -165,7 +164,6 @@ async def get_event_exhibit_tracking_obj(
inc_event_badge: bool = False,
inc_event_exhibit: bool = False,
inc_event_person: bool = False,
return_obj: bool = True,
commons: Common_Route_Params = Depends(common_route_params),
):