Files
OSIT-AE-API-FastAPI/app/models/event_models.py
Scott Idem 0f4b4d2f51 feat: Implement V3 ID Vision and fields_to_exclude_from_db across core models
This commit refactors numerous Pydantic models to align with the V3 ID Vision standard, ensuring that primary and foreign key fields are represented as clean string IDs in the API. It also introduces and populates the  ClassVar in each model to prevent view-only fields and linked objects from being inadvertently written to the database during PATCH/POST operations.

Specifically, this includes:
- Adding  to exclude view-derived or joined fields such as , , nested objects (e.g., , ), and convenience fields (e.g., ).
- Adjusting root validators to correctly map string IDs and strip internal integer IDs for API responses.
- Resolving a KeyError by adding  to .

These changes are crucial for maintaining data integrity and consistency with the V3 API architecture.
2026-02-24 16:21:27 -05:00

483 lines
21 KiB
Python

import datetime, pytz
from typing import Dict, List, Optional, Set, Union, ClassVar
from pydantic import BaseModel, EmailStr, Field, Json, PrivateAttr, ValidationError, validator, root_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
from app.models.address_models import Address_Base
from app.models.contact_models import Contact_Base
from app.models.event_cfg_models import Event_Cfg_Base
from app.models.event_location_models import Event_Location_Base
from app.models.event_person_models import Event_Person_Base
from app.models.event_session_models import Event_Session_Base
from app.models.person_models import Person_Base
from app.models.user_models import User_Base
# ### BEGIN ### API Event Models ### Event_Base() ###
class Event_Base(BaseModel):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
# --- Standardized Vision IDs (Strings for API, Integers for DB) ---
# We use Union[int, str] to allow both public string IDs and resolved DB integers to pass validation.
id: Optional[Union[int, str]] = Field(None, **base_fields['event_id_random'])
event_id: Optional[Union[int, str]] = Field(None, **base_fields['event_id_random'])
account_id: Optional[Union[int, str]] = Field(None, **base_fields['account_id_random'])
poc_event_person_id: Optional[Union[int, str]] = Field(None, **base_fields['event_person_id_random'])
poc_person_id: Optional[Union[int, str]] = Field(None, **base_fields['person_id_random'])
user_id: Optional[Union[int, str]] = Field(None, **base_fields['user_id_random'])
address_location_id: Optional[Union[int, str]] = Field(None, **base_fields['address_id_random'])
contact_1_id: Optional[Union[int, str]] = Field(None, **base_fields['contact_id_random'])
contact_2_id: Optional[Union[int, str]] = Field(None, **base_fields['contact_id_random'])
contact_3_id: Optional[Union[int, str]] = Field(None, **base_fields['contact_id_random'])
# --- Standardized Legacy / Internal IDs (Excluded) ---
id_random: Optional[str] = Field(None, alias='event_id_random', exclude=True)
account_id_random: Optional[str] = Field(None, exclude=True)
poc_event_person_id_random: Optional[str] = Field(None, exclude=True)
poc_person_id_random: Optional[str] = Field(None, exclude=True)
user_id_random: Optional[str] = Field(None, exclude=True)
address_location_id_random: Optional[str] = Field(None, exclude=True)
contact_1_id_random: Optional[str] = Field(None, exclude=True)
contact_2_id_random: Optional[str] = Field(None, exclude=True)
contact_3_id_random: Optional[str] = Field(None, exclude=True)
@root_validator(pre=True)
def map_v3_ids(cls, values):
"""
Vision Transformer:
Map DB keys to clean API keys and strip internal integers during READ operations.
"""
# 1. Map Random Strings to Clean Names
rid = values.get('id_random') or values.get('event_id_random')
if rid and isinstance(rid, str):
values['id'] = rid
values['event_id'] = rid
if a_rid := values.get('account_id_random'): values['account_id'] = a_rid
if pep_rid := values.get('poc_event_person_id_random'): values['poc_event_person_id'] = pep_rid
if pp_rid := values.get('poc_person_id_random'): values['poc_person_id'] = pp_rid
if u_rid := values.get('user_id_random'): values['user_id'] = u_rid
if al_rid := values.get('address_location_id_random'): values['address_location_id'] = al_rid
if c1_rid := values.get('contact_1_id_random'): values['contact_1_id'] = c1_rid
if c2_rid := values.get('contact_2_id_random'): values['contact_2_id'] = c2_rid
if c3_rid := values.get('contact_3_id_random'): values['contact_3_id'] = c3_rid
# 2. Prevent "Collision Population" or leakage of integers during API responses
# WE MUST NOT DELETE these if they are already integers during a POST operation
# as they have been resolved by sanitize_payload.
for k in ['id', 'event_id', 'account_id', 'poc_event_person_id', 'poc_person_id', 'user_id', 'address_location_id', 'contact_1_id', 'contact_2_id', 'contact_3_id']:
val = values.get(k)
if val is not None and not isinstance(val, str):
if values.get(f'{k}_random') or (k=='id' and values.get('id_random')):
del values[k]
return values
code: Optional[str] = Field(
alias = 'event_code'
)
external_person_id: Optional[str] # Person ID generated by external system (should be stable and not change)
lu_event_type_id: Optional[int] = Field(None, exclude=True)
#lu_event_type: Optional[str] # Needs to be reviewed
conference: Optional[bool] # Also in Event_Cfg_Base model
# type_name: Optional[str] = Field(
# alias = 'type'
# )
type: Optional[str]
name: Optional[str]
summary: Optional[str]
description: Optional[str]
format: Optional[str]
lu_time_zone_id: Optional[int]
timezone: Optional[str]
start_datetime: Optional[datetime.datetime] = None
end_datetime: Optional[datetime.datetime] = None
recurring: Optional[bool]
recurring_pattern: Optional[str]
recurring_start_time: Optional[str]
recurring_end_time: Optional[str]
recurring_text: Optional[str]
weekday_sunday: Optional[bool]
weekday_monday: Optional[bool]
weekday_tuesday: Optional[bool]
weekday_wednesday: Optional[bool]
weekday_thursday: Optional[bool]
weekday_friday: Optional[bool]
weekday_saturday: Optional[bool]
location_address_json: Optional[Union[Json, None]]
location_text: Optional[str]
online_start: Optional[datetime.datetime] = None
online_end: Optional[datetime.datetime] = None
reg_deadline_1: Optional[datetime.datetime] = None
reg_deadline_2: Optional[datetime.datetime] = None
reg_deadline_3: Optional[datetime.datetime] = None
reg_deadline_4: Optional[datetime.datetime] = None
max_registrants: Optional[int]
private: Optional[bool] # invite only
physical: Optional[bool] # physical in person event
virtual: Optional[bool] # virtual remote access event
contact_li_json: Optional[Union[Json, None]] # list of dicts (custom for client); this is SQL FULLTEXT() indexed
attend_url: Optional[str]
attend_url_code: Optional[str] # ID, code, nickname
attend_url_passcode: Optional[str]
attend_phone: Optional[str]
attend_phone_passcode: Optional[str]
attend_text: Optional[str]
attend_json: Optional[Union[Json, None]]
# access_key: Optional[str] # Maybe use in the future?
passcode: Optional[str]
file_count: Optional[int]
internal_use_count: Optional[int] # Should be renamed to "internal_use_file_count"???
event_file_id_li_json: Optional[Union[Json, None]] # List of file IDs (actually id_random)
file_count_all: Optional[int] # Of all files under a session
status: Optional[str]
review: Optional[bool]
approve: Optional[bool]
ready: Optional[bool]
ready_on: Optional[datetime.datetime]
archive: Optional[bool] # Also in Event_Cfg_Base model
archive_on: Optional[datetime.datetime] # Also in Event_Cfg_Base model
mod_abstracts_json: Optional[Union[Json, None]]
mod_badges_json: Optional[Union[Json, None]]
mod_exhibits_json: Optional[Union[Json, None]]
mod_meetings_json: Optional[Union[Json, None]]
mod_pres_mgmt_json: Optional[Union[Json, None]]
cfg_json: Optional[Union[Json, None]] # Store per event config options; Not currently used 2024-06-11
data_json: Optional[Union[Json, None]] # For key value data. Careful with overwriting existing fields! Not currently used 2024-06-11
enable: Optional[bool] # Also in Event_Cfg_Base model
enable_from: Optional[datetime.datetime] = None
enable_to: Optional[datetime.datetime] = None
hide: Optional[bool] # Also in Event_Cfg_Base model
priority: Optional[bool]
sort: Optional[int]
group: Optional[str]
notes: Optional[str]
created_on: Optional[datetime.datetime] = None
updated_on: Optional[datetime.datetime] = None
# Including JSON data
event_session_proposal_questions: Optional[Json] # list of dicts (custom for client) for event_session proposals
# Including other related objects
address_location: Optional[Address_Base]
contact_1: Optional[Contact_Base]
contact_2: Optional[Contact_Base]
contact_3: Optional[Contact_Base]
event_abstract_list: Optional[list] # Optional[Event_Abstract_Base]
event_cfg: Optional[Event_Cfg_Base]
event_device_list: Optional[list] # Optional[Event_Device_Base]
event_exhibit_list: Optional[list]
event_file_list: Optional[list] # Optional[Event_File_Base]
event_location_list: Optional[list[Event_Location_Base]] # Optional[list]
event_person_list: Optional[list]
event_presentation_list: Optional[list] # Optional[Event_Presentation_Base]
event_presenter_list: Optional[list] # Optional[Event_Presenter_Base]
event_session_list: Optional[list[Event_Session_Base]] # Optional[list]
event_track_list: Optional[list] # Optional[Event_Track_Base]
# poc_event_person: Optional[Event_Person_Base]
poc_person: Optional[Person_Base]
# user: Optional[User_Base]
_processed_at: datetime.datetime = PrivateAttr(default_factory=datetime.datetime.now)
@validator('created_on', always=True)
def created_on_utc(cls, v, values, **kwargs):
if isinstance(v, datetime.datetime):
return v.astimezone(pytz.UTC).isoformat()
else: return v
@validator('updated_on', always=True)
def updated_on_utc(cls, v, values, **kwargs):
if isinstance(v, datetime.datetime):
return v.astimezone(pytz.UTC).isoformat()
else: return v
@validator('recurring_start_time', 'recurring_end_time', pre=True, always=True)
def time_to_str(cls, v):
if isinstance(v, (datetime.time, datetime.timedelta)):
return str(v)
return v
# Fields that are part of the model (for reading) but should not be saved to the DB table.
# These convenience fields and related objects are joined in the view.
fields_to_exclude_from_db: ClassVar[list] = [
'file_count', 'internal_use_count', 'event_file_id_li_json', 'file_count_all',
'address_location', 'contact_1', 'contact_2', 'contact_3',
'event_abstract_list', 'event_cfg', 'event_device_list', 'event_exhibit_list',
'event_file_list', 'event_location_list', 'event_person_list',
'event_presentation_list', 'event_presenter_list', 'event_session_list',
'event_track_list', 'poc_person'
]
class Config:
underscore_attrs_are_private = True
allow_population_by_field_name = True
fields = base_fields
# ### END ### API Event Models ### Event_Base() ###
# ### BEGIN ### API Event Models ### Event_Meeting_Flat_Base() ###
# Updated 2021-12-13
class Event_Meeting_Flat_Base(BaseModel):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
# --- Standardized Vision IDs (Strings for API, Integers for DB) ---
id: Optional[Union[int, str]] = Field(None, **base_fields['event_id_random'])
event_id: Optional[Union[int, str]] = Field(None, **base_fields['event_id_random'])
account_id: Optional[Union[int, str]] = Field(None, **base_fields['account_id_random'])
poc_event_person_id: Optional[Union[int, str]] = Field(None, **base_fields['event_person_id_random'])
poc_person_id: Optional[Union[int, str]] = Field(None, **base_fields['person_id_random'])
user_id: Optional[Union[int, str]] = Field(None, **base_fields['user_id_random'])
address_location_id: Optional[Union[int, str]] = Field(None, **base_fields['address_id_random'])
contact_1_id: Optional[Union[int, str]] = Field(None, **base_fields['contact_id_random'])
contact_2_id: Optional[Union[int, str]] = Field(None, **base_fields['contact_id_random'])
contact_3_id: Optional[Union[int, str]] = Field(None, **base_fields['contact_id_random'])
# --- Standardized Legacy / Internal IDs (Excluded) ---
id_random: Optional[str] = Field(None, alias='event_id_random', exclude=True)
account_id_random: Optional[str] = Field(None, exclude=True)
poc_event_person_id_random: Optional[str] = Field(None, exclude=True)
poc_person_id_random: Optional[str] = Field(None, exclude=True)
user_id_random: Optional[str] = Field(None, exclude=True)
address_location_id_random: Optional[str] = Field(None, exclude=True)
contact_1_id_random: Optional[str] = Field(None, exclude=True)
contact_2_id_random: Optional[str] = Field(None, exclude=True)
contact_3_id_random: Optional[str] = Field(None, exclude=True)
@root_validator(pre=True)
def map_v3_ids(cls, values):
"""
Vision Transformer:
Map DB keys to clean API keys and strip internal integers during READ operations.
"""
# 1. Map Random Strings to Clean Names
rid = values.get('id_random') or values.get('event_id_random')
if rid and isinstance(rid, str):
values['id'] = rid
values['event_id'] = rid
if a_rid := values.get('account_id_random'): values['account_id'] = a_rid
if pep_rid := values.get('poc_event_person_id_random'): values['poc_event_person_id'] = pep_rid
if pp_rid := values.get('poc_person_id_random'): values['poc_person_id'] = pp_rid
if u_rid := values.get('user_id_random'): values['user_id'] = u_rid
if al_rid := values.get('address_location_id_random'): values['address_location_id'] = al_rid
if c1_rid := values.get('contact_1_id_random'): values['contact_1_id'] = c1_rid
if c2_rid := values.get('contact_2_id_random'): values['contact_2_id'] = c2_rid
if c3_rid := values.get('contact_3_id_random'): values['contact_3_id'] = c3_rid
# 2. Prevent "Collision Population" or leakage of integers during API responses
for k in ['id', 'event_id', 'account_id', 'poc_event_person_id', 'poc_person_id', 'user_id', 'address_location_id', 'contact_1_id', 'contact_2_id', 'contact_3_id']:
val = values.get(k)
if val is not None and not isinstance(val, str):
if values.get(f'{k}_random') or (k=='id' and values.get('id_random')):
del values[k]
return values
code: Optional[str] = Field(
alias = 'event_code'
)
external_person_id: Optional[str] # Person ID generated by external system (should be stable and not change)
lu_event_type_id: Optional[int] = Field(None, exclude=True)
#lu_event_type: Optional[str] # Needs to be reviewed
conference: Optional[bool] # Also in Event_Cfg_Base model
# type_name: Optional[str] = Field(
# alias = 'type'
# )
type: Optional[str]
name: Optional[str]
summary: Optional[str]
description: Optional[str]
format: Optional[str]
lu_time_zone_id: Optional[int]
timezone: Optional[str]
start_datetime: Optional[datetime.datetime] = None
end_datetime: Optional[datetime.datetime] = None
recurring: Optional[bool]
recurring_pattern: Optional[str]
recurring_start_time: Optional[str]
recurring_end_time: Optional[str]
recurring_text: Optional[str]
weekday_sunday: Optional[bool]
weekday_monday: Optional[bool]
weekday_tuesday: Optional[bool]
weekday_wednesday: Optional[bool]
weekday_thursday: Optional[bool]
weekday_friday: Optional[bool]
weekday_saturday: Optional[bool]
location_address_json: Optional[Union[Json, None]]
location_text: Optional[str]
online_start: Optional[datetime.datetime] = None
online_end: Optional[datetime.datetime] = None
reg_deadline_1: Optional[datetime.datetime] = None
reg_deadline_2: Optional[datetime.datetime] = None
reg_deadline_3: Optional[datetime.datetime] = None
reg_deadline_4: Optional[datetime.datetime] = None
max_registrants: Optional[int]
private: Optional[bool] # invite only
physical: Optional[bool] # physical in person event
virtual: Optional[bool] # virtual remote access event
contact_li_json: Optional[Union[Json, None]] # list of dicts (custom for client); this is SQL FULLTEXT() indexed
attend_url: Optional[str]
attend_url_passcode: Optional[str]
attend_phone: Optional[str]
attend_phone_passcode: Optional[str]
attend_text: Optional[str]
attend_json: Optional[Union[Json, None]]
# access_key: Optional[str] # Maybe use in the future?
passcode: Optional[str]
file_count: Optional[int]
internal_use_count: Optional[int] # Should be renamed to "internal_use_file_count"???
event_file_id_li_json: Optional[Union[Json, None]] # List of file IDs (actually id_random)
file_count_all: Optional[int] # Of all files under a session
status: Optional[str]
review: Optional[bool]
approve: Optional[bool]
ready: Optional[bool]
ready_on: Optional[datetime.datetime]
archive: Optional[bool] # Also in Event_Cfg_Base model
archive_on: Optional[datetime.datetime] # Also in Event_Cfg_Base model
mod_abstracts_json: Optional[Union[Json, None]]
mod_badges_json: Optional[Union[Json, None]]
mod_exhibits_json: Optional[Union[Json, None]]
mod_pres_mgmt_json: Optional[Union[Json, None]]
cfg_json: Optional[Union[Json, None]] # Store per event config options; Not currently used 2024-06-11
data_json: Optional[Union[Json, None]] # For key value data. Careful with overwriting existing fields! Not currently used 2024-06-11
enable: Optional[bool] # Also in Event_Cfg_Base model
enable_from: Optional[datetime.datetime] = None
enable_to: Optional[datetime.datetime] = None
hide: Optional[bool] # Also in Event_Cfg_Base model
priority: Optional[bool]
sort: Optional[int]
group: Optional[str]
notes: Optional[str]
created_on: Optional[datetime.datetime] = None
updated_on: Optional[datetime.datetime] = None
# --- IDAA Recovery Meetings: Convenience Data (Flat) ---
# These fields are primarily for the flat "Meeting" view used by the IDAA mobile/web apps.
# Note: We prioritize string IDs (id_random) for all external API consumers.
address_id_random: Optional[str] = Field(None, **base_fields['address_id_random'])
address_name: Optional[str]
address_line_1: Optional[str]
address_line_2: Optional[str]
address_line_3: Optional[str]
address_city: Optional[str]
address_country_subdivision_code: Optional[str]
address_country_subdivision_name: Optional[str]
address_postal_code: Optional[str]
address_country_alpha_2_code: Optional[str]
address_country_name: Optional[str]
contact_1_name: Optional[str] # Avoid using or use as something different?
contact_1_full_name: Optional[str] # Yes... it is the same as "name"
contact_1_email: Optional[str]
contact_1_phone_mobile: Optional[str]
contact_1_phone_home: Optional[str]
contact_1_phone_office: Optional[str]
contact_1_phone_land: Optional[str]
contact_1_phone_fax: Optional[str]
contact_1_phone_other: Optional[str]
contact_1_other_text: Optional[str]
contact_2_name: Optional[str] # Avoid using or use as something different?
contact_2_full_name: Optional[str] # Yes... it is the same as "name"
contact_2_email: Optional[str]
contact_2_phone_mobile: Optional[str]
contact_2_phone_home: Optional[str]
contact_2_phone_office: Optional[str]
contact_2_phone_land: Optional[str]
contact_2_phone_fax: Optional[str]
contact_2_phone_other: Optional[str]
contact_2_other_text: Optional[str]
_processed_at: datetime.datetime = PrivateAttr(default_factory=datetime.datetime.now)
@validator('created_on', always=True)
def created_on_utc(cls, v, values, **kwargs):
if isinstance(v, datetime.datetime):
return v.astimezone(pytz.UTC).isoformat()
else: return v
@validator('updated_on', always=True)
def updated_on_utc(cls, v, values, **kwargs):
if isinstance(v, datetime.datetime):
return v.astimezone(pytz.UTC).isoformat()
else: return v
# Fields that are part of the model (for reading) but should not be saved to the DB table.
# These convenience fields are joined in the view.
fields_to_exclude_from_db: ClassVar[list] = [
'address_name', 'address_line_1', 'address_line_2', 'address_line_3', 'address_city',
'address_country_subdivision_code', 'address_country_subdivision_name', 'address_postal_code',
'address_country_alpha_2_code', 'address_country_name',
'contact_1_name', 'contact_1_full_name', 'contact_1_email', 'contact_1_phone_mobile',
'contact_1_phone_home', 'contact_1_phone_office', 'contact_1_phone_land', 'contact_1_phone_fax',
'contact_1_phone_other', 'contact_1_other_text',
'contact_2_name', 'contact_2_full_name', 'contact_2_email', 'contact_2_phone_mobile',
'contact_2_phone_home', 'contact_2_phone_office', 'contact_2_phone_land', 'contact_2_phone_fax',
'contact_2_phone_other', 'contact_2_other_text'
]
class Config:
underscore_attrs_are_private = True
allow_population_by_field_name = True
fields = base_fields
# ### END ### API Event Models ### Event_Meeting_Flat_Base() ###