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.INFO) # 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() ###