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 class Event_Base(BaseModel): log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(locals()) # --- Standardized Vision IDs (Strings) --- id: Optional[str] = Field(None, **base_fields['event_id_random']) event_id: Optional[str] = Field(None, **base_fields['event_id_random']) account_id: Optional[str] = Field(None, **base_fields['account_id_random']) poc_event_person_id: Optional[str] = Field(None, **base_fields['event_person_id_random']) poc_person_id: Optional[str] = Field(None, **base_fields['person_id_random']) user_id: Optional[str] = Field(None, **base_fields['user_id_random']) address_location_id: Optional[str] = Field(None, **base_fields['address_id_random']) contact_1_id: Optional[str] = Field(None, **base_fields['contact_id_random']) contact_2_id: Optional[str] = Field(None, **base_fields['contact_id_random']) contact_3_id: Optional[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. """ # 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 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']: if k in values and not isinstance(values[k], str): 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 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) --- id: Optional[str] = Field(None, **base_fields['event_id_random']) event_id: Optional[str] = Field(None, **base_fields['event_id_random']) account_id: Optional[str] = Field(None, **base_fields['account_id_random']) poc_event_person_id: Optional[str] = Field(None, **base_fields['event_person_id_random']) poc_person_id: Optional[str] = Field(None, **base_fields['person_id_random']) user_id: Optional[str] = Field(None, **base_fields['user_id_random']) address_location_id: Optional[str] = Field(None, **base_fields['address_id_random']) contact_1_id: Optional[str] = Field(None, **base_fields['contact_id_random']) contact_2_id: Optional[str] = Field(None, **base_fields['contact_id_random']) contact_3_id: Optional[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. """ # 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 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']: if k in values and not isinstance(values[k], str): 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 class Config: underscore_attrs_are_private = True allow_population_by_field_name = True fields = base_fields # ### END ### API Event Models ### Event_Meeting_Flat_Base() ###