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

Working on contact and address create update v4

Browse Source
This commit is contained in:
Scott Idem
2021-08-25 18:48:55 -04:00
parent a508307df3
commit 107366d4f8
4 changed files with 147 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

126
app/methods/address_methods.py
View File

@@ -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, 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.models.address_models import Address_Base
@@ -98,6 +98,130 @@ def get_address_rec_list(
# ### END ### API Address Methods ### get_address_rec_list() ###
# ### BEGIN ### API Address Methods ### create_update_address_obj_v4() ###
# NOTE: This will create or update an address.
# Rewrite and updated 2021-08-25
def create_update_address_obj_v4(
address_dict_obj: Address_Base|dict,
address_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 address_id:
log.info(f'Address ID passed. Update existing Address. Address ID: {address_id}')
if address_id := redis_lookup_id_random(record_id_random=address_id, table_name='address'): pass
else:
log.error('Address 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='address', for_id=address_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 Address ID passed. Create new Address. 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(address_dict_obj))
if isinstance(address_dict_obj, dict):
address_dict = address_dict_obj
if address_id:
address_dict['address_id'] = address_id
if account_id:
address_dict['account_id'] = account_id
if for_type:
address_dict['for_type'] = for_type
if for_id:
address_dict['for_id'] = for_id
try:
address_obj = Address_Base(**address_dict)
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(address_obj)
except ValidationError as e:
log.error(e.json())
return False
else:
address_obj = address_dict_obj
if address_id:
# NOTE: Can't update the ID alias if it was never set.
address_obj.id = address_id
if account_id:
address_obj.account_id = account_id
if for_type:
address_obj.for_type = for_type
if for_id:
address_obj.for_id = for_id
address_dict = address_obj.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'created_on', 'updated_on'})
if address_id:
if address_dict_up_result := sql_update(data=address_dict, table_name='address', rm_id_random=True): pass
else:
log.warning(f'Address not updated. Address ID: {address_id}')
log.debug(address_dict_up_result)
return False
log.debug(address_dict_up_result)
else:
if address_dict_in_result := sql_insert(data=address_dict, table_name='address', rm_id_random=True, id_random_length=8): pass
else:
log.warning(f'Address not created.')
log.debug(address_dict_in_result)
return False
log.debug(address_dict_in_result)
address_id = address_dict_in_result
address_outline = {}
address_outline['address_id'] = address_id
# Should this outline include for_type and for_id? Probably later.
if return_outline:
log.debug(f'Returning the Address Outline: {address_outline}')
return address_outline
else:
log.debug(f'Returning the Address ID: {address_id}')
return address_id
# ### END ### API Address Methods ### create_update_address_obj_v4() ###
# ### BEGIN ### API Address Methods ### create_address_obj() ###
# NOTE: This will create an address.
# Reviewed and updated 2021-08-10

16
app/methods/contact_methods.py
View File

@@ -21,7 +21,7 @@ def load_contact_obj(
exclude_unset: bool = True,
model_as_dict: bool = False,
enabled: str = 'enabled', # enabled, disabled, all
inc_address:bool=False
inc_address: bool = False
) -> Contact_Base|dict|bool:
# log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
@@ -41,7 +41,12 @@ def load_contact_obj(
if inc_address:
log.info('Need to include address data...')
address_id = contact_rec.get('address_id', None)
# NOTE: The address_id field needs to be removed from the contact table.
# NOTE: This is not ideal and the view needs to be updated when possible and anything currently pointing to the original contact.address_id needs to be fixed.
if address_id := contact_rec.get('address_id', None): pass # WARNING
elif address_id := contact_rec.get('linked_address_id', None): pass # WARNING
else: pass # WARNING
# address_id = contact_rec.get('address_id', None)
log.debug(address_id)
from app.methods.address_methods import load_address_obj
if address_result := load_address_obj(
@@ -147,12 +152,8 @@ def 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.
# NOTE: This will create or update a contact.
# Rewrite and updated 2021-08-25
def create_update_contact_obj_v4(
contact_dict_obj: Contact_Base|dict,
@@ -270,6 +271,7 @@ def create_update_contact_obj_v4(
contact_outline['address_id'] = None
address_obj = contact_obj.address
if address_id := contact_obj.address_id: pass
elif address_id := address_obj.id: pass
else: address_id = None
address_obj.id
address_obj.for_type = 'contact'

12
app/models/contact_models.py
View File

@@ -30,6 +30,9 @@ class Contact_Base(BaseModel):
address_id_random: Optional[str]
address_id: Optional[int]
linked_address_id_random: Optional[str]
linked_address_id: Optional[int]
for_type: Optional[str]
for_id_random: Optional[str]
for_id: Optional[int]
@@ -119,6 +122,15 @@ class Contact_Base(BaseModel):
return redis_lookup_id_random(record_id_random=values['address_id_random'], table_name='address')
return None
@validator('linked_address_id', always=True)
def linked_address_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values.get('linked_address_id_random', None):
return redis_lookup_id_random(record_id_random=values['linked_address_id_random'], table_name='linked_address')
return None
@validator('for_id', always=True)
def for_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)

2
app/routers/contact.py
View File

@@ -147,7 +147,7 @@ async def patch_contact_obj_exist_v4(
if update_contact_obj_result := create_update_contact_obj_v4(
contact_dict_obj = contact_obj,
contact_id = contact_obj.contact_id,
contact_id = contact_id,
account_id = x_account_id,
for_type = for_type,
for_id = for_id,