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

Working on membership, person, and user

Browse Source
This commit is contained in:
Scott Idem
2021-06-25 18:49:08 -04:00
parent ca43cc4dce
commit f55d9c2c62
10 changed files with 768 additions and 333 deletions
Download Patch File Download Diff File Expand all files Collapse all files

162
app/methods/address_methods.py
View File

@@ -8,29 +8,7 @@ from app.db_sql import redis_lookup_id_random, sql_insert, sql_select, sql_updat
from app.lib_general import log, logging
from app.models.address_models import Address_Base
# ### BEGIN ### API Address Methods ### create_address_obj() ###
def create_address_obj(address_obj_new:Address_Base):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if not address_obj_new:
return False
address_obj_data = address_obj_new.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'created_on', 'updated_on'})
if address_obj_in_result := sql_insert(data=address_obj_data, table_name='address', rm_id_random=True, id_random_length=8): pass
else:
return False
log.debug(address_obj_in_result)
address_id = address_obj_in_result
log.debug(f'Returning the new address_id: {address_id}')
return address_id
# ### END ### API Address Methods ### create_address_obj() ###
from app.models.common_field_schema import default_num_bytes
# ### BEGIN ### API Address Methods ### load_address_obj() ###
@@ -65,36 +43,6 @@ def load_address_obj(
# ### END ### API Address Methods ### load_address_obj() ###
# ### BEGIN ### API Address Methods ### update_address_obj() ###
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())
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.debug(address_obj_up_result)
return False
# ### END ### API Address Methods ### update_address_obj() ###
# ### BEGIN ### API Address Methods ### get_address_rec_list() ###
def get_address_rec_list(
for_obj_type: str,
@@ -148,3 +96,111 @@ def get_address_rec_list(
return address_rec_li
# ### END ### API Address Methods ### get_address_rec_list() ###
# ### BEGIN ### API Address Methods ### create_address_obj() ###
def create_address_obj(address_obj_new:Address_Base):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if not address_obj_new:
return False
address_obj_data = address_obj_new.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'created_on', 'updated_on'})
if address_obj_in_result := sql_insert(data=address_obj_data, table_name='address', rm_id_random=True, id_random_length=8): pass
else:
return False
log.debug(address_obj_in_result)
address_id = address_obj_in_result
log.debug(f'Returning the new address_id: {address_id}')
return address_id
# ### END ### API Address Methods ### create_address_obj() ###
# ### BEGIN ### API Address Methods ### update_address_obj() ###
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())
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.debug(address_obj_up_result)
return False
# ### END ### API Address Methods ### update_address_obj() ###
# ### BEGIN ### API Address Methods ### create_update_address_obj() ###
def create_update_address_obj(
address_id: int|str|None, # Ideally the int ID should be passed. This allows for updating of the id_random value.
address_obj: Address_Base,
process_address: bool = False,
process_organization: bool = False,
) -> bool:
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if address_id:
if address_id := redis_lookup_id_random(record_id_random=address_id, table_name='address'): pass
else: return False
address_obj.id = address_id
else:
# Insert record now and update later
address_dict_in = address_obj.dict(by_alias=False, exclude_unset=True, exclude={'address', 'organization', 'user'})
log.debug(address_dict_in)
address_in_result = sql_insert(
data = address_dict_in,
table_name = 'address',
rm_id_random = True,
id_random_length = default_num_bytes,
)
log.debug(address_in_result)
if isinstance(address_in_result, bool) and address_in_result is True:
return address_in_result
elif isinstance(address_in_result, int):
address_id = address_in_result
address_obj.id = address_id
else:
return False # This should not happen.
# Process address data
address_dict_up = address_obj.dict(by_alias=False, exclude_unset=True)
log.debug(address_dict_up)
# Update record
address_up_result = sql_update(
data = address_dict_up,
table_name = 'address',
rm_id_random = True,
)
log.debug(address_up_result)
if isinstance(address_up_result, bool) and address_up_result is True:
return address_id
elif isinstance(address_up_result, bool) and address_up_result is False:
return False
elif isinstance(address_up_result, int):
return address_up_result
else:
return False
# ### END ### API Address Methods ### create_update_address_obj() ###

256
app/methods/contact_methods.py
View File

@@ -7,53 +7,12 @@ 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, update_address_obj
from app.methods.address_methods import create_address_obj, create_update_address_obj, update_address_obj
from app.models.common_field_schema import default_num_bytes
from app.models.contact_models import Contact_Base
# ### BEGIN ### API Contact Methods ### create_contact_obj() ###
def create_contact_obj(contact_obj_new:Contact_Base):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if not contact_obj_new:
return False
contact_obj_data = contact_obj_new.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'address', 'created_on', 'updated_on'})
if contact_obj_in_result := sql_insert(data=contact_obj_data, table_name='contact', rm_id_random=True, id_random_length=8): pass
else:
return False
log.debug(contact_obj_in_result)
contact_id = contact_obj_in_result
if contact_obj_new.address:
address_obj_new = contact_obj_new.address
address_obj_new.for_type = 'contact'
address_obj_new.for_id = contact_id
create_address_obj_result = create_address_obj(address_obj_new=address_obj_new)
if isinstance(create_address_obj_result, int):
address_id = create_address_obj_result
# Need to update the contact with the new address_id
contact_obj_up = {}
contact_obj_up['id'] = contact_id
contact_obj_up['address_id'] = address_id
if contact_obj_up_result := sql_update(data=contact_obj_up, table_name='contact'): pass
else:
return False
log.debug(contact_obj_up_result)
else:
log.debug(f'No address_id was returned when tyring to create_address_obj(): {create_address_obj_result}')
address_id = None
log.debug(f'Returning the new contact_id: {contact_id}')
return contact_id
# ### END ### API Contact Methods ### create_contact_obj() ###
# ### BEGIN ### API Contact Methods ### load_contact_obj() ###
def load_contact_obj(
contact_id:int|str,
@@ -101,6 +60,103 @@ def load_contact_obj(
# ### END ### API Contact Methods ### load_contact_obj() ###
# ### BEGIN ### API Contact Methods ### get_contact_rec_list() ###
def get_contact_rec_list(
for_obj_type: str,
for_obj_id: str,
limit: int = 1000,
enabled: str = 'enabled', # enabled, disabled, all
) -> list|bool:
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if for_obj_id := redis_lookup_id_random(record_id_random=for_obj_id, table_name='for_obj_type'): pass
else: return False
data = {}
data[f'{for_obj_type}_id'] = for_obj_id
# data['for_obj_type'] = for_obj_type
sql_obj_type_id = f'`tbl`.{for_obj_type}_id = :{for_obj_type}_id'
# if enabled in ['enabled', 'disabled', 'all']:
# if enabled == 'enabled':
# data['enable'] = True
# sql_enabled = f'AND `tbl`.enable = :enable'
# elif enabled == 'disabled':
# data['enable'] = False
# sql_enabled = f'AND `tbl`.enable = :enable'
# elif enabled == 'all':
# sql_enabled = ''
sql_enabled = ''
if limit:
data['limit'] = limit
sql_limit = f'LIMIT :limit'
else:
sql_limit = ''
sql = f"""
SELECT `tbl`.id AS 'contact_id', `tbl`.id_random AS 'contact_id_random'
FROM `contact` AS `tbl`
WHERE
{sql_obj_type_id}
{sql_enabled}
ORDER BY `tbl`.created_on DESC, `tbl`.updated_on DESC
{sql_limit};
"""
if contact_rec_li_result := sql_select(data=data, sql=sql, as_list=True):
contact_rec_li = contact_rec_li_result
else:
contact_rec_li = []
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(contact_rec_li_result)
return contact_rec_li
# ### END ### API Contact Methods ### get_contact_rec_list() ###
# ### BEGIN ### API Contact Methods ### create_contact_obj() ###
def create_contact_obj(contact_obj_new:Contact_Base):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if not contact_obj_new:
return False
contact_obj_data = contact_obj_new.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'address', 'created_on', 'updated_on'})
if contact_obj_in_result := sql_insert(data=contact_obj_data, table_name='contact', rm_id_random=True, id_random_length=8): pass
else:
return False
log.debug(contact_obj_in_result)
contact_id = contact_obj_in_result
if contact_obj_new.address:
address_obj_new = contact_obj_new.address
address_obj_new.for_type = 'contact'
address_obj_new.for_id = contact_id
create_address_obj_result = create_address_obj(address_obj_new=address_obj_new)
if isinstance(create_address_obj_result, int):
address_id = create_address_obj_result
# Need to update the contact with the new address_id
contact_obj_up = {}
contact_obj_up['id'] = contact_id
contact_obj_up['address_id'] = address_id
if contact_obj_up_result := sql_update(data=contact_obj_up, table_name='contact'): pass
else:
return False
log.debug(contact_obj_up_result)
else:
log.debug(f'No address_id was returned when tyring to create_address_obj(): {create_address_obj_result}')
address_id = None
log.debug(f'Returning the new contact_id: {contact_id}')
return contact_id
# ### END ### API Contact Methods ### create_contact_obj() ###
# ### BEGIN ### API Contact Methods ### update_contact_obj() ###
def update_contact_obj(
contact_id: int|str, # Ideally the int ID should be passed. This allows for updating of the id_random value.
@@ -161,56 +217,76 @@ def update_contact_obj(
# ### END ### API Contact Methods ### update_contact_obj() ###
# ### BEGIN ### API Contact Methods ### get_contact_rec_list() ###
def get_contact_rec_list(
for_obj_type: str,
for_obj_id: str,
limit: int = 1000,
enabled: str = 'enabled', # enabled, disabled, all
) -> list|bool:
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
# ### BEGIN ### API Contact Methods ### create_update_contact_obj() ###
def create_update_contact_obj(
contact_id: int|str|None, # Ideally the int ID should be passed. This allows for updating of the id_random value.
contact_obj: Contact_Base,
process_address: bool = False,
process_organization: bool = False,
) -> bool:
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if for_obj_id := redis_lookup_id_random(record_id_random=for_obj_id, table_name='for_obj_type'): pass
else: return False
data = {}
data[f'{for_obj_type}_id'] = for_obj_id
# data['for_obj_type'] = for_obj_type
sql_obj_type_id = f'`tbl`.{for_obj_type}_id = :{for_obj_type}_id'
# if enabled in ['enabled', 'disabled', 'all']:
# if enabled == 'enabled':
# data['enable'] = True
# sql_enabled = f'AND `tbl`.enable = :enable'
# elif enabled == 'disabled':
# data['enable'] = False
# sql_enabled = f'AND `tbl`.enable = :enable'
# elif enabled == 'all':
# sql_enabled = ''
sql_enabled = ''
if limit:
data['limit'] = limit
sql_limit = f'LIMIT :limit'
if contact_id:
if contact_id := redis_lookup_id_random(record_id_random=contact_id, table_name='contact'): pass
else: return False
contact_obj.id = contact_id
else:
sql_limit = ''
# Insert record now and update later
contact_dict_in = contact_obj.dict(by_alias=False, exclude_unset=True, exclude={'contact', 'organization', 'user'})
log.debug(contact_dict_in)
contact_in_result = sql_insert(
data = contact_dict_in,
table_name = 'contact',
rm_id_random = True,
id_random_length = default_num_bytes,
)
log.debug(contact_in_result)
if isinstance(contact_in_result, bool) and contact_in_result is True:
return contact_in_result
elif isinstance(contact_in_result, int):
contact_id = contact_in_result
contact_obj.id = contact_id
else:
return False # This should not happen.
sql = f"""
SELECT `tbl`.id AS 'contact_id', `tbl`.id_random AS 'contact_id_random'
FROM `contact` AS `tbl`
WHERE
{sql_obj_type_id}
{sql_enabled}
ORDER BY `tbl`.created_on DESC, `tbl`.updated_on DESC
{sql_limit};
"""
# Process address data
if process_address and contact_obj.address:
address_obj = contact_obj.address
address_obj.for_type = 'contact'
address_obj.for_id = contact_id
address_id = contact_obj.address_id_random
address_result = create_update_address_obj(
address_id = address_id,
address_obj = address_obj,
)
log.debug(address_result)
if isinstance(address_result, bool) and address_result is True:
pass # Do not need to update contact object.
elif isinstance(address_result, bool) and address_result is False:
pass # Do not need to update contact object.
elif isinstance(address_result, int):
contact_obj.address_id = address_result
else:
log.warning('Something may have gone wrong while trying to create or update a address.')
if contact_rec_li_result := sql_select(data=data, sql=sql, as_list=True):
contact_rec_li = contact_rec_li_result
# Process contact data
contact_dict_up = contact_obj.dict(by_alias=False, exclude_unset=True, exclude={'address'})
log.debug(contact_dict_up)
# Update record
contact_up_result = sql_update(
data = contact_dict_up,
table_name = 'contact',
rm_id_random = True,
)
log.debug(contact_up_result)
if isinstance(contact_up_result, bool) and contact_up_result is True:
return contact_id
elif isinstance(contact_up_result, bool) and contact_up_result is False:
return False
elif isinstance(contact_up_result, int):
return contact_up_result
else:
contact_rec_li = []
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(contact_rec_li_result)
return contact_rec_li
# ### END ### API Contact Methods ### get_contact_rec_list() ###
return False
# ### END ### API Contact Methods ### create_update_contact_obj() ###

176
app/methods/organization_methods.py
View File

@@ -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.address_methods import load_address_obj
from app.methods.contact_methods import create_contact_obj, load_contact_obj, update_contact_obj
from app.methods.contact_methods import create_contact_obj, create_update_contact_obj, load_contact_obj, update_contact_obj
from app.models.common_field_schema import default_num_bytes
from app.models.organization_models import Organization_Base
@@ -66,54 +66,6 @@ def load_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() ###
# ### BEGIN ### API Organization Methods ### get_organization_rec_list() ###
def get_organization_rec_list(
for_obj_type: str,
@@ -167,3 +119,127 @@ def get_organization_rec_list(
return organization_rec_li
# ### END ### API Organization Methods ### get_organization_rec_list() ###
# ### 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() ###
# ### BEGIN ### API Organization Methods ### create_update_organization_obj() ###
def create_update_organization_obj(
organization_id: int|str|None, # Ideally the int ID should be passed. This allows for updating of the id_random value.
organization_obj: Organization_Base,
process_contact: bool = False,
) -> bool:
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if organization_id:
if organization_id := redis_lookup_id_random(record_id_random=organization_id, table_name='organization'): pass
else: return False
organization_obj.id = organization_id
else:
# Insert record now and update later
organization_dict_in = organization_obj.dict(by_alias=False, exclude_unset=True, exclude={'contact', 'person', 'user'})
log.debug(organization_dict_in)
organization_in_result = sql_insert(
data = organization_dict_in,
table_name = 'organization',
rm_id_random = True,
id_random_length = default_num_bytes,
)
log.debug(organization_in_result)
if isinstance(organization_in_result, bool) and organization_in_result is True:
return organization_in_result
elif isinstance(organization_in_result, int):
organization_id = organization_in_result
organization_obj.id = organization_id
else:
return False # This should not happen.
# Process contact data
if process_contact and organization_obj.contact:
contact_obj = organization_obj.contact
contact_obj.for_type = 'organization'
contact_obj.for_id = organization_id
contact_id = organization_obj.contact_id_random
contact_result = create_update_contact_obj(
contact_id = contact_id,
contact_obj = contact_obj,
process_address = True, # Setting to True under the assumption that if there is contact information then there is probably an address.
)
log.debug(contact_result)
if isinstance(contact_result, bool) and contact_result is True:
pass # Do not need to update organization object.
elif isinstance(contact_result, bool) and contact_result is False:
pass # Do not need to update organization object.
elif isinstance(contact_result, int):
organization_obj.contact_id = contact_result
# pass # Do not need to update organization object.
else:
log.warning('Something may have gone wrong while trying to create or update a contact.')
# Process organization data
organization_dict_up = organization_obj.dict(by_alias=False, exclude_unset=True, exclude={'contact', 'person', 'user'})
log.debug(organization_dict_up)
# Update record
organization_up_result = sql_update(
data = organization_dict_up,
table_name = 'organization',
rm_id_random = True,
)
log.debug(organization_up_result)
if isinstance(organization_up_result, bool) and organization_up_result is True:
return organization_id
elif isinstance(organization_up_result, bool) and organization_up_result is False:
return False
elif isinstance(organization_up_result, int):
return organization_up_result
else:
return False
# ### END ### API Organization Methods ### create_update_organization_obj() ###

287
app/methods/person_methods.py
View File

@@ -4,62 +4,19 @@ 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.db_sql import redis_lookup_id_random, sql_insert, sql_insert_or_update, sql_select, sql_update
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.address_methods import load_address_obj
from app.methods.contact_methods import create_contact_obj, create_update_contact_obj, load_contact_obj, update_contact_obj
from app.methods.order_methods import load_order_obj, get_order_rec_list
from app.methods.organization_methods import load_organization_obj, update_organization_obj
from app.methods.organization_methods import create_update_organization_obj, load_organization_obj, update_organization_obj
# from app.methods.user_methods import create_user_obj, load_user_obj, update_user_obj
from app.models.common_field_schema import default_num_bytes
from app.models.person_models import Person_Base
# ### BEGIN ### API Person Methods ### create_person_obj() ###
def create_person_obj(person_obj_new:Person_Base):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if not person_obj_new:
return False
person_obj_data = person_obj_new.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'contact', 'organization', 'created_on', 'updated_on'})
if person_obj_in_result := sql_insert(data=person_obj_data, table_name='person', rm_id_random=True, id_random_length=8): pass
else:
return False
#log.setLevel(logging.DEBUG)
log.debug(person_obj_in_result)
person_id = person_obj_in_result
if person_obj_new.contact:
contact_obj_new = person_obj_new.contact
contact_obj_new.for_type = 'person'
contact_obj_new.for_id = person_id
create_contact_obj_result = create_contact_obj(contact_obj_new=contact_obj_new)
if isinstance(create_contact_obj_result, int):
contact_id = create_contact_obj_result
log.debug(f'Update person with new contact_id: {contact_id}')
# Need to update the person with the new contact_id
person_obj_up = {}
person_obj_up['id'] = person_id
person_obj_up['contact_id'] = contact_id
if person_obj_up_result := sql_update(data=person_obj_up, table_name='person'): pass
else:
return False
log.debug(person_obj_up_result)
else:
log.debug(f'No contact_id was returned when tyring to create_contact_obj(): {create_contact_obj_result}')
contact_id = None
log.debug(f'Returning the new person_id: {person_id}')
return person_id
# ### END ### API Person Methods ### create_person_obj() ###
# ### BEGIN ### API Person Methods ### load_person_obj() ###
def load_person_obj(
person_id: int|str,
@@ -180,6 +137,105 @@ def load_person_obj(
# ### END ### API Person Methods ### load_person_obj() ###
# ### BEGIN ### API Person Methods ### get_person_rec_list() ###
def get_person_rec_list(
for_obj_type: str,
for_obj_id: str,
limit: int = 1000,
enabled: str = 'enabled', # enabled, disabled, all
) -> list|bool:
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if for_obj_id := redis_lookup_id_random(record_id_random=for_obj_id, table_name='for_obj_type'): pass
else: return False
data = {}
data[f'{for_obj_type}_id'] = for_obj_id
# data['for_obj_type'] = for_obj_type
sql_obj_type_id = f'`tbl`.{for_obj_type}_id = :{for_obj_type}_id'
# if enabled in ['enabled', 'disabled', 'all']:
# if enabled == 'enabled':
# data['enable'] = True
# sql_enabled = f'AND `tbl`.enable = :enable'
# elif enabled == 'disabled':
# data['enable'] = False
# sql_enabled = f'AND `tbl`.enable = :enable'
# elif enabled == 'all':
# sql_enabled = ''
sql_enabled = ''
if limit:
data['limit'] = limit
sql_limit = f'LIMIT :limit'
else:
sql_limit = ''
sql = f"""
SELECT `tbl`.id AS 'person_id', `tbl`.id_random AS 'person_id_random'
FROM `person` AS `tbl`
WHERE
{sql_obj_type_id}
{sql_enabled}
ORDER BY `tbl`.created_on DESC, `tbl`.updated_on DESC
{sql_limit};
"""
if person_rec_li_result := sql_select(data=data, sql=sql, as_list=True):
person_rec_li = person_rec_li_result
else:
person_rec_li = []
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(person_rec_li_result)
return person_rec_li
# ### END ### API Person Methods ### get_person_rec_list() ###
# ### BEGIN ### API Person Methods ### create_person_obj() ###
def create_person_obj(person_obj_new:Person_Base):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if not person_obj_new:
return False
person_obj_data = person_obj_new.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'contact', 'organization', 'created_on', 'updated_on'})
if person_obj_in_result := sql_insert(data=person_obj_data, table_name='person', rm_id_random=True, id_random_length=8): pass
else:
return False
#log.setLevel(logging.DEBUG)
log.debug(person_obj_in_result)
person_id = person_obj_in_result
if person_obj_new.contact:
contact_obj_new = person_obj_new.contact
contact_obj_new.for_type = 'person'
contact_obj_new.for_id = person_id
create_contact_obj_result = create_contact_obj(contact_obj_new=contact_obj_new)
if isinstance(create_contact_obj_result, int):
contact_id = create_contact_obj_result
log.debug(f'Update person with new contact_id: {contact_id}')
# Need to update the person with the new contact_id
person_obj_up = {}
person_obj_up['id'] = person_id
person_obj_up['contact_id'] = contact_id
if person_obj_up_result := sql_update(data=person_obj_up, table_name='person'): pass
else:
return False
log.debug(person_obj_up_result)
else:
log.debug(f'No contact_id was returned when tyring to create_contact_obj(): {create_contact_obj_result}')
contact_id = None
log.debug(f'Returning the new person_id: {person_id}')
return person_id
# ### END ### API Person Methods ### create_person_obj() ###
# ### BEGIN ### API Person Methods ### update_person_obj() ###
def update_person_obj(
person_id: int|str, # Ideally the int ID should be passed. This allows for updating of the id_random value.
@@ -293,56 +349,99 @@ def update_person_obj(
# ### END ### API Person Methods ### update_person_obj() ###
# ### BEGIN ### API Person Methods ### get_person_rec_list() ###
def get_person_rec_list(
for_obj_type: str,
for_obj_id: str,
limit: int = 1000,
enabled: str = 'enabled', # enabled, disabled, all
) -> list|bool:
# ### BEGIN ### API Person Methods ### create_update_person_obj() ###
def create_update_person_obj(
person_id: int|str|None, # Ideally the int ID should be passed. This allows for updating of the id_random value.
person_obj: Person_Base,
process_contact: bool = False,
process_organization: bool = False,
) -> bool:
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if for_obj_id := redis_lookup_id_random(record_id_random=for_obj_id, table_name='for_obj_type'): pass
else: return False
data = {}
data[f'{for_obj_type}_id'] = for_obj_id
# data['for_obj_type'] = for_obj_type
sql_obj_type_id = f'`tbl`.{for_obj_type}_id = :{for_obj_type}_id'
# if enabled in ['enabled', 'disabled', 'all']:
# if enabled == 'enabled':
# data['enable'] = True
# sql_enabled = f'AND `tbl`.enable = :enable'
# elif enabled == 'disabled':
# data['enable'] = False
# sql_enabled = f'AND `tbl`.enable = :enable'
# elif enabled == 'all':
# sql_enabled = ''
sql_enabled = ''
if limit:
data['limit'] = limit
sql_limit = f'LIMIT :limit'
if person_id:
if person_id := redis_lookup_id_random(record_id_random=person_id, table_name='person'): pass
else: return False
person_obj.id = person_id
else:
sql_limit = ''
# Insert record now and update later
person_dict_in = person_obj.dict(by_alias=False, exclude_unset=True, exclude={'contact', 'organization', 'user'})
log.debug(person_dict_in)
person_in_result = sql_insert(
data = person_dict_in,
table_name = 'person',
rm_id_random = True,
id_random_length = default_num_bytes,
)
log.debug(person_in_result)
if isinstance(person_in_result, bool) and person_in_result is True:
return person_in_result
elif isinstance(person_in_result, int):
person_id = person_in_result
person_obj.id = person_id
else:
return False # This should not happen.
sql = f"""
SELECT `tbl`.id AS 'person_id', `tbl`.id_random AS 'person_id_random'
FROM `person` AS `tbl`
WHERE
{sql_obj_type_id}
{sql_enabled}
ORDER BY `tbl`.created_on DESC, `tbl`.updated_on DESC
{sql_limit};
"""
# Process contact data
if process_contact and person_obj.contact:
contact_obj = person_obj.contact
contact_obj.for_type = 'person'
contact_obj.for_id = person_id
contact_id = person_obj.contact_id_random
contact_result = create_update_contact_obj(
contact_id = contact_id,
contact_obj = contact_obj,
process_address = True, # Setting to True under the assumption that if there is contact information then there is probably an address.
)
log.debug(contact_result)
if isinstance(contact_result, bool) and contact_result is True:
pass # Do not need to update person object.
elif isinstance(contact_result, bool) and contact_result is False:
pass # Do not need to update person object.
elif isinstance(contact_result, int):
person_obj.contact_id = contact_result
# pass # Do not need to update person object.
else:
log.warning('Something may have gone wrong while trying to create or update a contact.')
if person_rec_li_result := sql_select(data=data, sql=sql, as_list=True):
person_rec_li = person_rec_li_result
# Process organization data
if process_organization and person_obj.organization:
organization_obj = person_obj.organization
organization_id = person_obj.organization_id_random
organization_result = create_update_organization_obj(
organization_id = organization_id,
organization_obj = organization_obj,
process_contact = True, # Setting to True under the assumption that if there is organization information then there is probably a contact (and address).
)
log.debug(organization_result)
if isinstance(organization_result, bool) and organization_result is True:
pass # Do not need to update person object.
elif isinstance(organization_result, bool) and organization_result is False:
pass # Do not need to update person object.
elif isinstance(organization_result, int):
person_obj.organization_id = organization_result
else:
log.warning('Something may have gone wrong while trying to create or update a organization.')
# Process person data
person_dict_up = person_obj.dict(by_alias=False, exclude_unset=True, exclude={'contact', 'organization', 'user'})
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(person_obj)
log.debug(person_dict_up)
# Update record
person_up_result = sql_update(
data = person_dict_up,
table_name = 'person',
rm_id_random = True,
)
log.debug(person_up_result)
if isinstance(person_up_result, bool) and person_up_result is True:
return person_id
elif isinstance(person_up_result, bool) and person_up_result is False:
return False
elif isinstance(person_up_result, int):
return person_up_result
else:
person_rec_li = []
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(person_rec_li_result)
return person_rec_li
# ### END ### API Person Methods ### get_person_rec_list() ###
return False
# ### END ### API Person Methods ### create_update_person_obj() ###

10
app/methods/user_methods.py
View File

@@ -108,7 +108,6 @@ def load_user_obj(
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:
@@ -234,7 +233,6 @@ def load_user_obj(
# Updated 2021-06-25
if inc_user_role_list:
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
if user_role_rec_list_result := get_user_role_rec_list(
for_obj_type = 'user',
for_obj_id = user_id,
@@ -242,7 +240,6 @@ def load_user_obj(
enabled = enabled,
):
user_role_result_list = []
log.debug(user_role_rec_list_result)
for user_role_rec in user_role_rec_list_result:
if load_user_role_result := load_user_role_obj(
user_role_id = user_role_rec.get('user_role_id', None),
@@ -255,13 +252,6 @@ def load_user_obj(
user_obj.user_role_list = user_role_result_list
else: user_obj.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['user_role_list'] = role_rec_li
# user_obj.user_role_list = role_rec_li
# else:
# # user_rec['user_role_list'] = None
# user_obj.user_role_list = None
# log.debug(user_rec)
log.debug(user_obj)
if model_as_dict: