import datetime #from datetime import datetime, time, timedelta from fastapi import APIRouter, Body, Depends, Header, HTTPException, Query, status from pydantic import BaseModel, EmailStr, Field from typing import Dict, List, Optional, Set, Union from app.lib_general import * from ..log import * from app.config import settings from app.db_sql import * from .api_crud import delete_obj_template, get_obj_template, get_obj_li_template, patch_obj_template, post_obj_template from app.methods.event_methods import load_event_obj from app.models.event_models import Event_Base from app.models.response_models import * router = APIRouter() @router.post('', response_model=Resp_Body_Base) async def post_event_obj( obj: Event_Base, x_account_id: str = Header(...), return_obj: Optional[bool] = True, by_alias: Optional[bool] = True, exclude_unset: Optional[bool] = True, ): log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(locals()) obj_type = 'event' 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('/{obj_id}', response_model=Resp_Body_Base) async def patch_event_obj( obj_id: str = Query(..., min_length=1, max_length=22), obj: Event_Base = None, x_account_id: Optional[str] = Header(..., ), return_obj: Optional[bool] = True, by_alias: Optional[bool] = True, exclude_unset: Optional[bool] = True, ): log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(locals()) obj_type = 'event' 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 # @router.get('/list', response_model=Resp_Body_Base) # async def get_event_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), # x_account_id: str = Header(...), # by_alias: Optional[bool] = True, # exclude_unset: Optional[bool] = True, # ): # log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL # log.debug(locals()) # obj_type = 'event' # 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 @router.get('/list', response_model=Resp_Body_Base) async def get_event_obj_li( for_obj_type: Optional[str] = Query(None, min_length=2, max_length=50), # account for_obj_id: Optional[str] = Query(None, min_length=1, max_length=22), enabled: str = 'enabled', # enabled, disabled, all from_datetime: Optional[datetime.datetime] = None, # based on start_datetime to_datetime: Optional[datetime.datetime] = None, # based on start_datetime limit: int = None, inc_event_abstract_list: bool = False, inc_event_badge_list: bool = False, inc_event_device_list: bool = False, inc_event_exhibit_list: bool = False, inc_event_file_list: bool = False, inc_event_location_list: bool = False, inc_event_person_list: bool = False, inc_event_presentation_list: bool = False, inc_event_presenter_list: bool = False, inc_event_registration_list: bool = False, inc_event_session_list: bool = False, inc_event_track_list: bool = False, inc_event_cfg: bool = False, inc_event_registration_cfg: bool = False, inc_poc_event_person: bool = False, inc_poc_person: bool = False, inc_user: bool = False, x_account_id: str = Header(...), by_alias: bool = True, exclude_unset: bool = True, ): log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(locals()) data = {} if for_obj_type == 'account' and for_obj_id: data['account_id'] = redis_lookup_id_random(record_id_random=for_obj_id, table_name='account') sql_for_obj_type_id = f'`event`.account_id = :account_id' # elif for_obj_type == 'test' and for_obj_id: # data['test_id'] = redis_lookup_id_random(record_id_random=for_obj_id, table_name='test') # sql_for_obj_type_id = f'`test`.test_id = :test_id' else: sql_for_obj_type_id = '' return mk_resp(data=False, status_code=400) if enabled in ['enabled', 'disabled', 'all']: if enabled == 'enabled': data['enable'] = True sql_enabled = f'AND `event`.enable = :enable' elif enabled == 'disabled': data['enable'] = False sql_enabled = f'AND `event`.enable = :enable' elif enabled == 'all': sql_enabled = '' else: return mk_resp(data=False, status_code=400) if from_datetime and to_datetime: data['from_datetime'] = from_datetime data['to_datetime'] = to_datetime sql_from_to_datetime = f'AND `event`.start_datetime >= :from_datetime AND `event`.start_datetime <= :to_datetime' elif from_datetime: data['from_datetime'] = from_datetime sql_from_to_datetime = f'AND `event`.start_datetime >= :from_datetime' elif to_datetime: data['to_datetime'] = to_datetime sql_from_to_datetime = f'AND `event`.start_datetime <= :to_datetime' else: sql_from_to_datetime = '' if limit: data['limit'] = limit sql_limit = f'LIMIT :limit' else: sql_limit = '' sql = f""" SELECT `event`.id AS 'event_id', `event`.id_random AS 'event_id_random' FROM `event` AS `event` WHERE {sql_for_obj_type_id} {sql_enabled} {sql_from_to_datetime} ORDER BY `event`.created_on DESC, `event`.updated_on DESC {sql_limit}; """ if event_obj_li_result := sql_select(data=data, sql=sql, as_list=True): log.debug(event_obj_li_result) event_obj_li = [] for event_obj in event_obj_li_result: event_id = event_obj.get('event_id', None) if event_obj := load_event_obj( event_id=event_id, enabled=enabled, inc_event_abstract_list=inc_event_abstract_list, inc_event_badge_list=inc_event_badge_list, inc_event_device_list=inc_event_device_list, inc_event_exhibit_list=inc_event_exhibit_list, inc_event_file_list=inc_event_file_list, inc_event_location_list=inc_event_location_list, inc_event_person_list=inc_event_person_list, inc_event_presentation_list=inc_event_presentation_list, inc_event_presenter_list=inc_event_presenter_list, inc_event_registration_list=inc_event_registration_list, inc_event_session_list=inc_event_session_list, inc_event_track_list=inc_event_track_list, inc_event_cfg=inc_event_cfg, inc_event_registration_cfg=inc_event_registration_cfg, inc_poc_event_person=inc_poc_event_person, inc_poc_person=inc_poc_person, inc_user=inc_user, ): data = event_obj.dict(by_alias=by_alias, exclude_unset=exclude_unset) event_obj_li.append(data) return mk_resp(data=event_obj_li) else: log.debug(event_obj_li_result) return mk_resp(data=False, status_code=404) # ### BEGIN ### API Event ### get_event_obj() ### # Working well as of 2021-06-04. Using as a template for other routes. @router.get('/{event_id}', response_model=Resp_Body_Base) async def get_event_obj( event_id: str = Query(..., min_length=1, max_length=22), enabled: str = 'enabled', # For now this covers any included objects or object lists limit: int = 500, # For now this covers any included objects or object lists inc_contact_1: bool = False, inc_contact_2: bool = False, inc_contact_3: bool = False, inc_event_abstract_list: bool = False, inc_event_badge_list: bool = False, inc_event_device_list: bool = False, inc_event_exhibit_list: bool = False, inc_event_file_list: bool = False, inc_event_location: bool = False, inc_event_location_list: bool = False, inc_event_person_list: bool = False, inc_event_presentation_list: bool = False, inc_event_presenter_cat: bool = False, inc_event_presenter_list: bool = False, inc_event_registration_list: bool = False, inc_event_session_list: bool = False, inc_event_track: bool = False, inc_event_track_list: bool = False, inc_event_cfg: bool = False, inc_event_registration_cfg: bool = False, inc_location_address: bool = False, inc_poc_event_person: bool = False, x_account_id: str = Header(...), by_alias: Optional[bool] = True, exclude_unset: Optional[bool] = True, ): log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(locals()) if event_id := redis_lookup_id_random(record_id_random=event_id, table_name='event'): pass else: return mk_resp(data=None, status_code=404) if event_obj := load_event_obj( event_id=event_id, enabled=enabled, limit=limit, inc_contact_1=inc_contact_1, inc_contact_2=inc_contact_2, inc_contact_3=inc_contact_3, inc_event_abstract_list=inc_event_abstract_list, inc_event_badge_list=inc_event_badge_list, inc_event_device_list=inc_event_device_list, inc_event_exhibit_list=inc_event_exhibit_list, inc_event_file_list=inc_event_file_list, inc_event_location=inc_event_location, inc_event_location_list=inc_event_location_list, inc_event_person_list=inc_event_person_list, inc_event_presentation_list=inc_event_presentation_list, inc_event_presenter_cat=inc_event_presenter_cat, inc_event_presenter_list=inc_event_presenter_list, inc_event_registration_list=inc_event_registration_list, inc_event_session_list=inc_event_session_list, inc_event_track=inc_event_track, inc_event_track_list=inc_event_track_list, inc_event_cfg=inc_event_cfg, inc_event_registration_cfg=inc_event_registration_cfg, inc_location_address=inc_location_address, inc_poc_event_person=inc_poc_event_person, ): # event_dict = event_obj.dict(by_alias=by_alias, exclude_unset=exclude_unset) pass else: return mk_resp(data=False, status_code=400) # Bad Request return mk_resp(data=event_obj) # ### END ### API Event ### get_event_obj() ### @router.delete('/{obj_id}', response_model=Resp_Body_Base) async def delete_event_obj( obj_id: str = Query(..., min_length=1, max_length=22), x_account_id: str = Header(...), ): log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(locals()) obj_type = 'event' result = delete_obj_template( obj_type=obj_type, obj_id=obj_id, ) return result