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

Clean up of code...

Browse Source
This commit is contained in:
Scott Idem
2021-06-11 18:42:28 -04:00
parent 64d6e87ca2
commit fe232b8cba
11 changed files with 198 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
app/methods/account_methods.py
View File

@@ -8,7 +8,9 @@ from app.db_sql import redis_lookup_id_random, sql_select
from app.lib_general import log, logging
from app.methods.account_cfg_methods import load_account_cfg_obj
# from app.methods.address_methods import load_address_obj_list
from app.methods.archive_methods import load_archive_obj_list
# from app.methods.contact_methods import load_contact_obj_list
from app.methods.event_methods import load_event_obj_list
from app.methods.post_methods import load_post_obj_list
@@ -110,7 +112,7 @@ def load_account_obj(
if inc_account_cfg:
if account_cfg_dict := load_account_cfg_obj(
account_id = account_id,
model_as_dict = True,
model_as_dict = model_as_dict,
# inc_event_cfg = inc_event_cfg,
inc_fundraising_cfg = inc_fundraising_cfg,
inc_membership_cfg = inc_membership_cfg,
@@ -118,34 +120,37 @@ def load_account_obj(
account_obj.account_cfg = account_cfg_dict
else: account_obj.account_cfg = None
if inc_address_list:
if address_dict_list := load_address_obj_list(
account_id = account_id,
limit = limit,
model_as_dict = model_as_dict,
enabled = enabled,
):
account_obj.address_list = address_dict_list
else: account_obj.address_list = []
if inc_archive_list:
if archive_dict_list := load_archive_obj_list(
account_id = account_id,
limit = limit,
model_as_dict = True,
model_as_dict = model_as_dict,
enabled = enabled,
inc_archive_content_list = inc_archive_content_list,
):
account_obj.archive_list = archive_dict_list
else: account_obj.archive_list = []
# if inc_contact:
# contact_id = account_rec.get('contact_id', None)
# if contact_obj_result := load_contact_obj(contact_id=contact_id):
# contact_obj = contact_obj_result
# #account_obj.contact = contact_obj.dict(by_alias=True, exclude_unset=True)
# account_obj.contact = contact_obj
# else: account_obj.contact = None
# if inc_address:
# address_id = contact_obj.address_id
# if address_obj_result := load_address_obj(address_id=address_id):
# address_obj = address_obj_result
# # account_rec['contact'].address = address_obj
# # log.debug(account_rec)
# #account_obj.contact.address = address_obj.dict(by_alias=True, exclude_unset=True)
# account_obj.contact.address = address_obj
# else: account_obj.contact.address = None
if inc_contact_list:
if contact_dict_list := load_contact_obj_list(
account_id = account_id,
limit = limit,
model_as_dict = model_as_dict,
enabled = enabled,
inc_address = inc_address,
):
account_obj.contact_list = contact_dict_list
else: account_obj.contact_list = []
# if inc_organization:
# organization_id = account_rec.get('organization_id', None)
@@ -161,7 +166,7 @@ def load_account_obj(
if event_dict_list := load_event_obj_list(
account_id = account_id,
limit = limit,
model_as_dict = True,
model_as_dict = model_as_dict,
enabled = enabled,
# inc_location_address = inc_address,
# inc_contact_1 = inc_contact,
@@ -189,7 +194,7 @@ def load_account_obj(
if post_dict_list := load_post_obj_list(
account_id = account_id,
limit = limit,
model_as_dict = True,
model_as_dict = model_as_dict,
enabled = enabled,
inc_post_comment_list = inc_post_comment_list,
inc_person = inc_person,

17
app/methods/address_methods.py
View File

@@ -34,18 +34,20 @@ def create_address_obj(address_obj_new:Address_Base):
# ### BEGIN ### API Address Methods ### load_address_obj() ###
def load_address_obj(address_id:int|str) -> Address_Base|bool:
def load_address_obj(
address_id:int|str,
limit: int = 1000, # Probably not needed for the address
model_as_dict: bool = False,
enabled: str = 'enabled', # enabled, disabled, all # Probably not needed for the address
) -> Address_Base|bool:
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 False
if address_rec := sql_select(table_name='v_address', record_id=address_id):
#log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(address_rec)
else:
return False
if address_rec := sql_select(table_name='v_address', record_id=address_id): pass
else: return False
try:
address_obj = Address_Base(**address_rec)
@@ -54,6 +56,9 @@ def load_address_obj(address_id:int|str) -> Address_Base|bool:
log.error(e.json())
return False
if model_as_dict:
return address_obj.dict(by_alias=True, exclude_unset=True) # pylint: disable=no-member
else:
return address_obj
# ### END ### API Address Methods ### load_address_obj() ###

38
app/methods/contact_methods.py
View File

@@ -55,26 +55,21 @@ def create_contact_obj(contact_obj_new:Contact_Base):
# ### BEGIN ### API Contact Methods ### load_contact_obj() ###
def load_contact_obj(contact_id:int|str, inc_address:bool=False) -> Contact_Base|bool:
def load_contact_obj(
contact_id:int|str,
limit: int = 1000,
model_as_dict: bool = False,
enabled: str = 'enabled', # enabled, disabled, all
inc_address:bool=False
) -> Contact_Base|bool:
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 False
if contact_rec := sql_select(table_name='v_contact', record_id=contact_id):
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(contact_rec)
if inc_address:
address_id = contact_rec.get('address_id', None)
if address_obj_result := load_address_obj(address_id=address_id):
address_obj = address_obj_result
contact_rec['address'] = address_obj
log.debug(contact_rec)
#else: contact_rec['address'] = None
else:
return False
if contact_rec := sql_select(table_name='v_contact', record_id=contact_id): pass
else: return False
try:
contact_obj = Contact_Base(**contact_rec)
@@ -83,6 +78,21 @@ def load_contact_obj(contact_id:int|str, inc_address:bool=False) -> Contact_Base
log.error(e.json())
return False
if inc_address:
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
address_id = contact_rec.get('address_id', None)
log.debug(address_id)
from app.methods.address_methods import load_address_obj
if address_dict := load_address_obj(
address_id = address_id,
model_as_dict = model_as_dict,
):
contact_obj.address = address_dict
else: contact_obj.address = None
if model_as_dict:
return contact_obj.dict(by_alias=True, exclude_unset=True) # pylint: disable=no-member
else:
return contact_obj
# ### END ### API Contact Methods ### load_contact_obj() ###

17
app/methods/event_methods.py
View File

@@ -204,7 +204,8 @@ def load_event_cfg_obj(
# ### BEGIN ### API Event Methods ### load_event_obj_list() ###
def load_event_obj_list(
account_id: int|str,
account_id: int|str|None = None,
user_id: int|str|None = None,
limit: int = 1000,
model_as_dict: bool = False,
enabled: str = 'enabled', # enabled, disabled, all
@@ -236,11 +237,17 @@ def load_event_obj_list(
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
data = {}
if account_id:
if account_id := redis_lookup_id_random(record_id_random=account_id, table_name='account'): pass
else: return False
data = {}
data['account_id'] = account_id
sql_obj_type_id = f'`tbl`.account_id = :account_id'
elif user_id:
if user_id := redis_lookup_id_random(record_id_random=user_id, table_name='user'): pass
else: return False
data['user_id'] = user_id
sql_obj_type_id = f'`tbl`.user_id = :user_id'
if enabled in ['enabled', 'disabled', 'all']:
if enabled == 'enabled':
@@ -262,7 +269,8 @@ def load_event_obj_list(
sql = f"""
SELECT `tbl`.id AS 'event_id', `tbl`.id_random AS 'event_id_random'
FROM `event` AS `tbl`
WHERE `tbl`.account_id = :account_id
WHERE
{sql_obj_type_id}
{sql_enabled}
ORDER BY `tbl`.created_on DESC, `tbl`.updated_on DESC
{sql_limit};
@@ -308,7 +316,6 @@ def load_event_obj_list(
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(event_rec_li_result)
event_result_li = []
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
return event_result_li
# ### END ### API Event Methods ### load_event_obj_list() ###

48
app/methods/organization_methods.py
View File

@@ -15,42 +15,48 @@ 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:
def load_organization_obj(
organization_id:int|str,
limit: int = 1000,
model_as_dict: bool = False,
enabled: str = 'enabled', # enabled, disabled, all
inc_contact:bool=False,
inc_address:bool=False
) -> Organization_Base|bool:
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if organization_id := redis_lookup_id_random(record_id_random=organization_id, table_name='organization'): pass
else: return False
if organization_rec := sql_select(table_name='v_organization', record_id=organization_id):
#log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(organization_rec)
if organization_rec := sql_select(table_name='v_organization', record_id=organization_id): pass
else: return False
if inc_contact:
contact_id = organization_rec.get('contact_id', None)
if contact_obj_result := load_contact_obj(contact_id=contact_id):
contact_obj = contact_obj_result
organization_rec['contact'] = contact_obj
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(organization_rec)
#else: organization_rec['contact'] = None
if inc_address:
address_id = contact_obj.address_id
if address_obj_result := load_address_obj(address_id=address_id):
address_obj = address_obj_result
organization_rec['contact'].address = address_obj
log.debug(organization_rec)
#else: organization_rec['contact'].address = None
else:
return False
try:
organization_obj = Organization_Base(**organization_rec)
log.debug(organization_obj)
except ValidationError as e:
log.error(e.json())
return False
if inc_contact:
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
contact_id = organization_rec.get('contact_id', None)
log.debug(contact_id)
from app.methods.contact_methods import load_contact_obj
if contact_dict := load_contact_obj(
contact_id = contact_id,
model_as_dict = model_as_dict,
inc_address = inc_address,
):
organization_obj.contact = contact_dict
else: organization_obj.contact = None
if model_as_dict:
return organization_obj.dict(by_alias=True, exclude_unset=True) # pylint: disable=no-member
else:
return organization_obj
# ### END ### API Organization Methods ### load_organization_obj() ###

50
app/methods/person_methods.py
View File

@@ -62,10 +62,13 @@ def create_person_obj(person_obj_new:Person_Base):
# ### BEGIN ### API Person Methods ### load_person_obj() ###
def load_person_obj(
person_id: int|str,
limit: int = 1000,
model_as_dict: bool = False,
enabled: str = 'enabled', # enabled, disabled, all
inc_address: bool = False, # Under contact
inc_contact: bool = False,
inc_organization: bool = False,
inc_user: bool = False
inc_user: bool = False,
) -> Person_Base|bool:
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
@@ -76,9 +79,6 @@ def load_person_obj(
if person_rec := sql_select(table_name='v_person', record_id=person_id): pass
else: return False
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(person_rec)
try:
person_obj = Person_Base(**person_rec)
log.debug(person_obj)
@@ -87,30 +87,29 @@ def load_person_obj(
if inc_contact:
contact_id = person_rec.get('contact_id', None)
if contact_obj_result := load_contact_obj(contact_id=contact_id):
contact_obj = contact_obj_result
#person_obj.contact = contact_obj.dict(by_alias=True, exclude_unset=True)
person_obj.contact = contact_obj
log.debug(contact_id)
if contact_dict := load_contact_obj(
contact_id = contact_id,
limit = limit,
model_as_dict = model_as_dict,
enabled = enabled,
inc_address = inc_address,
):
person_obj.contact = contact_dict
else: person_obj.contact = None
if inc_address:
address_id = contact_obj.address_id
if address_obj_result := load_address_obj(address_id=address_id):
address_obj = address_obj_result
# person_rec['contact'].address = address_obj
# log.debug(person_rec)
#person_obj.contact.address = address_obj.dict(by_alias=True, exclude_unset=True)
person_obj.contact.address = address_obj
else: person_obj.contact.address = None
if inc_organization:
organization_id = person_rec.get('organization_id', None)
if organization_obj_result := load_organization_obj(organization_id=organization_id):
organization_obj = organization_obj_result
# person_rec['organization'] = organization_obj
# log.debug(person_rec)
#person_obj.organization = organization_obj.dict(by_alias=True, exclude_unset=True)
person_obj.organization = organization_obj
log.debug(organization_id)
if organization_dict := load_organization_obj(
organization_id = organization_id,
limit = limit,
model_as_dict = model_as_dict,
enabled = enabled,
inc_address = inc_address,
inc_contact = inc_contact,
):
person_obj.organization = organization_dict
else: person_obj.organization = None
if inc_user:
@@ -123,6 +122,9 @@ def load_person_obj(
person_obj.user = user_obj
else: person_obj.user = None
if model_as_dict:
return person_obj.dict(by_alias=True, exclude_unset=True) # pylint: disable=no-member
else:
return person_obj
# ### END ### API Person Methods ### load_person_obj() ###

17
app/methods/post_methods.py
View File

@@ -109,7 +109,8 @@ def load_post_obj(
# ### BEGIN ### API Post Methods ### load_post_obj_list() ###
def load_post_obj_list(
account_id: int|str,
account_id: int|str|None = None,
user_id: int|str|None = None,
limit: int = 1000,
model_as_dict: bool = False,
enabled: str = 'enabled', # enabled, disabled, all
@@ -120,11 +121,17 @@ def load_post_obj_list(
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
data = {}
if account_id:
if account_id := redis_lookup_id_random(record_id_random=account_id, table_name='account'): pass
else: return False
data = {}
data['account_id'] = account_id
sql_obj_type_id = f'`tbl`.account_id = :account_id'
elif user_id:
if user_id := redis_lookup_id_random(record_id_random=user_id, table_name='user'): pass
else: return False
data['user_id'] = user_id
sql_obj_type_id = f'`tbl`.user_id = :user_id'
if enabled in ['enabled', 'disabled', 'all']:
if enabled == 'enabled':
@@ -146,7 +153,8 @@ def load_post_obj_list(
sql = f"""
SELECT `tbl`.id AS 'post_id', `tbl`.id_random AS 'post_id_random'
FROM `post` AS `tbl`
WHERE `tbl`.account_id = :account_id
WHERE
{sql_obj_type_id}
{sql_enabled}
ORDER BY `tbl`.created_on DESC, `tbl`.updated_on DESC
{sql_limit};
@@ -176,7 +184,6 @@ def load_post_obj_list(
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(post_rec_li_result)
post_result_li = []
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
return post_result_li
# ### END ### API Post Methods ### load_post_obj_list() ###

70
app/methods/user_methods.py
View File

@@ -112,16 +112,19 @@ def load_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.contact = contact_obj
else:
user_rec['contact'] = None
user_obj.contact = None
log.debug(contact_id)
if contact_dict := load_contact_obj(
contact_id = contact_id,
limit = limit,
model_as_dict = model_as_dict,
enabled = enabled,
inc_address = inc_address,
):
user_obj.contact = contact_dict
else: user_obj.contact = None
if inc_event_list:
from app.methods.event_methods import load_event_obj_list
if event_dict_list := load_event_obj_list(
user_id = user_id,
limit = limit,
@@ -129,32 +132,41 @@ def load_user_obj(
enabled = enabled,
):
user_obj.event_list = event_dict_list
else: user_obj.event_list = None
else: user_obj.event_list = []
if inc_organization:
organization_id = user_rec.get('organization_id', None)
if organization_obj_result := load_organization_obj(organization_id=organization_id):
organization_obj = organization_obj_result
user_rec['organization'] = organization_obj
log.debug(user_rec)
user_obj.organization = organization_obj
else:
user_rec['organization'] = None
user_obj.organization = None
log.debug(organization_id)
if organization_dict := load_organization_obj(
organization_id = organization_id,
limit = limit,
model_as_dict = model_as_dict,
enabled = enabled,
inc_address = inc_address,
inc_contact = inc_contact,
):
user_obj.organization = organization_dict
else: user_obj.organization = None
# if inc_person:
# person_id = user_rec.get('person_id', None)
# if person_obj_result := load_person_obj(person_id=person_id):
# person_obj = person_obj_result
# user_rec['person'] = person_obj
# log.debug(user_rec)
# user_obj.person = person_obj
# else:
# user_rec['person'] = None
# user_obj.person = None
if inc_person:
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
person_id = user_rec.get('person_id', None)
log.debug(person_id)
from app.methods.person_methods import load_person_obj
if person_dict := load_person_obj(
person_id = person_id,
limit = limit,
model_as_dict = model_as_dict,
enabled = enabled,
inc_address = inc_address,
inc_contact = inc_contact,
inc_organization = inc_organization,
):
user_obj.person = person_dict
else: user_obj.person = None
if inc_post_list:
from app.methods.post_methods import load_post_obj_list
if post_dict_list := load_post_obj_list(
user_id = user_id,
limit = limit,
@@ -162,7 +174,7 @@ def load_user_obj(
enabled = enabled,
):
user_obj.post_list = post_dict_list
else: user_obj.post_list = None
else: user_obj.post_list = []
# NOTE: Including user roles should probably be reviewed

5
app/models/person_models.py
View File

@@ -18,11 +18,6 @@ class Person_Base(BaseModel):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
#from .account_models import Account_Base
#from .contact_models import Contact_Base
#from .organization_models import Organization_Base
#from .user_models import User_Base
#from .organization_models import Organization_Base
#if TYPE_CHECKING:
#from .supporting_core_models import Address_Base, Contact_Base, Organization_Base, Person_Base, User_Base

10
app/models/user_models.py
View File

@@ -181,6 +181,8 @@ class User_Out_Base(BaseModel):
updated_on: Optional[datetime.datetime]
# Including other related objects
# from app.models.person_models import Person_Base # Causes circular import
archive_list: Optional[list] # Archive_Base()
contact: Optional[Contact_Base]
event_list: Optional[list] # Event_Base() # Priority l1
@@ -190,7 +192,7 @@ class User_Out_Base(BaseModel):
order_list: Optional[list] # Order_Base() # Priority l2
order_cart_list: Optional[list] # Order_Base() # Priority l2
organization: Optional[Organization_Base] # Organization_Base() # Priority l3
# person: Optional[Person_Base] # Person_Base() # Priority l2
person: Optional[dict] # Person_Base() # Priority l2
post_list: Optional[list] # Post_Base() # Priority l1
user_role_list: Optional[list] # User_Role_Base()
role_list: Optional[list] = [] # User_Role_Base() # NOTE <- This is a duplicate of above!
@@ -270,7 +272,7 @@ class User_Base(BaseModel):
order_list: Optional[list] # Order_Base() # Priority l2
order_cart_list: Optional[list] # Order_Base() # Priority l2
organization: Optional[Organization_Base] # Organization_Base() # Priority l3
# person: Optional[Person_Base] # Person_Base() # Priority l2
person: Optional[dict] # Person_Base() # Priority l2
post_list: Optional[list] # Post_Base() # Priority l1
user_role_list: Optional[list] # User_Role_Base()
role_list: Optional[list] = [] # User_Role_Base() # NOTE <- This is a duplicate of above!
@@ -337,5 +339,5 @@ class User_Base(BaseModel):
fields = base_fields
# ### END ### API User Models ### User_Base() ###
#User_Base.update_forward_refs()
#User_Out_Base.update_forward_refs()
# User_Base.update_forward_refs()
# User_Out_Base.update_forward_refs()

4
app/routers/user.py
View File

@@ -561,9 +561,9 @@ async def lookup_username(
# ### BEGIN ### API User ### get_user_obj_new() ###
# Working well as of 2021-06-11. Using as a template for other routes.
@router.get('/{user_id}', response_model=Resp_Body_Base)
@router.get('/{user_id}/json', response_model=Resp_Body_Base)
async def get_user_obj_new(
account_id: str = Query(..., min_length=1, max_length=22),
user_id: str = Query(..., min_length=1, max_length=22),
limit: int = 500, # For now this covers any included objects or object lists
enabled: str = 'enabled', # For now this covers any included objects or object lists
inc_address: bool = False, # Priority l1