Work on better way to update objects.
This commit is contained in:
Scott Idem
@@ -108,7 +108,7 @@ def sql_insert(sql:str|None=None, data:dict|None=None, table_name:str|None=None,
|
||||
|
||||
|
||||
# ### BEGIN ### Core Help CRUD ### sql_update() ###
|
||||
def sql_update(sql:str|None=None, data:dict|None=None, table_name:str|None=None, record_id:int|None=None, record_id_random:str|None=None, rm_id_random:bool=False, id_random_length:None|int=8):
|
||||
def sql_update(sql:str|None=None, data:dict|None=None, table_name:str|None=None, record_id:int|None=None, record_id_random:str|None=None, rm_id_random:bool=False, id_random_length:None|int=None):
|
||||
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(locals())
|
||||
|
||||
|
||||
@@ -4,7 +4,7 @@ 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
|
||||
from app.db_sql import redis_lookup_id_random, sql_insert, sql_select, sql_update
|
||||
from app.lib_general import log, logging
|
||||
|
||||
from app.models.address_models import Address_Base
|
||||
@@ -59,17 +59,30 @@ def load_address_obj(address_id:int|str) -> Address_Base|bool:
|
||||
|
||||
|
||||
# ### BEGIN ### API Address Methods ### update_address_obj() ###
|
||||
def update_address_obj(address_obj_update:Address_Base):
|
||||
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
def update_address_obj(
|
||||
address_id: int|str, # Ideally the int ID should be passed. This allows for updating of the id_random value.
|
||||
address_obj_up: Address_Base,
|
||||
create_missing_obj: bool = False,
|
||||
) -> bool:
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(locals())
|
||||
|
||||
log.debug(address_obj_update)
|
||||
if address_obj_up_result = update_address_obj(address_obj_update):
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
|
||||
if address_id := redis_lookup_id_random(record_id_random=address_id, table_name='address'): pass
|
||||
else: return False
|
||||
|
||||
address_obj_up.id = address_id
|
||||
|
||||
log.debug(address_obj_up)
|
||||
# log.debug(address_obj_up.dict(by_alias=True, exclude_unset=True))
|
||||
log.debug(address_obj_up.dict(by_alias=False, exclude_unset=True))
|
||||
# log.debug(address_obj_up.dict(by_alias=False, exclude_unset=False))
|
||||
|
||||
address_dict_up = address_obj_up.dict(by_alias=False, exclude_unset=True)
|
||||
|
||||
if address_obj_up_result := sql_update(data=address_dict_up, table_name='address', rm_id_random=True):
|
||||
log.debug(address_obj_up_result)
|
||||
return True
|
||||
else:
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(address_obj_up_result)
|
||||
return False
|
||||
|
||||
# ### END ### API Address Methods ### update_address_obj() ###
|
||||
|
||||
@@ -7,7 +7,7 @@ from pydantic import BaseModel, EmailStr, Field, PrivateAttr, ValidationError, v
|
||||
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.address_methods import create_address_obj
|
||||
from app.methods.address_methods import create_address_obj, update_address_obj
|
||||
|
||||
from app.models.contact_models import Contact_Base
|
||||
|
||||
@@ -88,37 +88,60 @@ def load_contact_obj(contact_id:int|str, inc_address:bool=False) -> Contact_Base
|
||||
|
||||
|
||||
# ### BEGIN ### API Contact Methods ### update_contact_obj() ###
|
||||
def update_contact_obj(contact_obj_update:Contact_Base):
|
||||
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
def update_contact_obj(
|
||||
contact_id: int|str, # Ideally the int ID should be passed. This allows for updating of the id_random value.
|
||||
contact_obj_up: Contact_Base,
|
||||
create_missing_obj: bool = False,
|
||||
) -> bool:
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(locals())
|
||||
|
||||
if address_obj_update := contact_obj_update.address:
|
||||
log.debug(address_obj_update)
|
||||
if address_obj_up_result := update_address_obj(address_obj_update):
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
if contact_id := redis_lookup_id_random(record_id_random=contact_id, table_name='contact'): pass
|
||||
else: return False
|
||||
|
||||
contact_obj_up.id = contact_id
|
||||
|
||||
log.debug(contact_obj_up)
|
||||
# log.debug(contact_obj_up.dict(by_alias=True, exclude_unset=True))
|
||||
log.debug(contact_obj_up.dict(by_alias=False, exclude_unset=True))
|
||||
# log.debug(contact_obj_up.dict(by_alias=False, exclude_unset=False))
|
||||
|
||||
#contact_dict_up = contact_obj_up.dict(by_alias=False, exclude_unset=True)
|
||||
|
||||
if contact_obj_up.address_id and contact_obj_up.address:
|
||||
address_id = contact_obj_up.address_id
|
||||
address_obj_up = contact_obj_up.address
|
||||
log.debug(address_id)
|
||||
log.debug(address_obj_up)
|
||||
if address_obj_up_result := update_address_obj(
|
||||
address_id=address_id,
|
||||
address_obj_up=address_obj_up,
|
||||
create_missing_obj=create_missing_obj,
|
||||
):
|
||||
log.debug(address_obj_up_result)
|
||||
return True
|
||||
else:
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(address_obj_up_result)
|
||||
return False
|
||||
else:
|
||||
if address_obj_in_result := insert_address_obj(address_obj_insert):
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(address_obj_in_result)c
|
||||
return True # NOTE: This needs to return the new address ID
|
||||
elif contact_obj_up.address and not contact_obj_up.address.id:
|
||||
# NOTE: This will blindly create a new address even if there was one associated but the contact.address_id was not found.
|
||||
address_obj_in = contact_obj_up.address
|
||||
log.debug(address_obj_in)
|
||||
if address_obj_in_result := create_address_obj(address_obj_new=address_obj_in):
|
||||
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(address_obj_in_result)
|
||||
contact_obj_up.address_id = address_obj_in_result
|
||||
else:
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(address_obj_in_result)
|
||||
return False
|
||||
|
||||
contact_obj_update.pop('address')
|
||||
log.debug(contact_obj_update)
|
||||
if contact_obj_up_result = update_contact_obj(contact_obj_update):
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
|
||||
contact_dict_up = contact_obj_up.dict(by_alias=False, exclude_unset=True, exclude={'address'})
|
||||
log.debug(contact_dict_up)
|
||||
|
||||
if contact_obj_up_result := sql_update(data=contact_dict_up, table_name='contact', rm_id_random=True):
|
||||
log.debug(contact_obj_up_result)
|
||||
return True
|
||||
else:
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(contact_obj_up_result)
|
||||
return False
|
||||
# ### END ### API Contact Methods ### update_contact_obj() ###
|
||||
|
||||
@@ -8,12 +8,13 @@ from app.db_sql import redis_lookup_id_random, sql_insert, sql_select
|
||||
from app.lib_general import log, logging
|
||||
|
||||
from app.methods.address_methods import load_address_obj
|
||||
from app.methods.contact_methods import create_contact_obj, load_contact_obj
|
||||
from app.methods.contact_methods import create_contact_obj, load_contact_obj, update_contact_obj
|
||||
|
||||
from app.models.organization_models import Organization_Base
|
||||
|
||||
|
||||
# ### BEGIN ### API Organization Methods ### load_organization_obj() ###
|
||||
# NOTE: This needs to be updated to the newer method template. Like address, contact, or person -STI 2021-06-10
|
||||
def load_organization_obj(organization_id:int|str, inc_contact:bool=False, inc_address:bool=False) -> Organization_Base|bool:
|
||||
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(locals())
|
||||
@@ -52,3 +53,51 @@ def load_organization_obj(organization_id:int|str, inc_contact:bool=False, inc_a
|
||||
|
||||
return organization_obj
|
||||
# ### END ### API Organization Methods ### load_organization_obj() ###
|
||||
|
||||
|
||||
# ### BEGIN ### API Organization Methods ### update_organization_obj() ###
|
||||
def update_organization_obj(
|
||||
organization_id: int|str, # Ideally the int ID should be passed. This allows for updating of the id_random value.
|
||||
organization_obj_up: Organization_Base,
|
||||
create_missing_obj: bool = False,
|
||||
) -> bool:
|
||||
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(locals())
|
||||
|
||||
if organization_obj_up.contact_id and organization_obj_up.contact:
|
||||
contact_id = organization_obj_up.contact_id
|
||||
contact_obj_up = organization_obj_up.contact
|
||||
log.debug(contact_id)
|
||||
log.debug(contact_obj_up)
|
||||
if contact_obj_up_result := update_contact_obj(
|
||||
contact_id=contact_id,
|
||||
contact_obj_up=contact_obj_up,
|
||||
create_missing_obj=create_missing_obj,
|
||||
):
|
||||
log.debug(contact_obj_up_result)
|
||||
else:
|
||||
log.debug(contact_obj_up_result)
|
||||
return False
|
||||
elif organization_obj_up.contact and not organization_obj_up.contact.id:
|
||||
# NOTE: This will blindly create a new contact even if there was one associated but the organization.contact_id was not found.
|
||||
contact_obj_in = organization_obj_up.contact
|
||||
log.debug(contact_obj_in)
|
||||
if contact_obj_in_result := create_contact_obj(contact_obj_new=contact_obj_in):
|
||||
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(contact_obj_in_result)
|
||||
organization_obj_up.contact_id = contact_obj_in_result
|
||||
else:
|
||||
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(contact_obj_in_result)
|
||||
return False
|
||||
|
||||
organization_dict_up = organization_obj_up.dict(by_alias=False, exclude_unset=True, exclude={'contact', 'person', 'user'})
|
||||
log.debug(organization_dict_up)
|
||||
|
||||
if organization_obj_up_result := sql_update(data=organization_dict_up, table_name='organization', rm_id_random=True):
|
||||
log.debug(organization_obj_up_result)
|
||||
return True
|
||||
else:
|
||||
log.debug(organization_obj_up_result)
|
||||
return False
|
||||
# ### END ### API Organization Methods ### update_organization_obj() ###
|
||||
@@ -9,8 +9,8 @@ from app.lib_general import log, logging
|
||||
|
||||
from app.methods.address_methods import load_address_obj
|
||||
from app.methods.contact_methods import create_contact_obj, load_contact_obj, update_contact_obj
|
||||
from app.methods.organization_methods import load_organization_obj
|
||||
from app.methods.user_methods import load_user_obj
|
||||
from app.methods.organization_methods import load_organization_obj, update_organization_obj
|
||||
from app.methods.user_methods import load_user_obj, update_user_obj
|
||||
|
||||
from app.models.person_models import Person_Base
|
||||
|
||||
@@ -128,83 +128,113 @@ def load_person_obj(
|
||||
|
||||
|
||||
# ### BEGIN ### API Person Methods ### update_person_obj() ###
|
||||
def update_person_obj(person_obj_update:Person_Base):
|
||||
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
def update_person_obj(
|
||||
person_id: int|str, # Ideally the int ID should be passed. This allows for updating of the id_random value.
|
||||
person_obj_up: Person_Base,
|
||||
create_missing_obj: bool = False,
|
||||
) -> bool:
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(locals())
|
||||
|
||||
if contact_obj_update := person_obj_update.contact:
|
||||
log.debug(contact_obj_update)
|
||||
if contact_obj_up_result := update_contact_obj(contact_obj_update):
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
if person_id := redis_lookup_id_random(record_id_random=person_id, table_name='person'): pass
|
||||
else: return False
|
||||
|
||||
person_obj_up.id = person_id
|
||||
|
||||
log.debug(person_obj_up)
|
||||
# log.debug(person_obj_up.dict(by_alias=True, exclude_unset=True))
|
||||
log.debug(person_obj_up.dict(by_alias=False, exclude_unset=True))
|
||||
# log.debug(person_obj_up.dict(by_alias=False, exclude_unset=False))
|
||||
|
||||
if person_obj_up.contact_id and person_obj_up.contact:
|
||||
contact_id = person_obj_up.contact_id
|
||||
contact_obj_up = person_obj_up.contact
|
||||
log.debug(contact_id)
|
||||
log.debug(contact_obj_up)
|
||||
if contact_obj_up_result := update_contact_obj(
|
||||
contact_id=contact_id,
|
||||
contact_obj_up=contact_obj_up,
|
||||
create_missing_obj=create_missing_obj,
|
||||
):
|
||||
log.debug(contact_obj_up_result)
|
||||
return True
|
||||
else:
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(contact_obj_up_result)
|
||||
return False
|
||||
else:
|
||||
if contact_obj_in_result := insert_contact_obj(contact_obj_insert):
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
elif person_obj_up.contact and not person_obj_up.contact.id:
|
||||
# NOTE: This will blindly create a new contact even if there was one associated but the person.contact_id was not found.
|
||||
contact_obj_in = person_obj_up.contact
|
||||
log.debug(contact_obj_in)
|
||||
if contact_obj_in_result := create_contact_obj(contact_obj_new=contact_obj_in):
|
||||
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(contact_obj_in_result)
|
||||
return True # NOTE: This needs to return the new contact ID
|
||||
person_obj_up.contact_id = contact_obj_in_result
|
||||
else:
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(contact_obj_in_result)
|
||||
return False
|
||||
|
||||
# if contact_obj_in_result := sql_insert(data=contact_obj_update, table_name='contact', rm_id_random=True, id_random_length=8): pass
|
||||
# else:
|
||||
# return False
|
||||
|
||||
if organization_obj_update := person_obj_update.organization:
|
||||
log.debug(organization_obj_update)
|
||||
if organization_obj_up_result := update_organization_obj(organization_obj_update):
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
if person_obj_up.organization_id and person_obj_up.organization:
|
||||
organization_id = person_obj_up.organization_id
|
||||
organization_obj_up = person_obj_up.organization
|
||||
log.debug(organization_id)
|
||||
log.debug(organization_obj_up)
|
||||
if organization_obj_up_result := update_organization_obj(
|
||||
organization_id=organization_id,
|
||||
organization_obj_up=organization_obj_up,
|
||||
create_missing_obj=create_missing_obj,
|
||||
):
|
||||
log.debug(organization_obj_up_result)
|
||||
return True
|
||||
else:
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(organization_obj_up_result)
|
||||
return False
|
||||
else:
|
||||
if organization_obj_in_result := insert_organization_obj(organization_obj_insert):
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
elif person_obj_up.organization and not person_obj_up.organization.id:
|
||||
# NOTE: This will blindly create a new organization even if there was one associated but the person.organization_id was not found.
|
||||
organization_obj_in = person_obj_up.organization
|
||||
log.debug(organization_obj_in)
|
||||
if organization_obj_in_result := create_organization_obj(organization_obj_new=organization_obj_in):
|
||||
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(organization_obj_in_result)
|
||||
return True # NOTE: This needs to return the new organization ID
|
||||
person_obj_up.organization_id = organization_obj_in_result
|
||||
else:
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(organization_obj_in_result)
|
||||
return False
|
||||
|
||||
if user_obj_update := person_obj_update.user:
|
||||
log.debug(user_obj_update)
|
||||
if user_obj_up_result := update_user_obj(user_obj_update):
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(user_obj_up_result)
|
||||
return True
|
||||
else:
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(user_obj_up_result)
|
||||
return False
|
||||
else:
|
||||
if user_obj_in_result := insert_user_obj(user_obj_insert):
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(user_obj_in_result)
|
||||
return True # NOTE: This needs to return the new user ID
|
||||
else:
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(user_obj_in_result)
|
||||
return False
|
||||
|
||||
person_obj_update.pop('contact')
|
||||
person_obj_update.pop('organization')
|
||||
person_obj_update.pop('user')
|
||||
log.debug(person_obj_update)
|
||||
if person_obj_up_result = update_person_obj(person_obj_update):
|
||||
|
||||
if person_obj_up.user_id and person_obj_up.user:
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
user_id = person_obj_up.user_id
|
||||
user_obj_up = person_obj_up.user
|
||||
log.debug(user_id)
|
||||
log.debug(user_obj_up)
|
||||
if user_obj_up_result := update_user_obj(
|
||||
user_id=user_id,
|
||||
user_obj_up=user_obj_up,
|
||||
create_missing_obj=create_missing_obj,
|
||||
):
|
||||
log.debug(user_obj_up_result)
|
||||
else:
|
||||
log.debug(user_obj_up_result)
|
||||
return False
|
||||
elif person_obj_up.user and not person_obj_up.user.id:
|
||||
# NOTE: This will blindly create a new user even if there was one associated but the person.user_id was not found.
|
||||
user_obj_in = person_obj_up.user
|
||||
log.debug(user_obj_in)
|
||||
if user_obj_in_result := create_user_obj(user_obj_new=user_obj_in):
|
||||
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(user_obj_in_result)
|
||||
person_obj_up.user_id = user_obj_in_result
|
||||
else:
|
||||
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(user_obj_in_result)
|
||||
return False
|
||||
|
||||
person_dict_up = person_obj_up.dict(by_alias=False, exclude_unset=True, exclude={'contact', 'organization', 'user'})
|
||||
log.debug(person_dict_up)
|
||||
|
||||
if person_obj_up_result := sql_update(data=person_dict_up, table_name='person', rm_id_random=True):
|
||||
log.debug(person_obj_up_result)
|
||||
return True
|
||||
else:
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(person_obj_up_result)
|
||||
return False
|
||||
# ### END ### API Person Methods ### update_person_obj() ###
|
||||
|
||||
@@ -7,9 +7,9 @@ from pydantic import BaseModel, EmailStr, Field, PrivateAttr, ValidationError, v
|
||||
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.user_methods import load_user_obj
|
||||
from app.methods.organization_methods import load_organization_obj
|
||||
#from app.methods.person_methods import load_person_obj
|
||||
from app.methods.contact_methods import load_contact_obj, update_contact_obj
|
||||
from app.methods.organization_methods import load_organization_obj, update_organization_obj
|
||||
# from app.methods.person_methods import load_person_obj, update_person_obj
|
||||
|
||||
from app.models.user_models import User_Base, User_New_Base, User_Out_Base
|
||||
from app.models.user_role_models import User_Role_Base
|
||||
@@ -70,7 +70,13 @@ def create_user_obj(user_obj_new:User_New_Base) -> int|bool:
|
||||
|
||||
|
||||
# ### BEGIN ### API User Methods ### load_user_obj() ###
|
||||
def load_user_obj(user_id:int|str, inc_roles:bool=False, inc_user:bool=False, inc_organization:bool=False, inc_person:bool=False) -> User_Out_Base|bool:
|
||||
def load_user_obj(
|
||||
user_id: int|str,
|
||||
inc_contact: bool = False,
|
||||
inc_organization: bool = False,
|
||||
inc_person: bool = False,
|
||||
inc_user_role_li: bool = False,
|
||||
) -> User_Out_Base|bool:
|
||||
#log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(locals())
|
||||
|
||||
@@ -78,7 +84,7 @@ def load_user_obj(user_id:int|str, inc_roles:bool=False, inc_user:bool=False, in
|
||||
else: return False
|
||||
|
||||
if user_rec := sql_select(table_name='v_user', record_id=user_id): pass
|
||||
else: return false
|
||||
else: return False
|
||||
|
||||
#log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(user_rec)
|
||||
@@ -90,26 +96,16 @@ def load_user_obj(user_id:int|str, inc_roles:bool=False, inc_user:bool=False, in
|
||||
log.error(e.json())
|
||||
return False
|
||||
|
||||
if inc_roles:
|
||||
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
|
||||
|
||||
if inc_user:
|
||||
user_id = user_rec.get('user_id', None)
|
||||
if user_obj_result := load_user_obj(user_id=user_id):
|
||||
user_obj = user_obj_result
|
||||
user_rec['user'] = user_obj
|
||||
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.user = user_obj
|
||||
user_obj.contact = contact_obj
|
||||
else:
|
||||
user_rec['user'] = None
|
||||
user_obj.user = None
|
||||
user_rec['contact'] = None
|
||||
user_obj.contact = None
|
||||
|
||||
if inc_organization:
|
||||
organization_id = user_rec.get('organization_id', None)
|
||||
@@ -128,7 +124,21 @@ def load_user_obj(user_id:int|str, inc_roles:bool=False, inc_user:bool=False, in
|
||||
# person_obj = person_obj_result
|
||||
# user_rec['person'] = person_obj
|
||||
# log.debug(user_rec)
|
||||
# #else: user_rec['person'] = None
|
||||
# user_obj.person = person_obj
|
||||
# else:
|
||||
# user_rec['person'] = None
|
||||
# user_obj.person = None
|
||||
|
||||
# NOTE: Including user roles should probably be reviewed
|
||||
if inc_user_role_li:
|
||||
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)
|
||||
@@ -138,37 +148,106 @@ def load_user_obj(user_id:int|str, inc_roles:bool=False, inc_user:bool=False, in
|
||||
|
||||
|
||||
# ### BEGIN ### API User Methods ### update_user_obj() ###
|
||||
def update_user_obj(user_obj_update:User_Base):
|
||||
def update_user_obj(
|
||||
user_id: int|str, # Ideally the int ID should be passed. This allows for updating of the id_random value.
|
||||
user_obj_up: User_Base,
|
||||
create_missing_obj: bool = False,
|
||||
) -> bool:
|
||||
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(locals())
|
||||
|
||||
if contact_obj_update := user_obj_update.contact:
|
||||
log.debug(contact_obj_update)
|
||||
if contact_obj_up_result := update_contact_obj(contact_obj_update):
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(contact_obj_up_result)
|
||||
return True
|
||||
else:
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(contact_obj_up_result)
|
||||
return False
|
||||
else:
|
||||
if contact_obj_in_result := insert_contact_obj(contact_obj_insert):
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(contact_obj_in_result)c
|
||||
return True # NOTE: This needs to return the new contact ID
|
||||
else:
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(contact_obj_in_result)
|
||||
return False
|
||||
|
||||
user_obj_update.pop('contact')
|
||||
log.debug(user_obj_update)
|
||||
if user_obj_up_result = update_user_obj(user_obj_update):
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
if user_id := redis_lookup_id_random(record_id_random=user_id, table_name='user'): pass
|
||||
else: return False
|
||||
|
||||
user_obj_up.id = user_id
|
||||
|
||||
log.debug(user_obj_up)
|
||||
# log.debug(user_obj_up.dict(by_alias=True, exclude_unset=True))
|
||||
log.debug(user_obj_up.dict(by_alias=False, exclude_unset=True))
|
||||
# log.debug(user_obj_up.dict(by_alias=False, exclude_unset=False))
|
||||
|
||||
# if user_obj_up.contact_id and user_obj_up.contact:
|
||||
# contact_id = user_obj_up.contact_id
|
||||
# contact_obj_up = user_obj_up.contact
|
||||
# log.debug(contact_id)
|
||||
# log.debug(contact_obj_up)
|
||||
# if contact_obj_up_result := update_contact_obj(
|
||||
# contact_id=contact_id,
|
||||
# contact_obj_up=contact_obj_up,
|
||||
# create_missing_obj=create_missing_obj,
|
||||
# ):
|
||||
# log.debug(contact_obj_up_result)
|
||||
# else:
|
||||
# log.debug(contact_obj_up_result)
|
||||
# return False
|
||||
# elif user_obj_up.contact and not user_obj_up.contact.id:
|
||||
# # NOTE: This will blindly create a new contact even if there was one associated but the user.contact_id was not found.
|
||||
# contact_obj_in = user_obj_up.contact
|
||||
# log.debug(contact_obj_in)
|
||||
# if contact_obj_in_result := create_contact_obj(contact_obj_new=contact_obj_in):
|
||||
# # log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
# log.debug(contact_obj_in_result)
|
||||
# user_obj_up.contact_id = contact_obj_in_result
|
||||
# else:
|
||||
# # log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
# log.debug(contact_obj_in_result)
|
||||
# return False
|
||||
|
||||
# if organization_obj_update := user_obj_up.organization:
|
||||
# log.debug(organization_obj_update)
|
||||
# if organization_obj_up_result := update_organization_obj(organization_obj_update):
|
||||
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
# log.debug(organization_obj_up_result)
|
||||
# return True
|
||||
# else:
|
||||
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
# log.debug(organization_obj_up_result)
|
||||
# return False
|
||||
# else:
|
||||
# if organization_obj_in_result := insert_organization_obj(organization_obj_insert):
|
||||
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
# log.debug(organization_obj_in_result)
|
||||
# return True # NOTE: This needs to return the new organization ID
|
||||
# else:
|
||||
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
# log.debug(organization_obj_in_result)
|
||||
# return False
|
||||
|
||||
# if user_obj_up.person_id and user_obj_up.person:
|
||||
# person_id = user_obj_up.person_id
|
||||
# person_obj_up = user_obj_up.person
|
||||
# log.debug(person_id)
|
||||
# log.debug(person_obj_up)
|
||||
# if person_obj_up_result := update_person_obj(
|
||||
# person_id=person_id,
|
||||
# person_obj_up=person_obj_up,
|
||||
# create_missing_obj=create_missing_obj,
|
||||
# ):
|
||||
# log.debug(person_obj_up_result)
|
||||
# else:
|
||||
# log.debug(person_obj_up_result)
|
||||
# return False
|
||||
# elif user_obj_up.person and not user_obj_up.person.id:
|
||||
# # NOTE: This will blindly create a new person even if there was one associated but the user.person_id was not found.
|
||||
# person_obj_in = user_obj_up.person
|
||||
# log.debug(person_obj_in)
|
||||
# if person_obj_in_result := create_person_obj(person_obj_new=person_obj_in):
|
||||
# # log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
# log.debug(person_obj_in_result)
|
||||
# person_obj_up.person_id = person_obj_in_result
|
||||
# else:
|
||||
# # log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
# log.debug(person_obj_in_result)
|
||||
# return False
|
||||
|
||||
# IMPORTANT NOTE: Need to be extra careful if allowing an update to password, super, or manager. Maybe other fields?
|
||||
user_dict_up = user_obj_up.dict(by_alias=False, exclude_unset=True, exclude={'password', 'super', 'manager', 'contact', 'organization', 'person'})
|
||||
log.debug(user_dict_up)
|
||||
|
||||
if user_obj_up_result := sql_update(data=user_dict_up, table_name='user', rm_id_random=True):
|
||||
log.debug(user_obj_up_result)
|
||||
return True
|
||||
else:
|
||||
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(user_obj_up_result)
|
||||
return False
|
||||
# ### END ### API User Methods ### update_user_obj() ###
|
||||
|
||||
@@ -87,7 +87,7 @@ class Event_Session_Base(BaseModel):
|
||||
updated_on: Optional[datetime.datetime] = None
|
||||
|
||||
# Including other related objects
|
||||
#event: Optional[Event_Base]
|
||||
# event: Optional[Event_Base]
|
||||
event_abstract_list: Optional[list] # Optional[Event_Abstract_Base]
|
||||
event_badge_list: Optional[list] # Optional[Event_Abstract_Base]
|
||||
event_device_list: Optional[list] # Optional[Event_Device_Base]
|
||||
|
||||
@@ -232,6 +232,7 @@ class User_Base(BaseModel):
|
||||
email: Optional[str]
|
||||
email_verified: Optional[bool]
|
||||
password: Optional[str]
|
||||
new_password: Optional[str]
|
||||
|
||||
allow_auth_key: Optional[int]
|
||||
auth_key: Optional[str]
|
||||
|
||||
@@ -9,7 +9,9 @@ from app.lib_general import log, logging
|
||||
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.routers.api_crud import delete_obj_template, get_obj_template, get_obj_li_template, patch_obj_template, post_obj_template
|
||||
|
||||
from app.methods.address_methods import load_address_obj, update_address_obj
|
||||
|
||||
from app.models.address_models import Address_Base
|
||||
from app.models.response_models import *
|
||||
@@ -74,6 +76,45 @@ async def patch_address_obj(
|
||||
return result
|
||||
|
||||
|
||||
# ### BEGIN ### API Address ### patch_address_json() ###
|
||||
@router.patch('/{address_id}/json', response_model=Resp_Body_Base)
|
||||
async def patch_address_json(
|
||||
address_obj: Address_Base,
|
||||
address_id: str = Query(..., min_length=1, max_length=22),
|
||||
create_missing_obj: bool = False,
|
||||
x_account_id: Optional[str] = Header(..., ),
|
||||
return_obj: Optional[bool] = True,
|
||||
by_alias: Optional[bool] = True,
|
||||
include: Optional[list] = [],
|
||||
exclude: Optional[list] = [],
|
||||
exclude_unset: Optional[bool] = True,
|
||||
exclude_none: Optional[bool] = True,
|
||||
):
|
||||
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(locals())
|
||||
|
||||
if address_id := redis_lookup_id_random(record_id_random=address_id, table_name='address'): pass
|
||||
else:
|
||||
return mk_resp(data=None, status_code=404)
|
||||
|
||||
if address_obj_up_result := update_address_obj(
|
||||
address_id=address_id,
|
||||
address_obj_up=address_obj,
|
||||
create_missing_obj=create_missing_obj,
|
||||
):
|
||||
|
||||
log.debug(address_obj_up_result)
|
||||
if return_obj:
|
||||
address_obj = load_address_obj(address_id=address_id)
|
||||
address_dict = address_obj.dict(by_alias=by_alias, exclude_unset=exclude_unset)
|
||||
return mk_resp(data=address_dict)
|
||||
else:
|
||||
return mk_resp(data=address_obj_up_result)
|
||||
else:
|
||||
return mk_resp(data=False, status_code=400) # Bad Request
|
||||
# ### END ### API Address ### patch_address_json() ###
|
||||
|
||||
|
||||
@router.get('/list', response_model=Resp_Body_Base)
|
||||
async def get_address_obj_li(
|
||||
for_obj_type: Optional[str] = Query(None, min_length=2, max_length=50),
|
||||
|
||||
@@ -11,6 +11,8 @@ 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.contact_methods import load_contact_obj, update_contact_obj
|
||||
|
||||
from app.models.contact_models import Contact_Base
|
||||
from app.models.response_models import *
|
||||
|
||||
@@ -68,6 +70,45 @@ async def patch_contact_obj(
|
||||
return result
|
||||
|
||||
|
||||
# ### BEGIN ### API Contact ### patch_contact_json() ###
|
||||
@router.patch('/{contact_id}/json', response_model=Resp_Body_Base)
|
||||
async def patch_contact_json(
|
||||
contact_obj: Contact_Base,
|
||||
contact_id: str = Query(..., min_length=1, max_length=22),
|
||||
create_missing_obj: bool = False,
|
||||
x_account_id: Optional[str] = Header(..., ),
|
||||
return_obj: Optional[bool] = True,
|
||||
by_alias: Optional[bool] = True,
|
||||
include: Optional[list] = [],
|
||||
exclude: Optional[list] = [],
|
||||
exclude_unset: Optional[bool] = True,
|
||||
exclude_none: Optional[bool] = True,
|
||||
):
|
||||
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(locals())
|
||||
|
||||
if contact_id := redis_lookup_id_random(record_id_random=contact_id, table_name='contact'): pass
|
||||
else:
|
||||
return mk_resp(data=None, status_code=404)
|
||||
|
||||
if contact_obj_up_result := update_contact_obj(
|
||||
contact_id=contact_id,
|
||||
contact_obj_up=contact_obj,
|
||||
create_missing_obj=create_missing_obj,
|
||||
):
|
||||
|
||||
log.debug(contact_obj_up_result)
|
||||
if return_obj:
|
||||
contact_obj = load_contact_obj(contact_id=contact_id)
|
||||
contact_dict = contact_obj.dict(by_alias=by_alias, exclude_unset=exclude_unset)
|
||||
return mk_resp(data=contact_dict)
|
||||
else:
|
||||
return mk_resp(data=contact_obj_up_result)
|
||||
else:
|
||||
return mk_resp(data=False, status_code=400) # Bad Request
|
||||
# ### END ### API Contact ### patch_contact_json() ###
|
||||
|
||||
|
||||
@router.get('/list', response_model=Resp_Body_Base)
|
||||
async def get_contact_obj_li(
|
||||
for_obj_type: str = Query(None, min_length=2, max_length=50),
|
||||
|
||||
@@ -1,11 +1,9 @@
|
||||
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 *
|
||||
|
||||
|
||||
@@ -5,7 +5,7 @@ from pydantic import BaseModel, EmailStr, Field
|
||||
from typing import Dict, List, Optional, Set, Union
|
||||
|
||||
from app.lib_general import *
|
||||
from ..log import *
|
||||
# from ..log import *
|
||||
from app.config import settings
|
||||
from app.db_sql import *
|
||||
|
||||
|
||||
@@ -11,7 +11,7 @@ from app.db_sql import *
|
||||
|
||||
from app.routers.api_crud import delete_obj_template, get_obj_template, get_obj_li_template, patch_obj_template, post_obj_template
|
||||
|
||||
from app.methods.person_methods import load_person_obj
|
||||
from app.methods.person_methods import load_person_obj, update_person_obj
|
||||
|
||||
from app.models.person_models import Person_Base
|
||||
from app.models.response_models import *
|
||||
@@ -70,6 +70,45 @@ async def patch_person_obj(
|
||||
return result
|
||||
|
||||
|
||||
# ### BEGIN ### API Person ### patch_person_json() ###
|
||||
@router.patch('/{person_id}/json', response_model=Resp_Body_Base)
|
||||
async def patch_person_json(
|
||||
person_obj: Person_Base,
|
||||
person_id: str = Query(..., min_length=1, max_length=22),
|
||||
create_missing_obj: bool = False,
|
||||
x_account_id: Optional[str] = Header(..., ),
|
||||
return_obj: Optional[bool] = True,
|
||||
by_alias: Optional[bool] = True,
|
||||
include: Optional[list] = [],
|
||||
exclude: Optional[list] = [],
|
||||
exclude_unset: Optional[bool] = True,
|
||||
exclude_none: Optional[bool] = True,
|
||||
):
|
||||
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
||||
log.debug(locals())
|
||||
|
||||
if person_id := redis_lookup_id_random(record_id_random=person_id, table_name='person'): pass
|
||||
else:
|
||||
return mk_resp(data=None, status_code=404)
|
||||
|
||||
if person_obj_up_result := update_person_obj(
|
||||
person_id=person_id,
|
||||
person_obj_up=person_obj,
|
||||
create_missing_obj=create_missing_obj,
|
||||
):
|
||||
|
||||
log.debug(person_obj_up_result)
|
||||
if return_obj:
|
||||
person_obj = load_person_obj(person_id=person_id)
|
||||
person_dict = person_obj.dict(by_alias=by_alias, exclude_unset=exclude_unset)
|
||||
return mk_resp(data=person_dict)
|
||||
else:
|
||||
return mk_resp(data=person_obj_up_result)
|
||||
else:
|
||||
return mk_resp(data=False, status_code=400) # Bad Request
|
||||
# ### END ### API Person ### patch_person_json() ###
|
||||
|
||||
|
||||
@router.get('/list', response_model=Resp_Body_Base)
|
||||
async def get_person_obj_li(
|
||||
for_obj_type: Optional[str] = Query(None, min_length=2, max_length=50),
|
||||
|
||||
Reference in New Issue
Block a user