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

Working on contact create update v4

Browse Source
This commit is contained in:
Scott Idem
2021-08-25 17:57:51 -04:00
parent c93792634a
commit a508307df3
3 changed files with 302 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
app/db_sql.py
View File

@@ -1024,4 +1024,38 @@ def lookup_id_random_pop(obj_data:dict):
obj_data.pop('from_object_id_random')
return obj_data
# ### END ### API DB SQL ### lookup_id_random_pop() ###
# ### END ### API DB SQL ### lookup_id_random_pop() ###
# ### BEGIN ### API DB SQL Methods ### get_account_id_w_for_type_id() ###
# Updated 2021-08-24
def get_account_id_w_for_type_id(
for_type: str,
for_id: int|str,
) -> bool|int|None:
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if for_id := redis_lookup_id_random(record_id_random=for_id, table_name=for_type): pass
else: return False
data = {}
data['for_type'] = for_type
data['for_id'] = for_id
sql = f"""
SELECT `for`.id AS 'for_id', `for`.id_random AS 'for_id_random', `for`.account_id AS account_id
FROM :for_type AS `for`
WHERE `for`.id = :for_id
LIMIT 1;
"""
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
if for_data_result := sql_select(data=data, sql=sql):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(for_data_result)
if account_id := for_data_result.get('account_id', None): return account_id
else: return False
else: return None
# ### END ### API DB SQL Methods ### get_account_id_w_for_type_id() ###

163
app/methods/contact_methods.py
View File

@@ -4,10 +4,10 @@ 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 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.methods.address_methods import create_address_obj, create_update_address_obj, update_address_obj
from app.methods.address_methods import create_address_obj, create_update_address_obj, create_update_address_obj_v4, update_address_obj
from app.models.common_field_schema import default_num_bytes
from app.models.contact_models import Contact_Base
@@ -146,6 +146,165 @@ def get_account_id_w_contact_id(
# ### END ### API Contact Methods ### get_account_id_w_contact_id() ###
# ### BEGIN ### API Contact Methods ### create_update_contact_obj_v4() ###
# NOTE: This will create a contact and then also create a linked address if contact_obj.address data is passed.
# NOTE: In the future it should be required that account_id, for_type, and for_id should be passed separately. account_id might not be required *if* it can be looked up based on for_type and for_id.
# Rewrite and updated 2021-08-25
def create_update_contact_obj_v4(
contact_dict_obj: Contact_Base|dict,
contact_id: int|str = None,
account_id: int|str = None,
for_type: str = None,
for_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 contact_id:
log.info(f'Contact ID passed. Update existing Contact. Contact ID: {contact_id}')
if contact_id := redis_lookup_id_random(record_id_random=contact_id, table_name='contact'): pass
else:
log.error('Contact ID passed but is invalid. Failed requirement.')
return False
if account_id := redis_lookup_id_random(record_id_random=account_id, table_name='account'): pass
else:
log.error('Missing or invalid Account ID passed. Not required. Ignoring.')
log.info(f'Account ID: {account_id}')
log.info('Attempting to get Account ID from related object.')
if account_id := get_account_id_w_for_type_id(for_type='contact', for_id=contact_id): pass
else:
log.error('Unable to get Account ID from related object.')
False
if for_id := redis_lookup_id_random(record_id_random=for_id, table_name=for_type): pass
else:
log.error('Missing or invalid For Type and For ID ID passed. Not required. Ignoring.')
log.info(f'For Type: {for_type} and For ID: {for_id}')
else:
log.info('No Contact ID passed. Create new Contact. Required: Account ID, For Type, For ID')
if for_id := redis_lookup_id_random(record_id_random=for_id, table_name=for_type): pass
else:
log.error('Missing or invalid For Type and For ID ID passed. Failed requirement.')
log.info(f'For Type: {for_type} and For ID: {for_id}')
return False
if account_id := redis_lookup_id_random(record_id_random=account_id, table_name='account'): pass
else:
log.error('Missing or invalid Account ID passed. Failed requirement.')
log.info(f'Account ID: {account_id}')
if for_type and for_id:
log.info('Attempting to get Account ID from related object.')
if account_id := get_account_id_w_for_type_id(for_type=for_type, for_id=for_id): pass
else:
log.error('Unable to get Account ID from related object.')
False
else:
return False
log.debug(type(contact_dict_obj))
if isinstance(contact_dict_obj, dict):
contact_dict = contact_dict_obj
if contact_id:
contact_dict['contact_id'] = contact_id
if account_id:
contact_dict['account_id'] = account_id
if for_type:
contact_dict['for_type'] = for_type
if for_id:
contact_dict['for_id'] = for_id
try:
contact_obj = Contact_Base(**contact_dict)
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(contact_obj)
except ValidationError as e:
log.error(e.json())
return False
else:
contact_obj = contact_dict_obj
if contact_id:
# NOTE: Can't update the ID alias if it was never set.
contact_obj.id = contact_id
if account_id:
contact_obj.account_id = account_id
if for_type:
contact_obj.for_type = for_type
if for_id:
contact_obj.for_id = for_id
contact_dict = contact_obj.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'address', 'created_on', 'updated_on'})
if contact_id:
if contact_dict_up_result := sql_update(data=contact_dict, table_name='contact', rm_id_random=True): pass
else:
log.warning(f'Contact not updated. Contact ID: {contact_id}')
log.debug(contact_dict_up_result)
return False
log.debug(contact_dict_up_result)
else:
if contact_dict_in_result := sql_insert(data=contact_dict, table_name='contact', rm_id_random=True, id_random_length=8): pass
else:
log.warning(f'Contact not created.')
log.debug(contact_dict_in_result)
return False
log.debug(contact_dict_in_result)
contact_id = contact_dict_in_result
contact_outline = {}
contact_outline['contact_id'] = contact_id
# NOTE: Use object model version because of better type checking and validations
if contact_obj.address:
contact_outline['address_id'] = None
address_obj = contact_obj.address
if address_id := contact_obj.address_id: pass
else: address_id = None
address_obj.id
address_obj.for_type = 'contact'
address_obj.for_id = contact_id
create_update_address_obj_result = create_update_address_obj_v4(
address_dict_obj = address_obj,
address_id = address_id,
account_id = account_id,
for_type = 'contact',
for_id = contact_id,
fail_any = fail_any,
return_outline = return_outline,
)
if isinstance(create_update_address_obj_result, int):
address_id = create_update_address_obj_result
elif create_update_address_obj_result == True: pass
else:
log.warning(f'Create or Update failed while trying create_update_address_obj_v4(): {create_update_address_obj_result}')
address_id = None
contact_outline['address_id'] = address_id
if return_outline:
log.debug(f'Returning the Contact Outline: {contact_outline}')
return contact_outline
else:
log.debug(f'Returning the Contact ID: {contact_id}')
return contact_id
# ### END ### API Contact Methods ### create_update_contact_obj_v4() ###
# ### BEGIN ### API Contact Methods ### create_contact_obj() ###
# NOTE: This will create a contact and then also create a linked address if contact_obj.address data is passed.
# NOTE: In the future it should be required that account_id, for_type, and for_id should be passed separately. account_id might not be required *if* it can be looked up based on for_type and for_id.

108
app/routers/contact.py
View File

@@ -5,11 +5,11 @@ from typing import Dict, List, Optional, Set, Union
from app.lib_general import log, logging
from app.config import settings
from app.db_sql import sql_insert, sql_update, sql_insert_or_update, sql_select, sql_delete, redis_lookup_id_random
from app.db_sql import get_id_random, sql_insert, sql_update, sql_insert_or_update, sql_select, sql_delete, redis_lookup_id_random
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.contact_methods import load_contact_obj, update_contact_obj
from app.methods.contact_methods import create_update_contact_obj_v4, load_contact_obj, update_contact_obj
from app.models.contact_models import Contact_Base
from app.models.response_models import *
@@ -70,6 +70,110 @@ async def patch_contact_obj(
return result
# ### BEGIN ### API Contact ### post_contact_obj_new_v4() ###
# Updated 2021-08-25
@router.post('/new_v4', response_model=Resp_Body_Base)
async def post_contact_obj_new_v4(
contact_obj: Contact_Base,
for_type: str = None,
for_id: str = None,
create_sub_obj: bool = False,
fail_any: bool = True, # Fail if any thing goes wrong for sub objects
x_account_id: str = Header(...),
return_obj: bool = True,
inc_address: bool = False,
by_alias: bool = True,
exclude_unset: bool = True,
response: Response = Response,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
# There should probably be a check for the account ID before calling the create update function?
if create_update_contact_obj_result := create_update_contact_obj_v4(
contact_dict_obj = contact_obj,
contact_id = None,
account_id = x_account_id,
for_type = for_type,
for_id = for_id,
create_sub_obj = create_sub_obj,
fail_any = fail_any,
return_outline = False,
): pass
else: return mk_resp(data=False, status_code=400, response=response, status_message='The contact was not created. Check the field names and data types.')
if isinstance(create_update_contact_obj_result, int):
contact_id = create_update_contact_obj_result
if return_obj:
if load_contact_obj_result := load_contact_obj(contact_id=contact_id, inc_address=inc_address):
data = load_contact_obj_result
else:
data = False
else:
contact_id = create_update_contact_obj_result
contact_id_random = get_id_random(record_id=contact_id, table_name='contact')
data = {}
data['contact_id'] = contact_id
data['contact_id_random'] = contact_id_random
return mk_resp(data=data, response=response, status_message='The contact was created.')
else:
return mk_resp(data=False, status_code=400, response=response, status_message='The result from trying to create an contact was unexpected.')
# ### BEGIN ### API Contact ### post_contact_obj_new_v4() ###
# ### BEGIN ### API Contact ### patch_contact_obj_exist_v4() ###
# Updated 2021-08-25
@router.patch('/{contact_id}/exist_v4', response_model=Resp_Body_Base)
async def patch_contact_obj_exist_v4(
contact_obj: Contact_Base,
contact_id: str = Query(..., min_length=11, max_length=22),
for_type: str = None,
for_id: str = None,
create_sub_obj: bool = False,
fail_any: bool = True, # Fail if any thing goes wrong for sub objects
x_account_id: Optional[str] = Header(..., ),
return_obj: Optional[bool] = True,
inc_address: bool = False,
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
exclude_none: Optional[bool] = True,
response: Response = Response,
):
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 update_contact_obj_result := create_update_contact_obj_v4(
contact_dict_obj = contact_obj,
contact_id = contact_obj.contact_id,
account_id = x_account_id,
for_type = for_type,
for_id = for_id,
create_sub_obj = create_sub_obj,
fail_any = fail_any,
return_outline = False,
): pass
else: return mk_resp(data=False, status_code=400, response=response, status_message='The contact was not created. Check the field names and data types.')
if update_contact_obj_result:
if return_obj:
if load_contact_obj_result := load_contact_obj(contact_id=contact_id, inc_address=inc_address):
data = load_contact_obj_result
else:
data = False
else:
contact_id_random = get_id_random(record_id=contact_id, table_name='contact')
data = {}
data['contact_id'] = contact_id
data['contact_id_random'] = contact_id_random
return mk_resp(data=data, response=response, status_message='The contact was created.')
else:
return mk_resp(data=False, status_code=400, response=response, status_message='The result from trying to create an contact was unexpected.')
# ### END ### API Contact ### patch_contact_obj_exist_v4() ###
# ### BEGIN ### API Contact ### patch_contact_json() ###
@router.patch('/{contact_id}/json', response_model=Resp_Body_Base)
async def patch_contact_json(