413 lines
16 KiB
Python
413 lines
16 KiB
Python
import datetime, json
|
|
|
|
from typing import Dict, List, Optional, Set, Union
|
|
from pydantic import BaseModel, EmailStr, Field, PrivateAttr, ValidationError, validator
|
|
|
|
from app.db_sql import redis_lookup_id_random, sql_delete, sql_insert, sql_select, sql_update, sql_enable_part, sql_limit_offset_part
|
|
from app.lib_general import log, logging, logger_reset
|
|
|
|
from app.models.common_field_schema import default_num_bytes
|
|
from app.models.grant_models import Grant_Ext, Grant_In
|
|
|
|
|
|
# ### BEGIN ### API Grant Methods ### load_grant_obj() ###
|
|
# Updated 2023-06-23
|
|
@logger_reset
|
|
def load_grant_obj(
|
|
grant_id: int|str,
|
|
|
|
enabled: str = 'enabled', # enabled, disabled, all
|
|
hidden: str = 'not_hidden', # hidden, not_hidden, all
|
|
|
|
inc_event_abstract: bool = False,
|
|
|
|
limit: int = 1500,
|
|
offset: int = 0,
|
|
by_alias: bool = True,
|
|
exclude_unset: bool = True,
|
|
model_as_dict: bool = False,
|
|
) -> Grant_Ext|dict|bool:
|
|
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
|
log.debug(locals())
|
|
|
|
# if grant_id := redis_lookup_id_random(record_id_random=grant_id, table_name='grant'): pass
|
|
# else: return False
|
|
|
|
if grant_rec := sql_select(table_name='v_grant', record_id=grant_id): pass
|
|
else: return False
|
|
|
|
try:
|
|
grant_obj = Grant_Ext(**grant_rec)
|
|
log.debug(grant_obj)
|
|
except ValidationError as e:
|
|
log.error(e.json())
|
|
return False
|
|
|
|
# Updated 2023-06-23
|
|
if inc_event_abstract:
|
|
# log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
|
log.info('Need to include event abstract...')
|
|
|
|
if event_abstract_obj := load_event_abstract_obj(
|
|
event_abstract_id = grant_obj.event_abstract_id,
|
|
enabled = enabled,
|
|
):
|
|
log.debug(event_abstract_obj)
|
|
grant_obj.event_abstract = event_abstract_obj
|
|
else:
|
|
log.debug(event_abstract_obj)
|
|
grant_obj.event_abstract = None
|
|
# log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
|
|
|
if model_as_dict:
|
|
return grant_obj.dict(by_alias=by_alias, exclude_unset=exclude_unset) # pylint: disable=no-member
|
|
else:
|
|
return grant_obj
|
|
# ### END ### API Grant Methods ### load_grant_obj() ###
|
|
|
|
|
|
# ### BEGIN ### API Grant Methods ### get_grant_rec_list() ###
|
|
# Updated 2023-06-23
|
|
@logger_reset
|
|
def get_grant_rec_list(
|
|
account_id: None|str = None,
|
|
event_id: None|str = None,
|
|
|
|
grant_type_code: str = None, # not sure yet
|
|
|
|
hidden: str = 'all', # hidden, not_hidden, all
|
|
|
|
enabled: str = 'enabled', # enabled, disabled, all
|
|
limit: int = 250,
|
|
offset: int = 0,
|
|
) -> list|bool:
|
|
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
|
log.debug(locals())
|
|
|
|
if not account_id and not event_id: return False
|
|
|
|
if account_id := redis_lookup_id_random(record_id_random=account_id, table_name='account'): pass
|
|
elif account_id is None: pass
|
|
else: return False
|
|
|
|
if event_id := redis_lookup_id_random(record_id_random=event_id, table_name='event'): pass
|
|
elif event_id is None: pass
|
|
else: return False
|
|
|
|
data = {}
|
|
if account_id:
|
|
data['account_id'] = account_id
|
|
sql_account_id = f'`grant`.account_id = :account_id'
|
|
else: sql_account_id = '1=1'
|
|
|
|
if event_id:
|
|
data['event_id'] = event_id
|
|
sql_event_id = f'AND `grant`.event_id = :event_id'
|
|
else: sql_event_id = ''
|
|
|
|
# if abstract_type_code:
|
|
# data['abstract_type_code'] = abstract_type_code
|
|
# sql_abstract_type_code = f'AND `grant`.abstract_type_code = :abstract_type_code'
|
|
# else: sql_abstract_type_code = ''
|
|
|
|
if hidden in ['hidden', 'not_hidden', 'all']:
|
|
log.info(f'Creating partial SQL string for "hidden" check. Printed: {hidden}')
|
|
if hidden == 'not_hidden':
|
|
sql_hidden = f'AND (`grant`.hide IS NULL OR `grant`.hide = FALSE)'
|
|
elif hidden == 'hidden':
|
|
sql_hidden = f'AND `grant`.hide = TRUE'
|
|
elif hidden == 'all':
|
|
sql_hidden = f'AND (`grant`.hide IS NULL OR `grant`.hide IS NOT NULL)'
|
|
log.debug(sql_hidden)
|
|
else: sql_hidden = ''
|
|
|
|
sql_enabled, data['enable'] = sql_enable_part(table_name='grant', enabled=enabled) # Reasonably safe return str and bool
|
|
sql_limit = sql_limit_offset_part(limit=limit, offset=offset) # Reasonably safe return str
|
|
|
|
sql = f"""
|
|
SELECT `grant`.id AS 'grant_id', `grant`.id_random AS 'grant_id_random'
|
|
FROM `v_grant` AS `grant`
|
|
WHERE
|
|
{sql_account_id}
|
|
{sql_event_id}
|
|
{sql_hidden}
|
|
{sql_enabled}
|
|
ORDER BY grant.priority DESC, grant.sort DESC, grant.name ASC, `grant`.created_on DESC, `grant`.updated_on DESC
|
|
{sql_limit};
|
|
"""
|
|
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
|
log.debug(sql)
|
|
|
|
if grant_rec_li_result := sql_select(data=data, sql=sql, as_list=True):
|
|
grant_rec_li = grant_rec_li_result
|
|
else: # [] or False
|
|
grant_rec_li = grant_rec_li_result
|
|
|
|
log.debug(grant_rec_li_result)
|
|
|
|
return grant_rec_li
|
|
# ### END ### API Grant Methods ### get_grant_rec_list() ###
|
|
|
|
|
|
# ### BEGIN ### API Grant Methods ### create_update_grant_obj() ###
|
|
# Updated 2023-06-28
|
|
@logger_reset
|
|
def create_update_grant_obj(
|
|
grant_obj: Grant_In,
|
|
grant_id: int = None,
|
|
event_id: int = None,
|
|
# event_person_id: int = 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())
|
|
|
|
log.info('Checking requirements...')
|
|
if grant_id:
|
|
log.info(f'Got: grant_id={grant_id}; Update existing Grant')
|
|
grant_obj.id = grant_id
|
|
elif event_id:
|
|
log.info(f'Got: event_id={event_id}; Create Grant')
|
|
grant_obj.event_id = event_id
|
|
else:
|
|
return False
|
|
|
|
log.debug(type(grant_obj))
|
|
log.debug(grant_obj
|
|
)
|
|
|
|
grant_dict = grant_obj.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'event_abstract', 'event_person', 'created_on', 'updated_on'})
|
|
|
|
if grant_id:
|
|
if grant_dict_up_result := sql_update(data=grant_dict, table_name='grant', rm_id_random=True):
|
|
log.info(f'Grant updated. grant_id={grant_id}')
|
|
pass
|
|
else:
|
|
log.warning(f'Grant not updated. grant_id={grant_id}')
|
|
log.debug(grant_dict_up_result)
|
|
return False
|
|
log.debug(grant_dict_up_result)
|
|
else:
|
|
if grant_dict_in_result := sql_insert(data=grant_dict, table_name='grant', rm_id_random=True, id_random_length=None):
|
|
log.info(f'Grant created. grant_id={grant_dict_in_result}')
|
|
else:
|
|
log.warning(f'Grant not created.')
|
|
log.debug(grant_dict_in_result)
|
|
return False
|
|
log.debug(grant_dict_in_result)
|
|
|
|
grant_id = grant_dict_in_result # False, None, integer
|
|
|
|
grant_outline = {}
|
|
grant_outline['event_id'] = event_id
|
|
grant_outline['grant_id'] = grant_id
|
|
|
|
if return_outline:
|
|
log.debug(f'Returning the Grant Outline: {grant_outline}')
|
|
return grant_outline
|
|
else:
|
|
log.debug(f'Returning the Grant ID: {grant_id}')
|
|
return grant_id
|
|
# ### END ### API Grant Methods ### create_update_grant_obj() ###
|
|
|
|
|
|
|
|
# # ### BEGIN ### API Grant Methods ### create_update_grant_obj_old() ###
|
|
# # Updated 2023-03-20
|
|
# @logger_reset
|
|
# def create_update_grant_obj_old(
|
|
# grant_dict_obj: Grant_In|dict,
|
|
# grant_id: int|str = None,
|
|
# event_id: int|str = None,
|
|
# event_person_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())
|
|
|
|
# log.info('Checking requirements...')
|
|
# if grant_id:
|
|
# log.info(f'Grant ID passed. Update existing Grant. Grant ID: {grant_id}')
|
|
|
|
# if grant_id := redis_lookup_id_random(record_id_random=grant_id, table_name='grant'): pass
|
|
# else:
|
|
# log.error('Grant ID passed but is invalid. Failed requirement.')
|
|
# return False
|
|
|
|
# if event_id := redis_lookup_id_random(record_id_random=event_id, table_name='event'):
|
|
# log.info(f'Event ID: {event_id}')
|
|
# elif event_id is None:
|
|
# log.error('Missing Event ID. Not required. Ignoring.')
|
|
# log.info(f'Event ID: {event_id}')
|
|
# else:
|
|
# log.error('Invalid Event ID passed. Not required. But not ignoring since it is likely invalid.')
|
|
# log.info(f'Event ID: {event_id}')
|
|
# return False
|
|
# else:
|
|
# log.info('No Grant ID passed. Create new Grant. Required: Event ID')
|
|
|
|
# if event_id := redis_lookup_id_random(record_id_random=event_id, table_name='event'):
|
|
# log.info(f'Event ID: {event_id}')
|
|
# elif event_id is None:
|
|
# log.error('Missing Event ID. Failed requirement.')
|
|
# log.info(f'Event ID: {event_id}')
|
|
# return False
|
|
# else:
|
|
# log.error('Invalid Event ID passed. Failed requirement.')
|
|
# log.info(f'Event ID: {event_id}')
|
|
# return False
|
|
# # else:
|
|
# # log.error('Missing or invalid Event ID passed. Failed requirement.')
|
|
# # log.info(f'Event ID: {event_id}')
|
|
# # return False
|
|
|
|
# log.debug(type(grant_dict_obj))
|
|
# if isinstance(grant_dict_obj, dict):
|
|
# grant_dict = grant_dict_obj
|
|
# if grant_id:
|
|
# grant_dict['grant_id'] = grant_id
|
|
# if event_id:
|
|
# grant_dict['event_id'] = event_id
|
|
# if event_person_id:
|
|
# grant_dict['event_person_id'] = event_person_id
|
|
# try:
|
|
# grant_obj = Grant_In(**grant_dict)
|
|
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
|
# log.debug(grant_obj)
|
|
# except ValidationError as e:
|
|
# log.error(e.json())
|
|
# return False
|
|
# else:
|
|
# grant_obj = grant_dict_obj
|
|
# if grant_id:
|
|
# # NOTE: Can't update the ID alias if it was never set.
|
|
# grant_obj.id = grant_id
|
|
# if event_id:
|
|
# grant_obj.event_id = event_id
|
|
# if event_person_id:
|
|
# grant_obj.event_person_id = event_person_id
|
|
|
|
# grant_dict = grant_obj.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'event_person', 'event_person_list', 'created_on', 'updated_on'})
|
|
|
|
# if grant_id:
|
|
# if grant_dict_up_result := sql_update(data=grant_dict, table_name='grant', rm_id_random=True): pass
|
|
# else:
|
|
# log.warning(f'Grant not updated. Grant ID: {grant_id}')
|
|
# log.debug(grant_dict_up_result)
|
|
# return False
|
|
# log.debug(grant_dict_up_result)
|
|
# else:
|
|
# if grant_dict_in_result := sql_insert(data=grant_dict, table_name='grant', rm_id_random=True, id_random_length=None): pass
|
|
# else:
|
|
# log.warning(f'Grant not created.')
|
|
# log.debug(grant_dict_in_result)
|
|
# return False
|
|
# log.debug(grant_dict_in_result)
|
|
|
|
# grant_id = grant_dict_in_result
|
|
|
|
# grant_outline = {}
|
|
# grant_outline['event_id'] = event_id
|
|
# grant_outline['grant_id'] = grant_id
|
|
# grant_outline['event_person_id'] = event_person_id
|
|
# # grant_outline['event_presenter_list'] = []
|
|
|
|
# # if grant_obj.event_presenter_list and isinstance(grant_obj.event_presenter_list, list):
|
|
# # log.info(f'Event Presenter List was found. Loop through and create a new Event Presenter for each and link them to the new Grant. Grant ID: {grant_id}')
|
|
# # for event_presenter_obj in grant_obj.event_presenter_list:
|
|
# # # NOTE: Use object model version because of better type checking and validations
|
|
# # log.debug(event_presenter_obj)
|
|
# # if event_presenter_id := event_presenter_obj.id: pass
|
|
# # else: event_presenter_id = None
|
|
# # # event_presenter_obj.event_id = event_id
|
|
# # # event_presenter_obj.event_session_id = event_session_id
|
|
# # create_update_event_presenter_obj_result = create_update_event_presenter_obj_v4(
|
|
# # event_presenter_dict_obj = event_presenter_obj,
|
|
# # event_presenter_id = event_presenter_id,
|
|
# # event_id = event_id,
|
|
# # event_session_id = event_session_id,
|
|
# # grant_id = grant_id,
|
|
# # fail_any = fail_any,
|
|
# # return_outline = return_outline,
|
|
# # )
|
|
# # if isinstance(create_update_event_presenter_obj_result, int):
|
|
# # event_presenter_id = create_update_event_presenter_obj_result
|
|
# # elif create_update_event_presenter_obj_result == True: pass
|
|
# # else:
|
|
# # log.warning(f'Create or Update failed while trying create_update_event_presenter_obj_v4(): {create_update_event_presenter_obj_result}')
|
|
# # event_presenter_id = None
|
|
|
|
# # grant_outline['event_presenter_id'] = event_presenter_id
|
|
|
|
# if return_outline:
|
|
# log.debug(f'Returning the Grant Outline: {grant_outline}')
|
|
# return grant_outline
|
|
# else:
|
|
# log.debug(f'Returning the Grant ID: {grant_id}')
|
|
# return grant_id
|
|
# # ### END ### API Grant Methods ### create_update_grant_obj_old() ###
|
|
|
|
|
|
# # ### BEGIN ### API Grant Methods ### remove_grant_obj() ###
|
|
# # Updated 2023-03-22
|
|
# @logger_reset
|
|
# def remove_grant_obj(
|
|
# grant_id: int,
|
|
|
|
# method: None|str = None,
|
|
|
|
# log_lvl: int = logging.DEBUG, # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
|
# ) -> bool|None:
|
|
# log.setLevel(log_lvl)
|
|
|
|
# if method is None or method == 'disable':
|
|
# data = {'enable': False}
|
|
|
|
# if grant_dict_up_result := sql_update(
|
|
# table_name = 'grant',
|
|
# record_id = grant_id,
|
|
# data = data,
|
|
# log_lvl = log_lvl,
|
|
# ):
|
|
# log.info(f'Grant was disabled.')
|
|
# return True
|
|
# else:
|
|
# log.warning(f'Grant not disabled.')
|
|
# return grant_dict_up_result # False or None
|
|
# elif method == 'delete':
|
|
# if grant_dict_del_result := sql_delete(
|
|
# table_name = 'grant',
|
|
# record_id = grant_id,
|
|
# log_lvl = log_lvl,
|
|
# ):
|
|
# log.info(f'Grant was deleted.')
|
|
# return True
|
|
# else:
|
|
# log.warning(f'Grant not deleted.')
|
|
# return grant_dict_del_result # False or None
|
|
# elif method == 'hide':
|
|
# data = {'hide': True}
|
|
|
|
# if grant_dict_up_result := sql_update(
|
|
# table_name = 'grant',
|
|
# record_id = grant_id,
|
|
# data = data,
|
|
# log_lvl = log_lvl,
|
|
# ):
|
|
# log.info(f'Grant was hidden.')
|
|
# return True
|
|
# else:
|
|
# log.warning(f'Grant not hidden.')
|
|
# return grant_dict_up_result # False or None
|
|
# else:
|
|
# log.error('We should not be here. Something went wrong in remove_grant_obj()!')
|
|
# return False
|
|
|
|
|
|
# # ### END ### API Grant Methods ### remove_grant_obj() ###
|