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

Working on event_person and session proposals. Also general clean up of models and methods.

Browse Source
This commit is contained in:
Scott Idem
2021-05-27 16:29:27 -04:00
parent 8f6589cf1c
commit 65fd3ebe28
20 changed files with 198 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
app/methods/account_methods.py Normal file
View File

@@ -0,0 +1,34 @@
from __future__ import annotations
import datetime
from typing import Dict, List, Optional, Set, Union
from pydantic import BaseModel, EmailStr, Field, PrivateAttr, ValidationError, validator
from ..lib_general import *
from ..db_sql import redis_lookup_id_random, sql_select
from ..models.membership_models import Membership_Cfg_Base
# ### BEGIN ### API Account Methods ### load_account_cfg_obj() ###
def load_account_cfg_obj(account_id:int|str, inc_event_cfg:bool=False, inc_fundraising_cfg:bool=False, inc_membership_cfg:bool=False):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if account_id := redis_lookup_id_random(record_id_random=account_id, table_name='account'): pass
else: return False
if inc_membership_cfg:
if membership_cfg_rec := sql_select(table_name='v_membership_cfg', field_name='account_id', field_value=account_id):
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(membership_cfg_rec)
#account_rec['membership_cfg'] = membership_cfg_rec
try:
membership_cfg_obj = Membership_Cfg_Base(**membership_cfg_rec)
log.debug(membership_cfg_obj)
except ValidationError as e:
log.error(e.json())
return membership_cfg_obj
# ### END ### API Account Methods ### load_account_cfg_obj() ###

59
app/methods/event_person_methods.py Normal file
View File

@@ -0,0 +1,59 @@
from __future__ import annotations
import datetime
from typing import Dict, List, Optional, Set, Union
from pydantic import BaseModel, EmailStr, Field, PrivateAttr, ValidationError, validator
from ..lib_general import *
from ..db_sql import redis_lookup_id_random, sql_insert, sql_select, sql_update
from ..models.event_person_models import Event_Person_New_Base, Event_Person_Base
# ### BEGIN ### API Event Person Methods ### create_event_person_obj() ###
def create_event_person_obj(event_person_obj_new:Event_Person_Base) -> int|bool:
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if not event_person_obj_new:
return False
event_person_obj_data = event_person_obj_new.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'created_on', 'updated_on'})
log.debug(event_person_obj_data)
if event_person_obj_in_result := sql_insert(data=event_person_obj_data, table_name='event_person', rm_id_random=True, id_random_length=8): pass
else:
return False
log.setLevel(logging.DEBUG)
log.debug(event_person_obj_in_result)
event_person_id = event_person_obj_in_result
log.debug(f'Returning the new event_person_id: {event_person_id}')
return event_person_id
# ### END ### API Event Person Methods ### create_event_person_obj() ###
# ### BEGIN ### API Event Person Methods ### load_event_person_obj() ###
def load_event_person_obj(event_person_id:int|str, inc_person:bool=False, inc_user:bool=False, inc_event_person_detail:bool=False, inc_event_registration:bool=False, inc_event_badge:bool=False) -> Event_Person_Base|bool:
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if event_person_id := redis_lookup_id_random(record_id_random=event_person_id, table_name='event_person'): pass
else: return False
if event_person_rec := sql_select(table_name='v_event_person', record_id=event_person_id):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(event_person_rec)
else:
return False
try:
event_person_obj = Event_Person_Base(**event_person_rec)
log.debug(event_person_obj)
except ValidationError as e:
log.error(e.json())
return False
return event_person_obj

226
app/methods/membership_methods.py Normal file
View File

@@ -0,0 +1,226 @@
from __future__ import annotations
import datetime
from typing import Dict, List, Optional, Set, Union
from pydantic import BaseModel, EmailStr, Field, PrivateAttr, ValidationError, validator
from ..lib_general import *
from ..db_sql import redis_lookup_id_random, sql_select
#from .address_model import Address_Base
#from .contact_model import Contact_Base
from ..models.membership_models import Membership_Base
#from .organization_model import Organization_Base
#from .person_model import Person_Base
#from .user_model import User_Base
# ### BEGIN ### API Membership Methods ### save_membership_obj() ###
def save_membership_obj(order_obj_new:Membership_Base=None):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if not order_obj_new:
return False
log.debug(order_obj_new.dict(by_alias=False, exclude_defaults=False, exclude_unset=True))
order_line_obj_li_curr = [] # Initialize to store order_line list
if order_obj_new.id_random:
log.info(f'An order.id {order_obj_new.id} or order.id_random {order_obj_new.id_random} was included. We can update an existing order.')
log.info(f'Get the current order_line list to compare with what was sent...')
data = {}
data['order_id_random'] = order_obj_new.id_random
if order_line_rec_li_curr := sql_select(table_name='v_order_line', data=data, rm_id_random=True, as_list=True):
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(order_line_rec_li_curr)
for order_line_rec in order_line_rec_li_curr:
try:
order_line_obj = Order_Line_Base(**order_line_rec)
log.debug(order_line_obj)
except ValidationError as e:
log.error(e.json())
order_line_obj_li_curr.append(order_line_obj)
else:
log.info(f'No order_line records were found')
elif order_obj_new.account_id_random and (order_obj_new.person_id_random or order_obj_new.user_id_random):
log.info(f'An account.id_random {order_obj_new.account_id_random} was passed. And either a person.id_random {order_obj_new.person_id_random} or user.id_random {order_obj_new.user_id_random} was passed. We can create a new order.')
# Because there was not an order ID, assume there are no order lines yet. So no look up.
else:
log.info('Either an order ID is required to update an order or an account ID along with a person ID or user ID is required to create an order.')
return False
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(order_line_obj_li_curr)
if repl_order_line_list: # This will remove any order line list items not sent with the new order information.
log.info('Removing any order line list items not sent with the new order information...')
for index, order_line_obj_curr in enumerate(order_line_obj_li_curr):
log.info(f'Current: order line ID={order_line_obj_curr.id_random} and product ID= {order_line_obj_curr.product_id_random}')
matched_product_id = False
for order_line_obj_new in order_obj_new.order_line_list:
log.debug(f'Checking new: product ID={order_line_obj_new.product_id_random}')
if order_line_obj_curr.product_id_random == order_line_obj_new.product_id_random:
matched_product_id = True
log.debug(f'Matched: product ID={order_line_obj_new.product_id_random}')
break
else:
log.debug(f'No match: product ID={order_line_obj_new.product_id_random}')
if not matched_product_id: # Was not found in the new order line list sent
log.info(f'Current order line product ID did not match any of the new list. DELETE order line ID {order_line_obj_curr.id_random} with product ID {order_line_obj_curr.product_id_random}')
if order_line_del_result := sql_delete(table_name='order_line', record_id_random=order_line_obj_curr.id_random):
log.info(f'Deleted record and now pop the current list item {index}...')
order_line_obj_li_curr.pop(index)
else:
log.info(f'Current order line product ID matched. Keeping order line ID {order_line_obj_curr.id_random} with product ID {order_line_obj_curr.product_id_random}')
# NOTE: That this current order line item will be updated below.
log.debug(order_line_obj_li_curr)
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.info('Loop through the line list that was sent and compare with what was pulled from the DB')
# Loop through the new line list that was sent and compare with the current line list that was pulled from the DB
# Only insert if a product ID does not match
# Only update if a product ID does match
for order_line_obj_new in order_obj_new.order_line_list:
log.info(f'New: order line ID={order_line_obj_new.id_random} and product ID= {order_line_obj_new.product_id_random}')
matched_product_id = False
for index, order_line_obj_curr in enumerate(order_line_obj_li_curr):
log.debug(f'Checking current: product ID={order_line_obj_curr.product_id_random}')
if order_line_obj_new.product_id_random == order_line_obj_curr.product_id_random:
matched_product_id = True
log.debug(f'Matched: product ID={order_line_obj_curr.product_id_random}')
log.info(f'Updating the current line item with the new line item.')
order_line_obj_new.id_random = order_line_obj_curr.id_random
order_line_obj_new.id = order_line_obj_curr.id
order_line_obj_li_curr[index] = order_line_obj_new
break
else:
log.debug(f'No match: product ID={order_line_obj_curr.product_id_random}')
if not matched_product_id: # Was not found in the current order line list that was pulled from the DB
log.info(f'New order line product ID did not match any of the current list. Append order line ID {order_line_obj_new.id_random} with product ID {order_line_obj_new.product_id_random}')
log.info('Append to current list...')
order_line_obj_li_curr.append(order_line_obj_new) # These will be inserted/updated below
# Save merged current and new list to the new order object
order_obj_new.order_line_list = order_line_obj_li_curr
log.debug(order_obj_new)
# Final loop through to get the new order totals
# Calculate totals
order_total_amount:int = 0
order_total_quantity:int = 0
for order_line_obj_new in order_obj_new.order_line_list:
order_total_amount += order_line_obj_new.quantity * order_line_obj_new.amount
order_total_quantity += order_line_obj_new.quantity
order_obj_new.total_bill = order_total_amount # "amount" is used by order_cart and Stripe
order_obj_new.total_quantity = order_total_quantity
order_obj_new.balance = order_total_amount - order_obj_new.total_paid
log.debug(order_obj_new.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'order_line_list', 'cfg', 'created_on', 'updated_on'}))
order_obj_data = order_obj_new.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'order_line_list', 'cfg', 'created_on', 'updated_on'})
# SQL INSERT or UPDATE the order record
log.info('SQL INSERT or UPDATE the order record')
if order_obj_resp := sql_insert_or_update(data=order_obj_data, table_name='order', rm_id_random=True, id_random_length=8): pass
else: return False
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(order_obj_resp)
if isinstance(order_obj_resp, bool) and order_obj_resp:
if order_id := order_obj_new.id: pass
elif order_id := order_obj_new.id_random: pass
elif isinstance(order_obj_resp, int):
order_id = order_obj_resp
else:
return False
log.debug(f'Order ID={order_id}')
# Loop through the order_line list to SQL INSERT or UPDATE the records
log.info('Loop through the order_line list to SQL INSERT or UPDATE the records')
for order_line_obj_new in order_obj_new.order_line_list:
log.info(f"New order_line: order_line_id_random={order_line_obj_new.id_random}; product_id_random={order_line_obj_new.product_id_random}")
log.debug(order_line_obj_new.dict(by_alias=False, exclude_defaults=False, exclude_unset=False, exclude={'order_line_id_random', 'product_type_id', 'product_type', 'created_on', 'updated_on'}))
order_line_obj_data = order_line_obj_new.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'order_line_id_random', 'product_type_id', 'product_type', 'created_on', 'updated_on'})
order_line_obj_data['order_id'] = order_id
if order_line_obj_resp := sql_insert_or_update(sql=None, data=order_line_obj_data, table_name='order_line', rm_id_random=True, id_random_length=8): pass
else: return False
log.debug(order_line_obj_resp)
return order_id
# ### END ### API Membership Methods ### save_membership_obj() ###
# ### BEGIN ### API Membership Methods ### load_membership_obj() ###
def load_membership_obj(membership_id:int|str, inc_membership_profile:bool=False, inc_membership_cfg:bool=False, inc_extended_profile:bool=False, inc_person:bool=False, inc_user:bool=False) -> Membership_Base:
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if membership_id := redis_lookup_id_random(record_id_random=membership_id, table_name='membership'): pass
else: return False
if membership_rec := sql_select(table_name='v_membership', record_id=membership_id):
#log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(membership_rec)
if inc_membership_profile:
if membership_profile_rec := sql_select(table_name='v_membership_profile', field_name='membership_id', field_value=membership_id):
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(membership_profile_rec)
membership_rec['profile'] = membership_profile_rec
if inc_membership_cfg:
if membership_cfg_rec := sql_select(table_name='v_membership_cfg', field_name='account_id', field_value=membership_rec.get('account_id', None)):
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(membership_cfg_rec)
membership_rec['cfg'] = membership_cfg_rec
if inc_extended_profile:
account_code = membership_rec.get('account_code', None)
table_name = f'c_{account_code}_membership_profile'
if extended_profile_rec := sql_select(table_name=table_name, field_name='membership_id', field_value=membership_id):
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(extended_profile_rec)
membership_rec['extended_profile'] = extended_profile_rec
if inc_person:
if person_rec := sql_select(table_name='v_person', record_id=membership_rec.get('person_id')):
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(person_rec)
membership_rec['person'] = person_rec
if inc_user:
if user_rec := sql_select(table_name='v_user', record_id=membership_rec.get('user_id')):
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(user_rec)
membership_rec['user'] = user_rec
#log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(membership_rec)
else:
return False
try:
membership_obj = Membership_Base(**membership_rec)
log.debug(membership_obj)
except ValidationError as e:
log.error(e.json())
return membership_obj
# ### END ### API Membership Methods ### load_membership_obj() ###