Files
OSIT-AE-API-FastAPI/app/models/event_models.py

485 lines
22 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
default_qry_str: Optional[str] # Default query string used for searching and filtering events. Updated using SQL triggers and a SQL function
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
default_qry_str: Optional[str] # Default query string used for searching and filtering events. Updated using SQL triggers and a SQL function
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() ###