2659047d24
All create_update_*_v4 functions for event_badge, event_person, event_person_profile, event_presenter, and event_presentation were calling sql_update without record_id. Vision ID models use Optional[str] IDs with a root_validator that strips integer values, so the serialized dict contained no id key and sql_update could not identify the row. Fix: pass record_id=<integer_id> explicitly to sql_update in all five functions. Also fix walrus-operator false-negative: None return from sql_update (0 rows affected — record unchanged) is not an error and should not abort sub-object cascade; use explicit `is False` check. Also broadens Axonius badge_type_code mapping to substring match so future ticket name variants still resolve correctly. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
225 lines
11 KiB
Python
225 lines
11 KiB
Python
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 get_account_id_w_for_type_id, redis_lookup_id_random, sql_insert, sql_select, sql_update
|
|
from app.lib_general import log, logging
|
|
|
|
from app.methods.address_methods import create_address_obj, create_update_address_obj, create_update_address_obj_v4, update_address_obj
|
|
|
|
from app.models.common_field_schema import default_num_bytes
|
|
from app.models.event_person_profile_models import Event_Person_Profile_Base
|
|
|
|
|
|
# ### BEGIN ### API Event Person Profile Methods ### load_event_person_profile_obj() ###
|
|
def load_event_person_profile_obj(
|
|
event_person_profile_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
|
|
inc_address: bool = False,
|
|
inc_contact: bool = False,
|
|
) -> Event_Person_Profile_Base|dict|bool:
|
|
# log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
|
log.debug(locals())
|
|
|
|
if event_person_profile_id := redis_lookup_id_random(record_id_random=event_person_profile_id, table_name='event_person_profile'): pass
|
|
else: return False
|
|
|
|
if event_person_profile_rec := sql_select(table_name='v_event_person_profile', record_id=event_person_profile_id): pass
|
|
else: return False
|
|
|
|
try:
|
|
event_person_profile_obj = Event_Person_Profile_Base(**event_person_profile_rec)
|
|
log.debug(event_person_profile_obj)
|
|
except ValidationError as e:
|
|
log.error(e.json())
|
|
return False
|
|
|
|
if inc_contact:
|
|
log.info('Need to include contact data...')
|
|
contact_id = event_person_profile_rec.get('contact_id', None)
|
|
log.debug(contact_id)
|
|
from app.methods.contact_methods import load_contact_obj
|
|
if contact_result := load_contact_obj(
|
|
contact_id = contact_id,
|
|
by_alias = by_alias,
|
|
exclude_unset = exclude_unset,
|
|
model_as_dict = model_as_dict,
|
|
):
|
|
event_person_profile_obj.contact = contact_result
|
|
else: event_person_profile_obj.contact = None
|
|
|
|
if model_as_dict:
|
|
return event_person_profile_obj.dict(by_alias=by_alias, exclude_unset=exclude_unset) # pylint: disable=no-member
|
|
else:
|
|
return event_person_profile_obj
|
|
# ### END ### API Event Person Profile Methods ### load_event_person_profile_obj() ###
|
|
|
|
|
|
# ### BEGIN ### API Event Person Profile Methods ### get_event_person_id_w_event_person_profile_id() ###
|
|
# Updated 2021-09-07
|
|
def get_event_person_id_w_event_person_profile_id(
|
|
event_person_profile_id: int|str,
|
|
) -> bool|int|None:
|
|
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
|
log.debug(locals())
|
|
|
|
if event_person_profile_id := redis_lookup_id_random(record_id_random=event_person_profile_id, table_name='event_person_profile'): pass
|
|
else: return False
|
|
|
|
data = {}
|
|
data['event_person_profile_id'] = event_person_profile_id
|
|
|
|
sql = f"""
|
|
SELECT `event_person_profile`.id AS 'event_person_profile_id', `event_person_profile`.id_random AS 'event_person_profile_id_random', `event_person_profile`.account_id AS account_id
|
|
FROM `event_person_profile` AS `event_person_profile`
|
|
WHERE `event_person_profile`.id = :event_person_profile_id
|
|
LIMIT 1;
|
|
"""
|
|
|
|
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
|
if event_person_profile_data_result := sql_select(data=data, sql=sql):
|
|
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
|
log.debug(event_person_profile_data_result)
|
|
if account_id := event_person_profile_data_result.get('account_id', None): return account_id
|
|
else: return False
|
|
else: return None
|
|
# ### END ### API Event Person Profile Methods ### get_event_person_id_w_event_person_profile_id() ###
|
|
|
|
|
|
|
|
# ### BEGIN ### API Event Person Profile Methods ### create_update_event_person_profile_obj_v4() ###
|
|
# Updated 2022-02-23
|
|
def create_update_event_person_profile_obj_v4(
|
|
event_person_profile_dict_obj: Event_Person_Profile_Base|dict,
|
|
event_person_profile_id: int|str = None,
|
|
account_id: int = None,
|
|
# event_id: int|str = None,
|
|
event_person_id: int|str = None,
|
|
create_sub_obj: bool = False,
|
|
fail_any: bool = False, # Fail if any thing goes wrong for sub objects
|
|
return_outline: bool = False,
|
|
) -> int|bool:
|
|
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
|
|
log.debug(locals())
|
|
|
|
# ### SECTION ### Secondary data validation
|
|
log.info('Checking requirements...')
|
|
if event_person_profile_id:
|
|
log.info(f'Event Person Profile ID passed. Update existing Event Person Profile. Event Person Profile ID: {event_person_profile_id}')
|
|
|
|
if event_person_profile_id := redis_lookup_id_random(record_id_random=event_person_profile_id, table_name='event_person_profile'): pass
|
|
else:
|
|
log.error('Event Person Profile ID passed but is invalid. Failed requirement.')
|
|
return False
|
|
else:
|
|
log.info('No Event Person Profile ID passed. Create new Event Person Profile. Required: Event Person ID')
|
|
|
|
if event_person_id := redis_lookup_id_random(record_id_random=event_person_id, table_name='event_person'): pass
|
|
else:
|
|
log.error('Missing or invalid Event Person ID passed. Failed requirement.')
|
|
log.info(f'Event Person ID: {event_person_id}')
|
|
return False
|
|
|
|
log.info('Create dictionary or Pydantic object')
|
|
log.debug(type(event_person_profile_dict_obj))
|
|
if isinstance(event_person_profile_dict_obj, dict):
|
|
event_person_profile_dict = event_person_profile_dict_obj
|
|
if event_person_profile_id:
|
|
event_person_profile_dict['event_person_profile_id'] = event_person_profile_id
|
|
try:
|
|
event_person_profile_obj = Event_Person_Profile_Base(**event_person_profile_dict)
|
|
except ValidationError as e:
|
|
log.error(e.json())
|
|
return False
|
|
else:
|
|
event_person_profile_obj = event_person_profile_dict_obj
|
|
if event_person_profile_id:
|
|
# NOTE: Can't update the ID alias if it was never set.
|
|
event_person_profile_obj.id = event_person_profile_id
|
|
log.debug(event_person_profile_obj)
|
|
|
|
event_person_profile_dict = event_person_profile_obj.dict(by_alias=False, exclude_defaults=False, exclude_unset=True, exclude={'account_id', 'account_id_random', 'contact', 'event_cfg', 'event_id', 'event_id_random', 'event_person_id', 'event_person_id_random', 'organization', 'created_on', 'updated_on', 'external_id_old'})
|
|
|
|
# ### SECTION ### Process data
|
|
contact_id = None
|
|
if event_person_profile_obj.contact and event_person_profile_obj.contact.id:
|
|
contact_id = event_person_profile_obj.contact.id
|
|
elif event_person_profile_obj.contact and event_person_profile_obj.contact_id:
|
|
contact_id = event_person_profile_obj.contact_id
|
|
|
|
if event_person_profile_id:
|
|
event_person_profile_dict_up_result = sql_update(data=event_person_profile_dict, table_name='event_person_profile', record_id=event_person_profile_id, rm_id_random=True)
|
|
if event_person_profile_dict_up_result is False:
|
|
log.warning(f'Event Person Profile update failed (DB error). Event Person Profile ID: {event_person_profile_id}')
|
|
log.debug(event_person_profile_dict_up_result)
|
|
return False
|
|
# None means 0 rows affected (record unchanged) — not an error
|
|
log.debug(event_person_profile_dict_up_result)
|
|
else:
|
|
if event_person_profile_dict_in_result := sql_insert(data=event_person_profile_dict, table_name='event_person_profile', rm_id_random=True, id_random_length=8): pass
|
|
else:
|
|
log.warning(f'Event Person Profile not created.')
|
|
log.debug(event_person_profile_dict_in_result)
|
|
return False
|
|
log.debug(event_person_profile_dict_in_result)
|
|
|
|
event_person_profile_id = event_person_profile_dict_in_result
|
|
|
|
# NOTE: This is really only needed if a new contact is being created
|
|
|
|
log.info('Attempting to get Account ID from related object.')
|
|
if account_id: pass
|
|
elif account_id := event_person_profile_obj.account_id: pass
|
|
# elif account_id := event_person_profile_obj.account_id_random: pass
|
|
else:
|
|
if event_person_id: pass
|
|
elif event_person_id := event_person_profile_obj.event_person_id: pass
|
|
# elif event_person_id := event_person_profile_obj.event_person_id_random: pass
|
|
|
|
if event_person_id:
|
|
if account_id := get_account_id_w_for_type_id(for_type='event_person', for_id=event_person_id): pass
|
|
else:
|
|
log.error('Unable to get Account ID from related object.')
|
|
False
|
|
|
|
event_person_profile_outline = {}
|
|
event_person_profile_outline['event_person_profile_id'] = event_person_profile_id
|
|
|
|
# NOTE: Use object model version because of better type checking and validations
|
|
if event_person_profile_obj.contact:
|
|
event_person_profile_outline['contact_id'] = None
|
|
contact_obj = event_person_profile_obj.contact
|
|
contact_obj.for_type = 'event_person_profile'
|
|
contact_obj.for_id = event_person_profile_id
|
|
create_update_contact_obj_result = create_update_contact_obj_v4(
|
|
contact_dict_obj = contact_obj,
|
|
contact_id = contact_id,
|
|
account_id = account_id,
|
|
for_type = 'event_person_profile',
|
|
for_id = event_person_profile_id,
|
|
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
|
|
|
|
event_person_profile_outline['contact_id'] = contact_id
|
|
|
|
if return_outline:
|
|
log.debug(f'Returning the Event Person Profile Outline: {event_person_profile_outline}')
|
|
return event_person_profile_outline
|
|
else:
|
|
log.debug(f'Returning the Event Person Profile ID: {event_person_profile_id}')
|
|
return event_person_profile_id
|
|
# ### END ### API Event Person Profile Methods ### create_update_event_person_profile_obj_v4() ###
|