from __future__ import annotations import datetime 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_insert, sql_select, sql_update from app.lib_general import log, logging from app.methods.contact_methods import load_contact_obj # from app.methods.event_methods import load_event_obj_list from app.methods.organization_methods import load_organization_obj # from app.methods.person_methods import load_person_obj # from app.methods.post_methods import load_post_obj_list from app.models.user_models import User_Base, User_New_Base, User_Out_Base from app.models.user_role_models import User_Role_Base # ### BEGIN ### API User Methods ### load_user_obj() ### def load_user_obj( user_id: int|str, limit: int = 1000, model_as_dict: bool = False, enabled: str = 'enabled', # enabled, disabled, all inc_address: bool = False, inc_contact: bool = False, inc_event_list: bool = False, inc_journal_list: bool = False, inc_membership: bool = False, inc_order_list: bool = False, inc_order_cart_list: bool = False, inc_organization: bool = False, inc_person: bool = False, inc_post_list: bool = False, inc_post_comment_list: bool = False, inc_user_role_list: bool = False, ) -> User_Out_Base|bool: #log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(locals()) if user_id := redis_lookup_id_random(record_id_random=user_id, table_name='user'): pass else: return False if user_rec := sql_select(table_name='v_user', record_id=user_id): pass else: return False #log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(user_rec) try: user_obj = User_Out_Base(**user_rec) log.debug(user_obj) except ValidationError as e: log.error(e.json()) return False if inc_contact: contact_id = user_rec.get('contact_id', None) if contact_obj_result := load_contact_obj(contact_id=contact_id): contact_obj = contact_obj_result user_rec['contact'] = contact_obj log.debug(user_rec) user_obj.contact = contact_obj else: user_rec['contact'] = None user_obj.contact = None if inc_event_list: if event_dict_list := load_event_obj_list( user_id = user_id, limit = limit, model_as_dict = model_as_dict, enabled = enabled, ): user_obj.event_list = event_dict_list else: user_obj.event_list = None if inc_organization: organization_id = user_rec.get('organization_id', None) if organization_obj_result := load_organization_obj(organization_id=organization_id): organization_obj = organization_obj_result user_rec['organization'] = organization_obj log.debug(user_rec) user_obj.organization = organization_obj else: user_rec['organization'] = None user_obj.organization = None # if inc_person: # person_id = user_rec.get('person_id', None) # if person_obj_result := load_person_obj(person_id=person_id): # person_obj = person_obj_result # user_rec['person'] = person_obj # log.debug(user_rec) # user_obj.person = person_obj # else: # user_rec['person'] = None # user_obj.person = None if inc_post_list: if post_dict_list := load_post_obj_list( user_id = user_id, limit = limit, model_as_dict = model_as_dict, enabled = enabled, ): user_obj.post_list = post_dict_list else: user_obj.post_list = None # NOTE: Including user roles should probably be reviewed if inc_user_role_list: if role_rec_li := sql_select(table_name='v_user_role_detail', field_name='user_id', field_value=user_id, as_list=True): user_rec['role_list'] = role_rec_li #log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(role_rec_li) user_obj.role_list = role_rec_li else: user_rec['role_list'] = None user_obj.role_list = None #log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(user_rec) return user_obj # ### END ### API User Methods ### load_user_obj() ###