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

Work on person methods and related and event badge everything

Browse Source
This commit is contained in:
Scott Idem
2021-09-08 15:58:35 -04:00
parent d0d8392817
commit 54a43cc124
12 changed files with 1071 additions and 417 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
app/main.py
View File

@@ -18,7 +18,7 @@ from app.lib_general import log, logging
from app.log import log
# Import the routers here first:
from app.routers import api_crud, api, importing, account, address, archive, archive_content, contact, cont_edu_cert, cont_edu_cert_person, event, event_exhibit, event_file, event_person, event_person_detail, event_presentation, event_presenter, event_registration, event_session, flask_cfg, hosted_file, journal, journal_entry, log_client_viewing, lookup, membership_cfg, membership_group, membership_group_person, membership_person, membership_person_profile, membership_type, membership_type_person, order, order_cart, organization, page, person, post, post_comment, product, site, site_domain, user, user_person, websockets # , items, journals
from app.routers import api_crud, api, importing, account, address, archive, archive_content, contact, cont_edu_cert, cont_edu_cert_person, event, event_badge, event_badge_template, event_exhibit, event_file, event_person, event_person_detail, event_presentation, event_presenter, event_registration, event_session, flask_cfg, hosted_file, journal, journal_entry, log_client_viewing, lookup, membership_cfg, membership_group, membership_group_person, membership_person, membership_person_profile, membership_type, membership_type_person, order, order_cart, organization, page, person, post, post_comment, product, site, site_domain, user, user_person, websockets # , items, journals
from app.db_sql import db
@@ -112,6 +112,16 @@ app.include_router(
# prefix='/event',
tags=['Event'],
)
app.include_router(
event_badge.router,
# prefix='/event/badge',
tags=['Event Badge'],
)
app.include_router(
event_badge_template.router,
# prefix='/event/badge/template',
tags=['Event Badge Template'],
)
app.include_router(
event_exhibit.router,
prefix='/event/exhibit',

42
app/methods/event_badge_template_methods.py Normal file
View File

@@ -0,0 +1,42 @@
from __future__ import annotations
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.lib_general import log, logging
from app.models.event_badge_template_models import Event_Badge_Template_Base
# ### BEGIN ### API Event Badge Template Methods ### load_event_badge_template_obj() ###
def load_event_badge_template_obj(
event_badge_template_id: int|str,
limit: int = 1000,
by_alias: bool = True,
exclude_unset: bool = True,
model_as_dict: bool = False,
enabled: str = 'enabled', # enabled, disabled, all
) -> Event_Badge_Template_Base|dict|bool:
# log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if event_badge_template_id := redis_lookup_id_random(record_id_random=event_badge_template_id, table_name='event_badge_template'): pass
else: return False
if event_badge_template_rec := sql_select(table_name='v_event_badge_template', record_id=event_badge_template_id): pass
else: return False
try:
event_badge_template_obj = Event_Badge_Template_Base(**event_badge_template_rec)
log.debug(event_badge_template_obj)
except ValidationError as e:
log.error(e.json())
return False
if model_as_dict:
return event_badge_template_obj.dict(by_alias=by_alias, exclude_unset=exclude_unset) # pylint: disable=no-member
else:
return event_badge_template_obj
# ### END ### API Event Badge Template Methods ### load_event_badge_template_obj() ###

12
app/methods/event_methods.py
View File

@@ -11,7 +11,7 @@ from app.methods.address_methods import load_address_obj
from app.methods.contact_methods import load_contact_obj
from app.methods.event_cfg_methods import load_event_cfg_obj
from app.methods.event_session_methods import get_event_session_rec_list, load_event_session_obj
from app.methods.person_methods import create_person_obj_v3, load_person_obj, update_person_obj
from app.methods.person_methods import create_update_person_obj_v4b, load_person_obj
from app.methods.user_methods import create_user_obj, load_user_obj, update_user_obj
from app.models.event_models import Event_Base
@@ -623,10 +623,10 @@ def update_event_obj(
poc_person_obj_up = event_obj_up.poc_person
log.debug(poc_person_id)
log.debug(poc_person_obj_up)
if poc_person_obj_up_result := update_person_obj(
poc_person_id = poc_person_id,
poc_person_obj_up = poc_person_obj_up,
create_sub_obj = create_sub_obj,
if poc_person_obj_up_result := create_update_person_obj_v4b(
account_id = account_id,
person_dict_obj = poc_person_obj_up,
person_id = poc_person_id,
):
log.debug(poc_person_obj_up_result)
else:
@@ -636,7 +636,7 @@ def update_event_obj(
# NOTE: This will blindly create a new person even if there was one associated but the event.poc_person_id was not found.
poc_person_obj_in = event_obj_up.poc_person
log.debug(poc_person_obj_in)
if poc_person_obj_in_result := create_person_obj_v3(person_obj_new=poc_person_obj_in):
if poc_person_obj_in_result := create_update_person_obj_v4b(account_id=account_id, person_dict_obj=poc_person_obj_in):
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(poc_person_obj_in_result)
event_obj_up.poc_person_id = poc_person_obj_in_result

696
app/methods/person_methods.py
View File

@@ -304,6 +304,257 @@ def get_person_rec_list(
# ### END ### API Person Methods ### get_person_rec_list() ###
# ### BEGIN ### API Person Methods ### create_update_person_obj_v4b() ###
def create_update_person_obj_v4b(
account_id: int|str,
person_dict_obj: Person_Base,
person_id: int|str|None = None,
process_contact: bool = True, # For future v5
process_organization: bool = True, # For future v5
process_user: bool = True, # For future v5
create_sub_obj: bool = True, # For future v5
fail_any: bool = False, # Fail if any thing goes wrong for sub objects
return_outline: bool = False, # For future v5
) -> bool:
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
log.info('Checking requirements...')
if account_id := redis_lookup_id_random(record_id_random=account_id, table_name='account'): log.info(f'Account ID: {account_id}')
else:
log.error('Missing or invalid Account ID passed. Failed requirement.')
log.error(f'Account ID: {account_id}')
return False
if person_id:
log.info(f'Person ID passed. Update existing Person below. Person ID: {person_id}')
if person_id := redis_lookup_id_random(record_id_random=person_id, table_name='person'): pass
else:
log.error('Person ID passed but is invalid. Failed requirement.')
return False
# person_obj.id = person_id
else:
log.info('No Person ID passed. Create new Person below. Required: Account ID')
log.debug(type(person_dict_obj))
log.info('Create dictionary or Pydantic object variables...')
if isinstance(person_dict_obj, dict):
person_dict = person_dict_obj
person_dict['account_id'] = account_id
if person_id:
person_dict['person_id'] = person_id
try:
person_obj = Person_Base(**person_dict)
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(person_obj)
except ValidationError as e:
log.error(e.json())
return False
else:
person_obj = person_dict_obj
person_obj.account_id = account_id
if person_id:
# NOTE: Can't update the ID alias if it was never set.
person_obj.id = person_id
person_dict = person_obj.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'contact', 'contact_id', 'contact_id_random', 'email', 'cc_email', 'membership_person_id', 'membership_person_id_random', 'organization', 'user', 'created_on', 'updated_on'})
log.info(f'SQL INSERT or UPDATE record... Person ID: {person_id}')
if person_id:
if person_dict_up_result := sql_update(data=person_dict, table_name='person', rm_id_random=True): pass
else:
log.warning(f'Person not updated. Person ID: {person_id}')
log.debug(person_dict_up_result)
return False
log.debug(person_dict_up_result)
else:
if person_dict_in_result := sql_insert(data=person_dict, table_name='person', rm_id_random=True, id_random_length=default_num_bytes): pass
else:
log.warning(f'Person not created.')
log.debug(person_dict_in_result)
return False
log.debug(person_dict_in_result)
person_id = person_dict_in_result
person_outline = {}
person_outline['person_id'] = person_id
person_outline['contact_id'] = None
person_outline['contact'] = {}
person_outline['contact']['address_id'] = None
person_outline['organization_id'] = None
person_outline['user_id'] = None
# Updated 2021-09-08
log.info(f'Check if processing Contact data...')
# NOTE: Use object model version because of better type checking and validations
if process_contact and person_obj.contact:
person_outline['contact_id'] = None
contact_obj = person_obj.contact
if contact_id := person_obj.contact_id : pass
elif contact_id := contact_obj.id: pass
else: contact_id = None
contact_obj.id
contact_obj.for_type = 'person'
contact_obj.for_id = person_id
create_update_contact_obj_result = create_update_contact_obj_v4(
account_id = account_id,
contact_dict_obj = contact_obj,
contact_id = contact_id,
for_type = 'person',
for_id = person_id,
# process_address = process_address,
# create_sub_obj = create_sub_obj,
fail_any = fail_any,
return_outline = return_outline,
)
if isinstance(create_update_contact_obj_result, int):
contact_id = create_update_contact_obj_result
elif create_update_contact_obj_result == True: pass
else:
log.warning(f'Create or Update failed while trying create_update_contact_obj_v4(): {create_update_contact_obj_result}')
contact_id = None
person_outline['contact_id'] = contact_id
# Updated 2021-09-08
log.info(f'Check if processing Organization data...')
if process_organization and person_obj.organization:
organization_obj = person_obj.organization
organization_id = person_obj.organization_id_random
update_person_obj = False
organization_result = create_update_organization_obj(
organization_id = organization_id,
organization_obj = organization_obj,
process_contact = True, # Setting to True under the assumption that if there is organization information then there is probably a contact (and address).
)
log.debug(organization_result)
if isinstance(organization_result, bool) and organization_result is True:
pass # Do not need to update person object.
elif isinstance(organization_result, bool) and organization_result is False:
pass # Do not need to update person object.
elif isinstance(organization_result, int):
person_obj.organization_id = organization_result
organization_id = organization_result
else:
log.warning('Something may have gone wrong while trying to create or update a organization.')
log.info(f'Check if need to update the Person with the Organization ID... Update Person Obj: {update_person_obj} Organization ID: {organization_id}')
if update_person_obj and isinstance(organization_id, int):
log.info(f'Updating the Person with the new/current Organization ID. Organization ID: {organization_id}')
person_data_up = {}
person_data_up['id'] = person_id
person_data_up['organization_id'] = organization_id
if person_data_up_result := sql_update(data=person_data_up, table_name='person'):
log.info(f'Person updated with Organization ID. Person ID: {person_id} Organization ID: {organization_id}')
else:
log.error(f'Person not updated with current Organization ID. Person ID: {person_id} Organization ID: {organization_id}')
if fail_any: return False
log.debug(person_dict_up_result)
person_outline['organization_id'] = organization_id
# Updated 2021-09-08
log.info(f'Check if processing User data... Process User: {process_user}')
if process_user and person_obj.user:
log.info(f'User data was found. Create a new User and link it to the new Person or update existing User. Person ID: {person_id}')
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
user_obj = person_obj.user
user_id = person_obj.user_id
update_person_obj = False
log.debug(user_obj)
if user_id:
log.warning('User ID found. This is not expected, but should be ok.')
# from app.methods.user_methods import update_user_obj_v3
from app.methods.user_methods import update_user_obj
if update_user_obj_result := update_user_obj(
user_id = user_id,
user_obj_up = user_obj,
# create_sub_obj = create_sub_obj,
# fail_any = fail_any,
):
log.info(f'User updated. User ID: {user_id}')
log.debug(update_user_obj_result)
# NOTE: This should not be needed. Updating person just in case!
update_person_obj = True
else:
log.warning(f'User not updated. Person ID: {person_id}')
log.debug(update_user_obj_result)
user_id = None
if fail_any: return False
else:
log.info(f'No User ID found.')
# from app.methods.user_methods import create_user_obj_v3
from app.methods.user_methods import create_user_obj
if create_user_obj_result := create_user_obj(
account_id = account_id,
user_obj_new = user_obj,
# create_sub_obj = create_sub_obj,
# fail_any = fail_any,
):
if isinstance(create_user_obj_result, int):
log.info(f'User created. User ID: {user_id}')
log.debug(update_user_obj_result)
user_id = create_user_obj_result
# Need to update the person with the new user_id
update_person_obj = True
else:
log.warning(f'User not created. Updated? Person ID: {person_id}')
log.debug(create_user_obj_result)
user_id = None
if fail_any: return False
else:
log.warning(f'User not created. Person ID: {person_id}')
log.debug(create_user_obj_result)
user_id = None
if fail_any: return False
log.info(f'Check if need to update the Person with the User ID... Update Person Obj: {update_person_obj} User ID: {user_id}')
if update_person_obj and isinstance(user_id, int):
log.info(f'Updating the Person with the new/current User ID. User ID: {user_id}')
person_data_up = {}
person_data_up['id'] = person_id
person_data_up['user_id'] = user_id
if person_data_up_result := sql_update(data=person_data_up, table_name='person'):
log.info(f'Person updated with User ID. Person ID: {person_id} User ID: {user_id}')
else:
log.error(f'Person not updated with current User ID. Person ID: {person_id} User ID: {user_id}')
if fail_any: return False
log.debug(person_dict_up_result)
person_outline['user_id'] = user_id
else:
log.info('User data not found.')
pass
return person_id
# Process person data
person_dict_up = person_obj.dict(by_alias=False, exclude_unset=True, exclude={'contact_id', 'contact_id_random', 'contact', 'organization', 'user'})
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(person_obj)
log.debug(person_dict_up)
# Update record
person_up_result = sql_update(
data = person_dict_up,
table_name = 'person',
rm_id_random = True,
)
log.debug(person_up_result)
if isinstance(person_up_result, bool) and person_up_result is True:
return person_id
elif isinstance(person_up_result, bool) and person_up_result is False:
return False
elif isinstance(person_up_result, int):
return person_up_result
else:
return False
# ### END ### API Person Methods ### create_update_person_obj_v4b() ###
# ### BEGIN ### API Person Methods ### get_person_rec_w_external_id() ###
# Updated 2021-08-19
def get_person_rec_w_external_id(
@@ -371,6 +622,8 @@ def get_account_id_w_person_id(
# ### END ### API Person Methods ### get_account_id_w_person_id() ###
# ### BEGIN ### API Person Methods ### create_person_obj_v3() ###
# NOTE: This will create a person and then also create a linked contact if person_obj.contact data is passed. The create_contact_obj will create a contact and then also create a linked address if person_obj.contact.address data is passed.
# Updated 2021-08-25
@@ -1006,361 +1259,104 @@ def update_person_obj(
# ### END ### API Person Methods ### update_person_obj() ###
# ### BEGIN ### API Person Methods ### create_update_person_obj_v4b() ###
def create_update_person_obj_v4b(
account_id: int|str,
person_dict_obj: Person_Base,
person_id: int|str|None = None,
process_contact: bool = True, # For future v5
process_organization: bool = True, # For future v5
process_user: bool = True, # For future v5
create_sub_obj: bool = True, # For future v5
fail_any: bool = False, # Fail if any thing goes wrong for sub objects
return_outline: bool = False, # For future v5
) -> bool:
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
log.info('Checking requirements...')
if account_id := redis_lookup_id_random(record_id_random=account_id, table_name='account'): log.info(f'Account ID: {account_id}')
else:
log.error('Missing or invalid Account ID passed. Failed requirement.')
log.error(f'Account ID: {account_id}')
return False
# # ### BEGIN ### API Person Methods ### create_update_person_obj_v4b() ###
# def create_update_person_obj_v4b(
# person_id: int|str|None, # Ideally the int ID should be passed. This allows for updating of the id_random value.
# person_obj: Person_Base,
# process_contact: bool = False,
# process_organization: bool = False,
# ) -> bool:
# log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
# log.debug(locals())
if person_id:
log.info(f'Person ID passed. Update existing Person below. Person ID: {person_id}')
# if person_id:
# log.info(f'Person ID passed. Update existing Person. Person ID: {person_id}')
if person_id := redis_lookup_id_random(record_id_random=person_id, table_name='person'): pass
else:
log.error('Person ID passed but is invalid. Failed requirement.')
return False
# person_obj.id = person_id
else:
log.info('No Person ID passed. Create new Person below. Required: Account ID')
# if person_id := redis_lookup_id_random(record_id_random=person_id, table_name='person'): pass
# else: return False
# person_obj.id = person_id
# else:
# # Insert record now and update later
# log.info('No Person ID passed. Create new Person. Required: Account ID')
log.debug(type(person_dict_obj))
log.info('Create dictionary or Pydantic object variables...')
if isinstance(person_dict_obj, dict):
person_dict = person_dict_obj
person_dict['account_id'] = account_id
if person_id:
person_dict['person_id'] = person_id
try:
person_obj = Person_Base(**person_dict)
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(person_obj)
except ValidationError as e:
log.error(e.json())
return False
else:
person_obj = person_dict_obj
person_obj.account_id = account_id
if person_id:
# NOTE: Can't update the ID alias if it was never set.
person_obj.id = person_id
# person_dict_in = person_obj.dict(by_alias=False, exclude_unset=True, exclude={'contact_id', 'contact_id_random', 'contact', 'organization', 'user'})
# log.debug(person_dict_in)
# person_in_result = sql_insert(
# data = person_dict_in,
# table_name = 'person',
# rm_id_random = True,
# id_random_length = default_num_bytes,
# )
# log.debug(person_in_result)
# if isinstance(person_in_result, bool) and person_in_result is True:
# return person_in_result
# elif isinstance(person_in_result, int):
# person_id = person_in_result
# person_obj.id = person_id
# else:
# return False # This should not happen.
person_dict = person_obj.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'contact', 'organization', 'user', 'created_on', 'updated_on'})
# # Process contact data
# if process_contact and person_obj.contact:
# contact_obj = person_obj.contact
# contact_obj.for_type = 'person'
# contact_obj.for_id = person_id
# contact_id = person_obj.contact_id_random
# contact_result = create_update_contact_obj(
# contact_id = contact_id,
# contact_obj = contact_obj,
# process_address = True, # Setting to True under the assumption that if there is contact information then there is probably an address.
# )
# log.debug(contact_result)
# if isinstance(contact_result, bool) and contact_result is True:
# pass # Do not need to update person object.
# elif isinstance(contact_result, bool) and contact_result is False:
# pass # Do not need to update person object.
# elif isinstance(contact_result, int):
# person_obj.contact_id = contact_result
# # pass # Do not need to update person object.
# else:
# log.warning('Something may have gone wrong while trying to create or update a contact.')
log.info(f'SQL INSERT or UPDATE record... Person ID: {person_id}')
if person_id:
if person_dict_up_result := sql_update(data=person_dict, table_name='person', rm_id_random=True): pass
else:
log.warning(f'Person not updated. Person ID: {person_id}')
log.debug(person_dict_up_result)
return False
log.debug(person_dict_up_result)
else:
if person_dict_in_result := sql_insert(data=person_dict, table_name='person', rm_id_random=True, id_random_length=default_num_bytes): pass
else:
log.warning(f'Person not created.')
log.debug(person_dict_in_result)
return False
log.debug(person_dict_in_result)
# # Process organization data
# if process_organization and person_obj.organization:
# organization_obj = person_obj.organization
# organization_id = person_obj.organization_id_random
# organization_result = create_update_organization_obj(
# organization_id = organization_id,
# organization_obj = organization_obj,
# process_contact = True, # Setting to True under the assumption that if there is organization information then there is probably a contact (and address).
# )
# log.debug(organization_result)
# if isinstance(organization_result, bool) and organization_result is True:
# pass # Do not need to update person object.
# elif isinstance(organization_result, bool) and organization_result is False:
# pass # Do not need to update person object.
# elif isinstance(organization_result, int):
# person_obj.organization_id = organization_result
# else:
# log.warning('Something may have gone wrong while trying to create or update a organization.')
person_id = person_dict_in_result
# # Process person data
# person_dict_up = person_obj.dict(by_alias=False, exclude_unset=True, exclude={'contact_id', 'contact_id_random', 'contact', 'organization', 'user'})
# # log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
# log.debug(person_obj)
# log.debug(person_dict_up)
person_outline = {}
person_outline['person_id'] = person_id
# Updated 2021-09-08
log.info(f'Check if processing Contact data...')
# NOTE: Use object model version because of better type checking and validations
if process_contact and person_obj.contact:
person_outline['contact_id'] = None
contact_obj = person_obj.contact
if contact_id := person_obj.contact_id : pass
elif contact_id := contact_obj.id: pass
else: contact_id = None
contact_obj.id
contact_obj.for_type = 'person'
contact_obj.for_id = person_id
create_update_contact_obj_result = create_update_contact_obj_v4(
account_id = account_id,
contact_dict_obj = contact_obj,
contact_id = contact_id,
for_type = 'person',
for_id = person_id,
# process_address = process_address,
# create_sub_obj = create_sub_obj,
fail_any = fail_any,
return_outline = return_outline,
)
if isinstance(create_update_contact_obj_result, int):
contact_id = create_update_contact_obj_result
elif create_update_contact_obj_result == True: pass
else:
log.warning(f'Create or Update failed while trying create_update_contact_obj_v4(): {create_update_contact_obj_result}')
contact_id = None
person_outline['contact_id'] = contact_id
# Updated 2021-09-08
log.info(f'Check if processing Organization data...')
if process_organization and person_obj.organization:
organization_obj = person_obj.organization
organization_id = person_obj.organization_id_random
organization_result = create_update_organization_obj(
organization_id = organization_id,
organization_obj = organization_obj,
process_contact = True, # Setting to True under the assumption that if there is organization information then there is probably a contact (and address).
)
log.debug(organization_result)
if isinstance(organization_result, bool) and organization_result is True:
pass # Do not need to update person object.
elif isinstance(organization_result, bool) and organization_result is False:
pass # Do not need to update person object.
elif isinstance(organization_result, int):
person_obj.organization_id = organization_result
organization_id = organization_result
else:
log.warning('Something may have gone wrong while trying to create or update a organization.')
log.info(f'Check if need to update the Person with the Organization ID... Update Person Obj: {update_person_obj} Organization ID: {organization_id}')
if update_person_obj and isinstance(organization_id, int):
log.info(f'Updating the Person with the new/current Organization ID. Organization ID: {organization_id}')
person_data_up = {}
person_data_up['id'] = person_id
person_data_up['organization_id'] = organization_id
if person_data_up_result := sql_update(data=person_data_up, table_name='person'):
log.info(f'Person updated with Organization ID. Person ID: {person_id} Organization ID: {organization_id}')
else:
log.error(f'Person not updated with current Organization ID. Person ID: {person_id} Organization ID: {organization_id}')
if fail_any: return False
log.debug(person_dict_up_result)
person_outline['organization_id'] = organization_id
# Updated 2021-09-08
log.info(f'Check if processing User data... Process User: {process_user}')
if process_user and person_obj.user:
log.info(f'User data was found. Create a new User and link it to the new Person or update existing User. Person ID: {person_id}')
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
user_obj = person_obj.user
user_id = person_obj.user_id
update_person_obj = False
log.debug(user_obj)
if user_id:
log.warning('User ID found. This is not expected, but should be ok.')
# from app.methods.user_methods import update_user_obj_v3
from app.methods.user_methods import update_user_obj
if update_user_obj_result := update_user_obj(
user_id = user_id,
user_obj_up = user_obj,
# create_sub_obj = create_sub_obj,
# fail_any = fail_any,
):
log.info(f'User updated. User ID: {user_id}')
log.debug(update_user_obj_result)
# NOTE: This should not be needed. Updating person just in case!
update_person_obj = True
else:
log.warning(f'User not updated. Person ID: {person_id}')
log.debug(update_user_obj_result)
user_id = None
if fail_any: return False
# if isinstance(update_user_obj_result, int):
# user_id = update_user_obj_result
# log.info(f'User updated. User ID: {user_id}')
# # Need to update the person with the current user_id
# log.debug(person_dict_up_result)
# else:
# log.warning(f'User not updated. Person ID: {person_id}')
# log.debug(update_user_obj_result)
# user_id = None
# if fail_any: return False
else:
log.info(f'No User ID found.')
# from app.methods.user_methods import create_user_obj_v3
from app.methods.user_methods import create_user_obj
if create_user_obj_result := create_user_obj(
account_id = account_id,
user_obj_new = user_obj,
# create_sub_obj = create_sub_obj,
# fail_any = fail_any,
):
if isinstance(create_user_obj_result, int):
log.info(f'User created. User ID: {user_id}')
log.debug(update_user_obj_result)
user_id = create_user_obj_result
# Need to update the person with the new user_id
update_person_obj = True
else:
log.warning(f'User not created. Updated? Person ID: {person_id}')
log.debug(create_user_obj_result)
user_id = None
if fail_any: return False
else:
log.warning(f'User not created. Person ID: {person_id}')
log.debug(create_user_obj_result)
user_id = None
if fail_any: return False
log.info(f'Check if need to update the Person with the User ID... Update Person Obj: {update_person_obj} User ID: {user_id}')
if update_person_obj and isinstance(user_id, int):
log.info(f'Updating the Person with the new/current User ID. User ID: {user_id}')
person_data_up = {}
person_data_up['id'] = person_id
person_data_up['user_id'] = user_id
if person_data_up_result := sql_update(data=person_data_up, table_name='person'):
log.info(f'Person updated with User ID. Person ID: {person_id} User ID: {user_id}')
else:
log.error(f'Person not updated with current User ID. Person ID: {person_id} User ID: {user_id}')
if fail_any: return False
log.debug(person_dict_up_result)
person_outline['user_id'] = user_id
else:
log.info('User data not found.')
pass
return person_id
# Process person data
person_dict_up = person_obj.dict(by_alias=False, exclude_unset=True, exclude={'contact_id', 'contact_id_random', 'contact', 'organization', 'user'})
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(person_obj)
log.debug(person_dict_up)
# Update record
person_up_result = sql_update(
data = person_dict_up,
table_name = 'person',
rm_id_random = True,
)
log.debug(person_up_result)
if isinstance(person_up_result, bool) and person_up_result is True:
return person_id
elif isinstance(person_up_result, bool) and person_up_result is False:
return False
elif isinstance(person_up_result, int):
return person_up_result
else:
return False
# ### END ### API Person Methods ### create_update_person_obj_v4b() ###
# ### BEGIN ### API Person Methods ### create_update_person_obj() ###
def create_update_person_obj(
person_id: int|str|None, # Ideally the int ID should be passed. This allows for updating of the id_random value.
person_obj: Person_Base,
process_contact: bool = False,
process_organization: bool = False,
) -> bool:
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if person_id:
log.info(f'Person ID passed. Update existing Person. Person ID: {person_id}')
if person_id := redis_lookup_id_random(record_id_random=person_id, table_name='person'): pass
else: return False
person_obj.id = person_id
else:
# Insert record now and update later
log.info('No Person ID passed. Create new Person. Required: Account ID')
person_dict_in = person_obj.dict(by_alias=False, exclude_unset=True, exclude={'contact_id', 'contact_id_random', 'contact', 'organization', 'user'})
log.debug(person_dict_in)
person_in_result = sql_insert(
data = person_dict_in,
table_name = 'person',
rm_id_random = True,
id_random_length = default_num_bytes,
)
log.debug(person_in_result)
if isinstance(person_in_result, bool) and person_in_result is True:
return person_in_result
elif isinstance(person_in_result, int):
person_id = person_in_result
person_obj.id = person_id
else:
return False # This should not happen.
# Process contact data
if process_contact and person_obj.contact:
contact_obj = person_obj.contact
contact_obj.for_type = 'person'
contact_obj.for_id = person_id
contact_id = person_obj.contact_id_random
contact_result = create_update_contact_obj(
contact_id = contact_id,
contact_obj = contact_obj,
process_address = True, # Setting to True under the assumption that if there is contact information then there is probably an address.
)
log.debug(contact_result)
if isinstance(contact_result, bool) and contact_result is True:
pass # Do not need to update person object.
elif isinstance(contact_result, bool) and contact_result is False:
pass # Do not need to update person object.
elif isinstance(contact_result, int):
person_obj.contact_id = contact_result
# pass # Do not need to update person object.
else:
log.warning('Something may have gone wrong while trying to create or update a contact.')
# Process organization data
if process_organization and person_obj.organization:
organization_obj = person_obj.organization
organization_id = person_obj.organization_id_random
organization_result = create_update_organization_obj(
organization_id = organization_id,
organization_obj = organization_obj,
process_contact = True, # Setting to True under the assumption that if there is organization information then there is probably a contact (and address).
)
log.debug(organization_result)
if isinstance(organization_result, bool) and organization_result is True:
pass # Do not need to update person object.
elif isinstance(organization_result, bool) and organization_result is False:
pass # Do not need to update person object.
elif isinstance(organization_result, int):
person_obj.organization_id = organization_result
else:
log.warning('Something may have gone wrong while trying to create or update a organization.')
# Process person data
person_dict_up = person_obj.dict(by_alias=False, exclude_unset=True, exclude={'contact_id', 'contact_id_random', 'contact', 'organization', 'user'})
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(person_obj)
log.debug(person_dict_up)
# Update record
person_up_result = sql_update(
data = person_dict_up,
table_name = 'person',
rm_id_random = True,
)
log.debug(person_up_result)
if isinstance(person_up_result, bool) and person_up_result is True:
return person_id
elif isinstance(person_up_result, bool) and person_up_result is False:
return False
elif isinstance(person_up_result, int):
return person_up_result
else:
return False
# ### END ### API Person Methods ### create_update_person_obj() ###
# # Update record
# person_up_result = sql_update(
# data = person_dict_up,
# table_name = 'person',
# rm_id_random = True,
# )
# log.debug(person_up_result)
# if isinstance(person_up_result, bool) and person_up_result is True:
# return person_id
# elif isinstance(person_up_result, bool) and person_up_result is False:
# return False
# elif isinstance(person_up_result, int):
# return person_up_result
# else:
# return False
# # ### END ### API Person Methods ### create_update_person_obj_v4b() ###

1
app/models/common_field_schema.py
View File

@@ -35,6 +35,7 @@ base_fields['event_file_id_random'] = xxx_id_random_field_schema
base_fields['event_id_random'] = xxx_id_random_field_schema
base_fields['event_abstract_id_random'] = xxx_id_random_field_schema
base_fields['event_badge_id_random'] = xxx_id_random_field_schema
base_fields['event_badge_template_id_random'] = xxx_id_random_field_schema
base_fields['event_cfg_id_random'] = xxx_id_random_field_schema # Not really ready yet
base_fields['event_device_id_random'] = xxx_id_random_field_schema
base_fields['event_location_id_random'] = xxx_id_random_field_schema

2
app/models/event_badge_models.py
View File

@@ -56,6 +56,8 @@ class Event_Badge_Base(BaseModel):
country: Optional[str]
location_name: Optional[str] # Actual location name shown on badge and other "public" areas
full_address: Optional[str]
# NOTE: More badge fields need to be added here once things are cleaned up
priority: Optional[bool]

97
app/models/event_badge_template_models.py Normal file
View File

@@ -0,0 +1,97 @@
from __future__ import annotations
import datetime, hashlib, logging, os, pytz, redis, secrets
from typing import Dict, List, Optional, Set, Union
from pydantic import BaseModel, EmailStr, Field, Json, PrivateAttr, ValidationError, validator
from app.db_sql import redis_lookup_id_random
from app.lib_general import log, logging
from app.models.common_field_schema import base_fields, default_num_bytes
class Event_Badge_Template_Base(BaseModel):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
id_random: Optional[str] = Field(
**base_fields['event_badge_template_id_random'],
alias = 'event_badge_template_id_random',
default_factory = lambda:secrets.token_urlsafe(default_num_bytes),
)
id: Optional[int] = Field(
alias = 'event_badge_template_id'
)
# account_id_random: Optional[str]
# account_id: Optional[int]
event_id_random: Optional[str]
event_id: Optional[int]
name: Optional[str]
description: Optional[str]
logo_filename: Optional[str]
header_row_1: Optional[str]
header_row_2: Optional[str]
header_background: Optional[str]
footer_title: Optional[str]
footer_left: Optional[str]
footer_right: Optional[str]
footer_background: Optional[str]
ticket_1_text: Optional[str]
ticket_2_text: Optional[str]
ticket_3_text: Optional[str]
ticket_4_text: Optional[str]
ticket_5_text: Optional[str]
ticket_6_text: Optional[str]
ticket_7_text: Optional[str]
ticket_8_text: Optional[str]
wireless_ssid: Optional[str]
wireless_password: Optional[str]
style_filename: Optional[str]
passcode: Optional[str]
notes: Optional[str]
created_on: Optional[datetime.datetime] = None
updated_on: Optional[datetime.datetime] = None
_processed_at: datetime.datetime = PrivateAttr(default_factory=datetime.datetime.now)
#@validator('event_badge_template_id_random', always=True)
def event_badge_template_id_random_copy(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['id_random']:
return values['id_random']
return None
@validator('id', always=True)
def event_badge_template_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['id_random']:
log.debug(values['id_random'])
return redis_lookup_id_random(record_id_random=values['id_random'], table_name='event_badge_template')
return None
@validator('event_id', always=True)
def event_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['event_id_random']:
return redis_lookup_id_random(record_id_random=values['event_id_random'], table_name='event')
return None
class Config:
underscore_attrs_are_private = True
allow_population_by_field_name = True
fields = base_fields

365
app/routers/event_badge.py Normal file
View File

@@ -0,0 +1,365 @@
import datetime
from fastapi import APIRouter, Body, Depends, Header, HTTPException, Query, Response, status
from pydantic import BaseModel, EmailStr, Field
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.routers.api_crud import delete_obj_template, get_obj_template, get_obj_li_template, patch_obj_template, post_obj_template
# from app.methods.event_badge_methods import get_event_badge_rec_list, load_event_badge_obj
from app.methods.event_badge_methods import load_event_badge_obj
from app.models.event_badge_models import Event_Badge_Base
from app.models.response_models import Resp_Body_Base, mk_resp
router = APIRouter()
@router.post('/event/badge', response_model=Resp_Body_Base)
async def post_event_badge_obj(
obj: Event_Badge_Base,
x_account_id: str = Header(...),
return_obj: Optional[bool] = True,
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
response: Response = Response,
):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
obj_type = 'event_badge'
obj_data_dict = obj.dict(by_alias=False, exclude_unset=True)
result = post_obj_template(
obj_type = obj_type,
data = obj_data_dict,
return_obj = True,
by_alias = True,
exclude_unset = True,
)
return result
@router.patch('/event/badge/{obj_id}', response_model=Resp_Body_Base)
async def patch_event_badge_obj(
obj: Event_Badge_Base,
obj_id: str = Query(..., min_length=1, max_length=22),
x_account_id: Optional[str] = Header(..., ),
return_obj: Optional[bool] = True,
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
response: Response = Response,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
obj_type = 'event_badge'
obj_data_dict = obj.dict(by_alias=False, exclude_unset=True)
obj_data_dict['id'] = redis_lookup_id_random(record_id_random=obj_id, table_name=obj_type)
obj_data_dict['id_random'] = obj_id
result = patch_obj_template(
obj_type = obj_type,
data = obj_data_dict,
obj_id = obj_id,
return_obj = True,
by_alias = True,
exclude_unset = True,
)
return result
@router.get('/event/badge/list', response_model=Resp_Body_Base)
async def get_event_badge_obj_li(
for_obj_type: str = Query(None, min_length=2, max_length=50),
for_obj_id: str = Query(None, min_length=1, max_length=22),
group: str = Query(None, min_length=2, max_length=50),
x_account_id: str = Header(...),
by_alias: bool = True,
exclude_unset: bool = True,
response: Response = Response,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
obj_type = 'event_badge'
if for_obj_type == 'event_exhibit' and for_obj_id:
#base_name = obj_type_li[obj_type]['base_name']
base_name = Event_Badge_Base
for_obj_id_random = for_obj_id
for_obj_id = redis_lookup_id_random(record_id_random=for_obj_id_random, table_name=for_obj_type)
data = {}
data['for_obj_type'] = for_obj_type
data['for_obj_id'] = for_obj_id
data['for_obj_id_random'] = for_obj_id_random
data['group'] = group
if group:
sql = """
SELECT *
FROM `event_badge` AS event_badge
WHERE event_badge.for_type = :for_obj_type AND event_badge.for_id = :for_obj_id
AND event_badge.group = :group
"""
else:
sql = """
SELECT *
FROM `event_badge` AS event_badge
WHERE event_badge.for_type = :for_obj_type AND event_badge.for_id = :for_obj_id
"""
if sql_result := sql_select(data=data, sql=sql, as_list=True):
resp_data_li = []
for record in sql_result:
resp_data = base_name(**record).dict(by_alias=by_alias, exclude_unset=exclude_unset)
resp_data_li.append(resp_data)
return mk_resp(data=resp_data_li)
else:
log.debug(sql_result)
return mk_resp(data=False, status_code=404, response=response)
result = get_obj_li_template(
obj_type=obj_type,
for_obj_type=for_obj_type,
for_obj_id=for_obj_id,
by_alias=True,
exclude_unset=True,
)
return result
# ### BEGIN ### API Event Badge ### search_event_badge_obj_li() ###
# Updated 2021-07-28
@router.get('/event/{event_id}/badge/search', response_model=Resp_Body_Base)
async def search_event_badge_obj_li(
# account_id: str = Query(None, min_length=11, max_length=22),
event_id: str = Query(None, min_length=11, max_length=22),
external_id: str = Query('%', max_length=50),
given_name: str = Query('%', max_length=25),
family_name: str = Query('%', max_length=25),
email: str = Query('%', max_length=50),
limit: int = 50,
enabled: str = 'enabled',
x_account_id: str = Header(...),
by_alias: bool = True,
exclude_unset: bool = True,
response: Response = Response,
):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
data = {}
data['account_id_random'] = x_account_id
data['event_id_random'] = event_id
data['external_id'] = '%'+external_id+'%'
data['given_name'] = '%'+given_name+'%'
data['family_name'] = '%'+family_name+'%'
data['email'] = '%'+email+'%' # Adding the % symbol here because it turns certain combinations into special characters
# data['from_datetime'] = from_datetime
# data['to_datetime'] = to_datetime
log.debug(data)
if enabled in ['enabled', 'disabled', 'all']:
if enabled == 'enabled':
data['enable'] = True
sql_enabled = f'AND `event_badge`.enable = :enable'
elif enabled == 'disabled':
data['enable'] = False
sql_enabled = f'AND `event_badge`.enable = :enable'
elif enabled == 'all':
sql_enabled = ''
if limit:
data['limit'] = limit
sql_limit = f'LIMIT :limit'
else:
sql_limit = ''
sql = f"""
SELECT *
FROM `v_event_badge` AS event_badge
WHERE event_badge.account_id = :account_id
AND (event_badge.external_id LIKE :external_id OR event_badge.external_id IS NULL)
AND (event_badge.given_name LIKE :given_name
AND event_badge.family_name LIKE :family_name)
AND event_badge.email LIKE :email
{sql_enabled}
ORDER BY event_badge.given_name ASC, event_badge.family_name ASC, event_badge.created_on DESC, event_badge.updated_on DESC
{sql_limit};
"""
sql = f"""
SELECT *
FROM `v_event_badge` AS event_badge
WHERE event_badge.event_id = :event_id
AND (event_badge.given_name LIKE :given_name
AND event_badge.family_name LIKE :family_name)
AND event_badge.email LIKE :email
{sql_enabled}
ORDER BY event_badge.given_name ASC, event_badge.family_name ASC, event_badge.created_on DESC, event_badge.updated_on DESC
{sql_limit};
"""
log.debug(sql)
if sql_result := sql_select(data=data, sql=sql, as_list=True, rm_id_random=True):
resp_data_li = []
for record in sql_result:
resp_data = Event_Badge_Base(**record).dict(by_alias=by_alias, exclude_unset=exclude_unset)
resp_data_li.append(resp_data)
return mk_resp(data=resp_data_li)
else:
log.debug(sql_result)
return mk_resp(data=False, status_code=404, response=response)
# ### END ### API Event Badge ### search_event_badge_obj_li() ###
# ### BEGIN ### API Event Badge ### get_event_badge_obj() ###
# Updated 2021-08-05
@router.get('/event/badge/{event_badge_id}', response_model=Resp_Body_Base)
async def get_event_badge_obj(
event_badge_id: str = Query(..., min_length=11, max_length=22),
limit: int = 5, # For now this covers any included objects or object lists
enabled: str = 'enabled', # For now this covers any included objects or object lists
x_account_id: str = Header(...),
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
response: Response = Response,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if event_badge_id := redis_lookup_id_random(record_id_random=event_badge_id, table_name='event_badge'): pass
else: return mk_resp(data=None, status_code=404)
event_badge_obj = load_event_badge_obj(
event_badge_id = event_badge_id,
limit = limit,
by_alias = by_alias,
exclude_unset = exclude_unset,
# model_as_dict = model_as_dict,
enabled = enabled,
)
event_badge_dict = event_badge_obj.dict(by_alias=by_alias, exclude_unset=exclude_unset)
return mk_resp(data=event_badge_dict)
# ### END ### API Event Badge ### get_event_badge_obj() ###
# ### BEGIN ### API Event Badge ### get_account_obj_event_badge_list() ###
# Updated 2021-07-28
@router.get('/account/{account_id}/event/badge/list', response_model=Resp_Body_Base)
async def get_account_obj_event_badge_list(
account_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_event_badge: bool = False,
x_account_id: str = Header(...),
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
response: Response = Response,
):
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 mk_resp(data=None, status_code=404)
response_data = None
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
# Updated 2021-07-28
if event_badge_rec_list_result := get_event_badge_rec_list(
account_id = account_id,
limit = limit,
enabled = enabled,
):
event_badge_result_list = []
for event_badge_rec in event_badge_rec_list_result:
if load_event_badge_result := load_event_badge_obj(
event_badge_id = event_badge_rec.get('event_badge_id', None),
limit = limit,
by_alias = by_alias,
exclude_unset = exclude_unset,
# model_as_dict = model_as_dict,
enabled = enabled,
inc_event_badge = inc_event_badge,
):
event_badge_result_list.append(load_event_badge_result)
else:
event_badge_result_list.append(None)
response_data = event_badge_result_list
else:
return mk_resp(data=False, status_code=400, response=response) # Bad Request
return mk_resp(data=response_data)
# ### END ### API Event Badge ### get_account_obj_event_badge_list() ###
# ### BEGIN ### API Event Badge ### get_event_obj_event_badge_list() ###
# Updated 2021-07-28
@router.get('/event/{event_id}/event/badge/list', response_model=Resp_Body_Base)
async def get_event_obj_event_badge_list(
event_id: str = Query(..., min_length=11, 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
x_account_id: str = Header(...),
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
response: Response = Response,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if event_badge_id := redis_lookup_id_random(record_id_random=event_badge_id, table_name='event_badge'): pass
else: return mk_resp(data=None, status_code=404)
response_data = None
# log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
# Updated 2021-07-28
if event_badge_rec_list_result := get_event_badge_rec_list(
event_badge_id = event_badge_id,
limit = limit,
enabled = enabled,
):
event_badge_result_list = []
for event_badge_rec in event_badge_rec_list_result:
if load_event_badge_result := load_event_badge_obj(
event_badge_id = event_badge_rec.get('event_badge_id', None),
limit = limit,
by_alias = by_alias,
exclude_unset = exclude_unset,
# model_as_dict = model_as_dict,
enabled = enabled,
):
event_badge_result_list.append(load_event_badge_result)
else:
event_badge_result_list.append(None)
response_data = event_badge_result_list
else:
return mk_resp(data=False, status_code=400, response=response) # Bad Request
return mk_resp(data=response_data)
# ### END ### API Event Badge ### get_event_obj_event_badge_list() ###
@router.delete('/event/badge/{obj_id}', response_model=Resp_Body_Base)
async def delete_event_badge_obj(
obj_id: str = Query(..., min_length=1, max_length=22),
x_account_id: str = Header(...),
response: Response = Response,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
obj_type = 'event_badge'
result = delete_obj_template(
obj_type=obj_type,
obj_id=obj_id,
)
return result

112
app/routers/event_badge_template.py Normal file
View File

@@ -0,0 +1,112 @@
import datetime
from fastapi import APIRouter, Body, Depends, Header, HTTPException, Query, Response, status
from pydantic import BaseModel, EmailStr, Field
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.routers.api_crud import delete_obj_template, get_obj_template, get_obj_li_template, patch_obj_template, post_obj_template
from app.methods.event_badge_template_methods import load_event_badge_template_obj
from app.models.event_badge_template_models import Event_Badge_Template_Base
from app.models.response_models import Resp_Body_Base, mk_resp
router = APIRouter()
# ### BEGIN ### API Event Badge ### get_event_event_badge_template_obj() ###
# NOTE: This should probably return a list in the future :NOTE
# Updated 2021-09-08
@router.get('/event/{event_id}/badge/template', response_model=Resp_Body_Base)
async def get_event_event_badge_template_obj(
event_id: str = Query(None, min_length=11, max_length=22),
limit: int = 50,
enabled: str = 'enabled',
x_account_id: str = Header(...),
by_alias: bool = True,
exclude_unset: bool = True,
response: Response = Response,
):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if event_id := redis_lookup_id_random(record_id_random=event_id, table_name='event'): pass
else: return mk_resp(data=None, status_code=404)
data = {}
data['event_id'] = event_id
log.debug(data)
if enabled in ['enabled', 'disabled', 'all']:
if enabled == 'enabled':
data['enable'] = True
sql_enabled = f'AND `event_badge`.enable = :enable'
elif enabled == 'disabled':
data['enable'] = False
sql_enabled = f'AND `event_badge`.enable = :enable'
elif enabled == 'all':
sql_enabled = ''
if limit:
data['limit'] = limit
sql_limit = f'LIMIT :limit'
else:
sql_limit = ''
sql = f"""
SELECT *
FROM `event_badge_template` AS event_badge_template
WHERE event_badge_template.event_id = :event_id
{sql_limit};
"""
log.debug(sql)
if sql_result := sql_select(data=data, sql=sql): # NOTE: This will probably need to be a list in the future. Force return of list.
resp_data = Event_Badge_Template_Base(**sql_result).dict(by_alias=by_alias, exclude_unset=exclude_unset)
return mk_resp(data=resp_data)
# resp_data_li = []
# for record in sql_result:
# resp_data = Event_Badge_Template_Base(**record).dict(by_alias=by_alias, exclude_unset=exclude_unset)
# resp_data_li.append(resp_data)
# return mk_resp(data=resp_data_li)
else:
log.debug(sql_result)
return mk_resp(data=False, status_code=404, response=response)
# ### END ### API Event Badge ### get_event_event_badge_template_obj() ###
# ### BEGIN ### API Event Badge ### get_event_badge_template_obj() ###
# Updated 2021-09-08
@router.get('/event/badge/template/{event_badge_template_id}', response_model=Resp_Body_Base)
async def get_event_badge_template_obj(
event_badge_template_id: str = Query(..., min_length=11, max_length=22),
limit: int = 5, # For now this covers any included objects or object lists
enabled: str = 'enabled', # For now this covers any included objects or object lists
x_account_id: str = Header(...),
by_alias: Optional[bool] = True,
exclude_unset: Optional[bool] = True,
response: Response = Response,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if event_badge_template_id := redis_lookup_id_random(record_id_random=event_badge_template_id, table_name='event_badge_template'): pass
else: return mk_resp(data=None, status_code=404)
event_badge_template_obj = load_event_badge_template_obj(
event_badge_template_id = event_badge_template_id,
limit = limit,
by_alias = by_alias,
exclude_unset = exclude_unset,
# model_as_dict = model_as_dict,
enabled = enabled,
)
event_badge_template_dict = event_badge_template_obj.dict(by_alias=by_alias, exclude_unset=exclude_unset)
return mk_resp(data=event_badge_template_dict)
# ### END ### API Event Badge ### get_event_badge_template_obj() ###

2
app/routers/importing.py
View File

@@ -11,7 +11,7 @@ from app.db_sql import sql_insert, sql_update, sql_insert_or_update, sql_select,
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.person_methods import create_update_person_obj, get_person_rec_list, load_person_obj, update_person_obj
from app.methods.person_methods import create_update_person_obj_v4b, get_person_rec_list, load_person_obj
from app.methods.user_methods import load_user_obj
from app.models.contact_models import Contact_Base

2
app/routers/lookup.py
View File

@@ -28,7 +28,7 @@ async def get_lookup_li(
enabled: str = 'enabled', # enabled, disabled, all
response: Response = Response,
):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
allowed_lookup_tables = ['country', 'country_subdivision', 'event_session_type', 'file_purpose', 'membership_group_status', 'membership_person_status', 'membership_type_status', 'order_status', 'post_topic', 'product_type', 'time_zone', 'user_status']

145
app/routers/person.py
View File

@@ -11,7 +11,7 @@ from app.routers.api_crud import delete_obj_template, get_obj_template, get_obj_
# from app.methods.membership_person_methods import load_membership_person_obj
from app.methods.order_methods import get_order_rec_list, load_order_obj
from app.methods.person_methods import create_person_obj_v3, create_update_person_obj, get_person_rec_list, get_person_rec_w_external_id, load_person_obj, update_person_obj, update_person_obj_v3
from app.methods.person_methods import create_update_person_obj_v4b, get_person_rec_list, get_person_rec_w_external_id, load_person_obj, update_person_obj
from app.models.person_models import Person_Base
from app.models.response_models import Resp_Body_Base, mk_resp
@@ -73,121 +73,148 @@ async def patch_person_obj(
# ### BEGIN ### API Person ### v3_post_person_obj_new() ###
# Updated 2021-08-24
# Using create_update_person_obj_v4b() now
# Updated 2021-09-08
@router.post('/v3/person/new', response_model=Resp_Body_Base)
async def v3_post_person_obj_new(
person_obj: Person_Base,
process_contact: bool = False,
process_organization: bool = False,
process_user: bool = False,
create_sub_obj: bool = False,
fail_any: bool = True, # Fail if any thing goes wrong for sub objects
enabled: str = 'enabled', # For now this covers any included objects or object lists
inc_address: bool = False, # Priority l1
inc_contact: bool = False, # Priority l1
inc_user: bool = False, # Priority l1
x_account_id: str = Header(...),
inc_address: bool = False,
inc_contact: bool = False,
inc_organization: bool = False,
inc_user: bool = False,
return_obj: bool = True,
limit: int = 50,
by_alias: bool = True,
exclude_unset: bool = True,
# include: Optional[list] = [],
# exclude: Optional[list] = [],
exclude_unset: Optional[bool] = True,
# exclude_none: Optional[bool] = True,
x_account_id: str = Header(...),
response: Response = Response,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
log.debug(person_obj.account_id)
log.debug(person_obj.account_id_random)
# if person_obj.account_id: pass
# else: return mk_resp(data=False, status_code=500, response=response, status_message='Testing')
# abort(501)
# There should probably be a check for the account ID before calling the create function?
if create_person_obj_result := create_person_obj_v3(
if create_update_person_obj_result := create_update_person_obj_v4b(
account_id = x_account_id,
person_obj_new = person_obj,
create_sub_obj = create_sub_obj,
fail_any = fail_any,
return_outline = False,
person_dict_obj = person_obj,
person_id = None,
# process_contact = process_contact,
# process_organization = process_organization,
# process_user = process_user,
): pass
else: return mk_resp(data=False, status_code=400, response=response, status_message='The person was not created. Check the field names and data types.')
if isinstance(create_person_obj_result, int):
person_id = create_person_obj_result
if isinstance(create_update_person_obj_result, int):
person_id = create_update_person_obj_result
if return_obj:
if load_person_obj_result := load_person_obj(
person_id = person_id,
enabled = enabled,
inc_address = inc_address,
inc_contact = inc_contact,
inc_organization = inc_organization,
inc_user = inc_user,
):
data = load_person_obj_result
else:
data = False
return mk_resp(data=data, response=response, status_message='The person was probably created, but there was a problem returning the data.')
else:
person_id = create_person_obj_result
person_id = create_update_person_obj_result
person_id_random = get_id_random(record_id=person_id, table_name='person')
data = {}
data['person_id'] = person_id
data['person_id_random'] = person_id_random
return mk_resp(data=data, response=response, status_message='The person was created.')
else:
return mk_resp(data=False, status_code=400, response=response, status_message='The result from trying to create an person was unexpected.')
return mk_resp(data=False, status_code=400, response=response, status_message='The result from trying to create a person was unexpected.')
# ### BEGIN ### API Person ### v3_post_person_obj_new() ###
# ### BEGIN ### API Person ### v3_patch_person_obj_exist ###
# Updated 2021-08-24
# Using create_update_person_obj_v4b() now
# Updated 2021-09-08
@router.patch('/v3/person/{person_id}/exist', response_model=Resp_Body_Base)
async def v3_patch_person_obj_exist(
person_obj: Person_Base,
person_id: str = Query(..., min_length=11, max_length=22),
process_contact: bool = False,
process_organization: bool = False,
process_user: bool = False,
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,
by_alias: Optional[bool] = True,
enabled: str = 'enabled', # For now this covers any included objects or object lists
inc_address: bool = False,
inc_contact: bool = False,
inc_organization: bool = False,
inc_user: bool = False,
return_obj: bool = True,
limit: int = 50,
by_alias: bool = True,
# include: Optional[list] = [],
# exclude: Optional[list] = [],
exclude_unset: Optional[bool] = True,
exclude_none: Optional[bool] = True,
# exclude_none: Optional[bool] = True,
x_account_id: str = Header(...),
response: Response = Response,
):
log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
if person_id := redis_lookup_id_random(record_id_random=person_id, table_name='person'): pass
else: return mk_resp(data=None, status_code=404)
else: return mk_resp(data=None, status_code=404, response=response, status_message='The person was not updated. The person ID was invalid or not found.')
if update_person_obj_result := update_person_obj_v3(
if create_update_person_obj_result := create_update_person_obj_v4b(
account_id = x_account_id,
person_dict_obj = person_obj,
person_id = person_id,
person_obj_exist = person_obj,
create_sub_obj = create_sub_obj,
fail_any = fail_any,
return_outline = False,
# process_contact = process_contact,
# process_organization = process_organization,
# process_user = process_user,
): pass
else: return mk_resp(data=False, status_code=400, response=response, status_message='The person was not created. Check the field names and data types.')
else: return mk_resp(data=False, status_code=400, response=response, status_message='The person was not updated. Check the field names and data types.')
if update_person_obj_result:
if isinstance(create_update_person_obj_result, int):
person_id = create_update_person_obj_result
if return_obj:
if load_person_obj_result := load_person_obj(person_id=person_id):
if load_person_obj_result := load_person_obj(
person_id = person_id,
enabled = enabled,
inc_address = inc_address,
inc_contact = inc_contact,
inc_organization = inc_organization,
inc_user = inc_user,
):
data = load_person_obj_result
else:
data = False
return mk_resp(data=data, response=response, status_message='The person was probably updated, but there was a problem returning the data.')
else:
person_id = create_update_person_obj_result
person_id_random = get_id_random(record_id=person_id, table_name='person')
data = {}
data['person_id'] = person_id
data['person_id_random'] = person_id_random
return mk_resp(data=data, response=response, status_message='The person was created.')
return mk_resp(data=data, response=response, status_message='The person was updated.')
else:
return mk_resp(data=False, status_code=400, response=response, status_message='The result from trying to create an person was unexpected.')
return mk_resp(data=False, status_code=400, response=response, status_message='The result from trying to update the person was unexpected.')
# ### END ### API Person ### v3_patch_person_obj_exist ###
# ### BEGIN ### API Person ### post_person_json() ###
# Updated 2021-06-25
# Updated 2021-09-08
@router.post('/person/json', response_model=Resp_Body_Base)
async def post_person_json(
person_obj: Person_Base,
@@ -195,26 +222,26 @@ async def post_person_json(
# create_sub_obj: bool = False,
process_contact: bool = False,
process_organization: bool = False,
x_account_id: Optional[str] = Header(..., ),
process_user: bool = False,
return_obj: Optional[bool] = True,
by_alias: Optional[bool] = True,
include: Optional[list] = [],
exclude: Optional[list] = [],
# include: Optional[list] = [],
# exclude: Optional[list] = [],
exclude_unset: Optional[bool] = True,
exclude_none: Optional[bool] = True,
# exclude_none: Optional[bool] = True,
x_account_id: Optional[str] = Header(..., ),
response: Response = Response,
):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
# if person_id := redis_lookup_id_random(record_id_random=person_id, table_name='person'): pass
# else: return mk_resp(data=None, status_code=404)
if person_obj_in_result := create_update_person_obj(
if person_obj_in_result := create_update_person_obj_v4b(
account_id = x_account_id,
person_dict_obj = person_obj,
person_id = None,
person_obj = person_obj,
process_contact = process_contact,
process_organization = process_organization,
process_user = process_user,
):
log.debug(person_obj_in_result)
if return_obj:
@@ -233,9 +260,10 @@ async def post_person_json(
@router.patch('/person/{person_id}/json', response_model=Resp_Body_Base)
async def patch_person_json(
person_obj: Person_Base,
person_id: str = Query(..., min_length=1, max_length=22),
person_id: str = Query(..., min_length=11, max_length=22),
process_contact: bool = False,
process_organization: bool = False,
process_user: bool = False,
x_account_id: Optional[str] = Header(..., ),
return_obj: Optional[bool] = True,
by_alias: Optional[bool] = True,
@@ -251,11 +279,12 @@ async def patch_person_json(
if person_id := redis_lookup_id_random(record_id_random=person_id, table_name='person'): pass
else: return mk_resp(data=None, status_code=404, response=response)
if person_obj_up_result := create_update_person_obj(
person_id = person_id,
person_obj = person_obj,
if person_obj_up_result := create_update_person_obj_v4b(
account_id = x_account_id,
person_dict_obj = person_obj,
process_contact = process_contact,
process_organization = process_organization,
process_user = process_user,
):
log.debug(person_obj_up_result)
if return_obj: