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

Mostly working on event device model, methods, and routes. Also general clean up of code.

Browse Source
This commit is contained in:
Scott Idem
2022-03-09 13:43:47 -05:00
parent f073dd337f
commit bca7defb4c
4 changed files with 215 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

101
app/methods/event_device_methods.py
View File

@@ -29,7 +29,7 @@ def load_event_device_obj(
inc_event_cfg: bool = False,
inc_event_location: bool = False,
) -> Event_Device_Base|dict|bool:
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if event_device_rec := sql_select(table_name='v_event_device', record_id=event_device_id): pass
@@ -83,7 +83,7 @@ def get_event_device_rec_list(
limit: int = 100,
offset: int = 0,
) -> list|bool:
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if for_id := redis_lookup_id_random(record_id_random=for_id, table_name=for_type): pass
@@ -124,3 +124,100 @@ def get_event_device_rec_list(
return event_device_rec_li
# ### END ### API Event Device Methods ### get_event_device_rec_list() ###
# ### BEGIN ### API Event Device Methods ### create_update_event_device_obj() ###
# Updated 2022-03-09
def create_update_event_device_obj(
event_device_dict_obj: Event_Device_Base|dict,
event_device_id: int|str = None,
event_id: int|str = None,
event_location_id: int|str = None,
create_sub_obj: bool = False,
fail_any: bool = False, # Fail if any thing goes wrong for sub objects
return_outline: bool = False,
) -> int|bool:
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
# ### SECTION ### Secondary data validation
log.info('Checking requirements...')
if event_device_id:
log.info(f'Event Device ID passed. Update existing Event Device. Event Device ID: {event_device_id}')
if event_device_id := redis_lookup_id_random(record_id_random=event_device_id, table_name='event_device'): pass
else:
log.error('Event Device ID passed but is invalid. Failed requirement.')
return False
else:
log.info('No Event Device ID passed. Create new Event Device. Required: Event ID; Optional: Event Location ID')
if event_id := redis_lookup_id_random(record_id_random=event_id, table_name='event'): pass
else:
log.error('Missing or invalid Event ID passed. Failed requirement.')
log.info(f'Event ID: {event_id}')
return False
if event_location_id := redis_lookup_id_random(record_id_random=event_location_id, table_name='event_location'): pass
elif event_location_id is None: pass
else:
log.warning('Missing or invalid Event Location ID passed. Failed requirement.')
log.info(f'Event Location ID: {event_location_id}')
return False
log.info('Create dictionary or Pydantic object')
log.debug(type(event_device_dict_obj))
if isinstance(event_device_dict_obj, dict):
event_device_dict = event_device_dict_obj
if event_device_id:
event_device_dict['event_device_id'] = event_device_id
if event_id:
event_dict['event_id'] = event_id
if event_location_id:
event_location_dict['event_location_id'] = event_location_id
try:
event_device_obj = Event_Device_Base(**event_device_dict)
except ValidationError as e:
log.error(e.json())
return False
else:
event_device_obj = event_device_dict_obj
if event_device_id:
# NOTE: Can't update the ID alias if it was never set.
event_device_obj.id = event_device_id
if event_id:
event_device_obj.event_id = event_id
if event_location_id:
event_device_obj.event_location_id = event_location_id
log.debug(event_device_obj)
event_device_dict = event_device_obj.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'event_cfg', 'event_location', 'account_id', 'account_id_random', 'created_on', 'updated_on'})
# ### SECTION ### Process data
if event_device_id:
if event_device_dict_up_result := sql_update(data=event_device_dict, table_name='event_device', rm_id_random=True): pass
else:
log.warning(f'Event Device not updated. Event Device ID: {event_device_id}')
log.debug(event_device_dict_up_result)
return False
log.debug(event_device_dict_up_result)
else:
if event_device_dict_in_result := sql_insert(data=event_device_dict, table_name='event_device', rm_id_random=True, id_random_length=8): pass
else:
log.warning(f'Event Device not created.')
log.debug(event_device_dict_in_result)
return False
log.debug(event_device_dict_in_result)
event_device_id = event_device_dict_in_result
event_device_outline = {}
event_device_outline['event_device_id'] = event_device_id
if return_outline:
log.debug(f'Returning the Event Device Outline: {event_device_outline}')
return event_device_outline
else:
log.debug(f'Returning the Event Device ID: {event_device_id}')
return event_device_id
# ### END ### API Event Device Methods ### create_update_event_device_obj() ###