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 ..lib_general import * from .common_field_schema import base_fields, default_num_bytes #from .account_model import Account_Base from .contact_model import Contact_Base #from .organization_model import Organization_Base #from .person_model import Person_Base from .user_role_model import User_Role_Base class User_New_Base(BaseModel): log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(locals()) id_random: Optional[str] = Field( **base_fields['user_id_random'], alias='user_id_random', default_factory=lambda:secrets.token_urlsafe(default_num_bytes), ) id: Optional[int] = Field( #alias='user_id' ) account_id_random: str account_id: Optional[int] username: str name: str email: str new_password: str password: Optional[str] allow_auth_key: Optional[int] enable: Optional[bool] = False enable_from: Optional[datetime.datetime] = datetime.datetime.now(datetime.timezone.utc) enable_to: Optional[datetime.datetime] = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(days=365) #super: Optional[bool] = False #manager: Optional[bool] = False administrator: Optional[bool] = False public: Optional[bool] = False verified: Optional[bool] = False notes: Optional[str] _processed_at: datetime.datetime = PrivateAttr(default_factory=datetime.datetime.now) #@validator('user_id_random', always=True) def user_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 user_id_lookup(cls, v, values, **kwargs): log.setLevel(logging.WARNING) log.debug(locals()) if values['id_random']: log.debug(values['id_random']) return redis_lookup_id_random(record_id_random=values['id_random'], table_name='user') 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('password', always=True) def hash_new_password(cls, v, values, **kwargs): log.setLevel(logging.WARNING) log.debug(locals()) if values['new_password']: return secure_hash_string(string=values['new_password']) return None class Config: underscore_attrs_are_private = True fields = base_fields class User_Out_Base(BaseModel): log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(locals()) id_random: Optional[str] = Field( **base_fields['user_id_random'], alias='user_id_random', ) 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] person_id_random: Optional[str] #person_id: Optional[int] username: Optional[str] name: Optional[str] email: Optional[str] email_verified: Optional[bool] password: Optional[str] allow_auth_key: Optional[int] auth_key: Optional[str] enable: Optional[bool] enable_from: Optional[datetime.datetime] enable_to: Optional[datetime.datetime] super: Optional[bool] manager: Optional[bool] administrator: Optional[bool] public: Optional[bool] verified: Optional[bool] status_id: Optional[int] status_name: Optional[str] password_set_on: Optional[datetime.datetime] password_reset_token: Optional[str] password_reset_expire_on: Optional[datetime.datetime] logged_in_on: Optional[datetime.datetime] last_activity_on: Optional[datetime.datetime] #account: Optional[Account_Base]# = Account_Base() #contact: Optional[Contact_Base]# = Contact_Base() #organization: Optional[Organization_Base]# = Organization_Base() #person: Optional[Person_Base]# = Person_Base() role_list: Optional[list] = []# = User_Role_Base() notes: Optional[str] created_on: Optional[datetime.datetime] updated_on: Optional[datetime.datetime] _processed_at: datetime.datetime = PrivateAttr(default_factory=datetime.datetime.now) class Config: underscore_attrs_are_private = True fields = base_fields class User_Base(BaseModel): log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(locals()) #from .account_model import Account_Base #from .contact_model import Contact_Base #from .organization_model import Organization_Base #from .person_model import Person_Base #if TYPE_CHECKING: #from .person_model import Person_Base id_random: Optional[str] = Field( **base_fields['user_id_random'], alias='user_id_random', default_factory=lambda:secrets.token_urlsafe(default_num_bytes), ) id: Optional[int] = Field( #alias='user_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] person_id_random: Optional[str] person_id: Optional[int] username: Optional[str] name: Optional[str] email: Optional[str] email_verified: Optional[bool] password: Optional[str] allow_auth_key: Optional[int] auth_key: Optional[str] enable: Optional[bool] enable_from: Optional[datetime.datetime] = None enable_to: Optional[datetime.datetime] = None super: Optional[bool] manager: Optional[bool] administrator: Optional[bool] public: Optional[bool] verified: Optional[bool] status_id: Optional[int] status_name: Optional[str] password_set_on: Optional[datetime.datetime] = None password_reset_token: Optional[str] = None password_reset_expire_on: Optional[datetime.datetime] = None logged_in_on: Optional[datetime.datetime] = None last_activity_on: Optional[datetime.datetime] = None #account: Optional[Account_Base]# = Account_Base() contact: Optional[Contact_Base]# = Contact_Base() #organization: Optional[Organization_Base]# = Organization_Base() #person: Optional[Person_Base]# = Person_Base() role_list: Optional[list] = []# = User_Role_Base() notes: Optional[str] created_on: Optional[datetime.datetime] = None updated_on: Optional[datetime.datetime] = None _processed_at: datetime.datetime = PrivateAttr(default_factory=datetime.datetime.now) #@validator('user_id_random', always=True) def user_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 user_id_lookup(cls, v, values, **kwargs): log.setLevel(logging.WARNING) log.debug(locals()) if values['id_random']: log.debug(values['id_random']) return redis_lookup_id_random(record_id_random=values['id_random'], table_name='user') 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['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('person_id', always=True) def person_id_lookup(cls, v, values, **kwargs): log.setLevel(logging.WARNING) log.debug(locals()) if values['person_id_random']: return redis_lookup_id_random(record_id_random=values['person_id_random'], table_name='person') return None class Config: underscore_attrs_are_private = True fields = base_fields #User_Base.update_forward_refs()