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

Code clean up. Bug fixes for person, user, contact, and address methods. Fix bug for get_account_id_w_for_type_id()

Browse Source
This commit is contained in:
Scott Idem
2022-01-07 13:35:44 -05:00
parent 29c6770581
commit 5489b80ff0
5 changed files with 117 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files

105
app/methods/address_methods.py
View File

@@ -5,7 +5,7 @@ from typing import Dict, List, Optional, Set, Union
from pydantic import BaseModel, EmailStr, Field, PrivateAttr, ValidationError, validator
from app.db_sql import get_account_id_w_for_type_id, redis_lookup_id_random, sql_insert, sql_select, sql_update
from app.lib_general import log, logging
from app.lib_general import log, logging, logger_reset
from app.models.address_models import Address_Base
from app.models.common_field_schema import default_num_bytes
@@ -44,32 +44,40 @@ def load_address_obj(
# ### BEGIN ### API Address Methods ### get_address_rec_list() ###
# Updated 2022-01-07
@logger_reset
def get_address_rec_list(
for_obj_type: str,
for_obj_id: str,
limit: int = 1000,
for_type: str, # 'account' is a special case
for_id: str,
limit: int = 500,
enabled: str = 'enabled', # enabled, disabled, all
) -> list|bool:
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.setLevel(logging.INFO) # 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
if for_id := redis_lookup_id_random(record_id_random=for_id, table_name=for_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 for_type == 'account':
sql_for_type_id = f'`address`.account_id = :for_id'
else:
sql_for_type_id = f'`address`.for_type = :for_type AND `address`.for_id = :for_id'
data = {}
# data[f'{for_type}_id'] = for_id
data['for_type'] = for_type
data['for_id'] = for_id
# sql_obj_type_id = f'`address`.{for_type}_id = :{for_type}_id'
if enabled in ['enabled', 'disabled', 'all']:
if enabled == 'enabled':
data['enable'] = True
sql_enabled = f'AND `address`.enable = :enable'
elif enabled == 'disabled':
data['enable'] = False
sql_enabled = f'AND `address`.enable = :enable'
elif enabled == 'all':
sql_enabled = ''
if limit:
data['limit'] = limit
@@ -78,20 +86,20 @@ def get_address_rec_list(
sql_limit = ''
sql = f"""
SELECT `tbl`.id AS 'address_id', `tbl`.id_random AS 'address_id_random'
FROM `address` AS `tbl`
SELECT `address`.id AS 'address_id', `address`.id_random AS 'address_id_random'
FROM `address` AS `address`
WHERE
{sql_obj_type_id}
{sql_for_type_id}
{sql_enabled}
ORDER BY `tbl`.created_on DESC, `tbl`.updated_on DESC
ORDER BY `address`.created_on DESC, `address`.updated_on DESC
{sql_limit};
"""
if address_rec_li_result := sql_select(data=data, sql=sql, as_list=True):
address_rec_li = address_rec_li_result
else:
address_rec_li = []
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
else: # [] or False
address_rec_li = address_rec_li_result
log.debug(address_rec_li_result)
return address_rec_li
@@ -231,7 +239,7 @@ def create_address_obj(
for_id: int|str = None,
fail_any: bool = False, # Fail if any thing goes wrong for sub objects
) -> int|bool:
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
# ### SECTION ### Secondary data validation
@@ -244,6 +252,7 @@ def create_address_obj(
log.info('Create dictionary or Pydantic object')
log.debug(type(address_dict_obj))
log.debug(address_dict_obj)
if isinstance(address_dict_obj, dict):
address_dict = address_dict_obj
try:
@@ -257,31 +266,7 @@ def create_address_obj(
address_obj.account_id = account_id
address_dict = address_obj.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'created_on', 'updated_on'})
# log.debug(type(address_dict_obj))
# if isinstance(address_dict_obj, dict):
# if account_id:
# address_obj_new['account_id'] = account_id
# if for_type:
# address_obj_new['for_type'] = for_type
# if for_id:
# address_obj_new['for_id'] = for_id
# try:
# address_obj_new = Address_Base(**address_obj_new)
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
# log.debug(address_obj_new)
# except ValidationError as e:
# log.error(e.json())
# return False
# else:
# if account_id:
# address_obj_new.account_id = account_id
# if for_type:
# address_obj_new.for_type = for_type
# if for_id:
# address_obj_new.for_id = for_id
# address_dict = address_obj_new.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'created_on', 'updated_on'})
log.debug(address_dict)
# ### SECTION ### Process data
# Look for an account_id in the address_obj
@@ -296,6 +281,16 @@ def create_address_obj(
address_dict['for_type'] = for_type
address_dict['for_id'] = for_id
if address_obj.id:
log.warning(f'There should not be an Address ID: {address_obj.id}')
return False
else:
log.info(f'Should there be a check for an existing address with? For Type: {for_type}; For ID: {for_id}')
if get_address_rec_result := get_address_rec_list(for_type=for_type, for_id=for_id):
log.warning(f'One or more addresses were found with: For Type: {for_type}; For ID: {for_id}')
return False
else: log.info(f'No existing address found with: For Type: {for_type}; For ID: {for_id}')
if address_dict_in_result := sql_insert(data=address_dict, table_name='address', rm_id_random=True, id_random_length=8): pass
else:
return False
@@ -303,7 +298,7 @@ def create_address_obj(
log.debug(address_dict_in_result)
address_id = address_dict_in_result
log.inf(f'Returning the new address_id: {address_id}')
log.info(f'Returning the new address_id: {address_id}')
return address_id
# ### END ### API Address Methods ### create_address_obj() ###
@@ -319,7 +314,7 @@ def update_address_obj(
create_sub_obj: bool = False,
fail_any: bool = False, # Fail if any thing goes wrong for sub objects
) -> bool:
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
# ### SECTION ### Secondary data validation