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 # from app.models.account_models import Account_Base from app.models.contact_models import Contact_Base from app.models.membership_person_models import Membership_Person_Base from app.models.organization_models import Organization_Base # from app.models.user_models import User_Base class Person_Base(BaseModel): log.setLevel(logging.DEBUG) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(locals()) id_random: Optional[str] = Field( **base_fields['person_id_random'], alias = 'person_id_random', default_factory = lambda:secrets.token_urlsafe(default_num_bytes), ) id: Optional[int] = Field( alias = 'person_id' ) account_id_random: Optional[str] account_id: Optional[int] contact_id_random: Optional[str] contact_id: Optional[int] organization_id_random: Optional[str] organization_id: Optional[int] user_id_random: Optional[str] user_id: Optional[int] membership_person_id_random: Optional[str] # Linked from membership_person using the v_person view membership_person_id: Optional[int] # Linked from membership_person using the v_person view given_name: Optional[str] family_name: Optional[str] middle_name: Optional[str] prefix: Optional[str] suffix: Optional[str] full_name: Optional[str] informal_name: Optional[str] title: Optional[str] organization_name: Optional[str] tagline: Optional[str] notes: Optional[str] created_on: Optional[datetime.datetime] = None updated_on: Optional[datetime.datetime] = None # Including other related objects # archive_list: Optional[list] # Archive_Base() event_list: Optional[list] # Event_Base() # Priority l1 hosted_file_list: Optional[list] # Hosted_File_Base() # Priority l2 journal_list: Optional[list] # Journal_Base() # Priority l3 contact: Optional[Union[Contact_Base, None]] membership_person: Optional[Membership_Person_Base] # Priority l2 member person record of the person membership_type_list: Optional[list] # Membership_Type_Base() list of member types the person has... for now this should be only one at most. membership_type_person_list: Optional[list] # Membership_Type_Person_Base() list of member type person records... for now this should be only one at most. membership_group_list: Optional[list] # Membership_Group_Base() list of member groups the person has membership_group_person_list: Optional[list] # Membership_Group_Person_Base() list of member group person records order_list: Optional[list] # Order_Base() # Priority l2 order_cart_list: Optional[list] # Order_Base() # Priority l2 organization: Optional[Union[Organization_Base, None]] post_list: Optional[list] # Post_Base() # Priority l1 user: Optional[dict] # Optional[Union[User_Base, None]] _processed_at: datetime.datetime = PrivateAttr(default_factory=datetime.datetime.now) #@validator('person_id_random', always=True) def person_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 person_id_lookup(cls, v, values, **kwargs): log.setLevel(logging.WARNING) log.debug(locals()) if values.get('id_random', None): # 'id_random' in values and values['id_random']: log.debug(values['id_random']) return redis_lookup_id_random(record_id_random=values['id_random'], table_name='person') return None @validator('account_id', always=True) def account_id_lookup(cls, v, values, **kwargs): log.setLevel(logging.WARNING) log.debug(locals()) if values['account_id_random']: return redis_lookup_id_random(record_id_random=values['account_id_random'], table_name='account') return None @validator('contact_id', always=True) def contact_id_lookup(cls, v, values, **kwargs): log.setLevel(logging.WARNING) log.debug(locals()) if values.get('contact_id_random', None): # values['contact_id_random']: return redis_lookup_id_random(record_id_random=values['contact_id_random'], table_name='contact') return None @validator('organization_id', always=True) def organization_id_lookup(cls, v, values, **kwargs): log.setLevel(logging.WARNING) log.debug(locals()) if values['organization_id_random']: return redis_lookup_id_random(record_id_random=values['organization_id_random'], table_name='organization') return None @validator('user_id', always=True) def user_id_lookup(cls, v, values, **kwargs): log.setLevel(logging.WARNING) log.debug(locals()) if values['user_id_random']: return redis_lookup_id_random(record_id_random=values['user_id_random'], table_name='user') return None class Config: underscore_attrs_are_private = True allow_population_by_field_name = True fields = base_fields #from .supporting_core_models import Address_Base, Contact_Base, Organization_Base, User_Base Person_Base.update_forward_refs()