From 5af3f44a534375d7ff64877de2bf27ec80885808 Mon Sep 17 00:00:00 2001 From: Scott Idem Date: Tue, 27 Jan 2026 12:12:51 -0500 Subject: [PATCH] Refactor Contact and User models to use Vision ID string pattern Updated Contact_Base and User_Base (including New/Out variants) to use standardized string IDs mapped from random IDs via root_validator. Removed legacy integer ID fields and lookup validators to support V3 Vision standards. This completes the refactor chain for Person and Post dependencies. --- app/models/contact_models.py | 110 +++++--------- app/models/user_models.py | 287 ++++++++++++----------------------- 2 files changed, 136 insertions(+), 261 deletions(-) diff --git a/app/models/contact_models.py b/app/models/contact_models.py index 1eb5f12..0201488 100644 --- a/app/models/contact_models.py +++ b/app/models/contact_models.py @@ -1,7 +1,7 @@ import datetime, pytz from typing import Dict, List, Optional, Set, Union -from pydantic import BaseModel, EmailStr, Field, Json, PrivateAttr, ValidationError, validator +from pydantic import BaseModel, EmailStr, Field, Json, PrivateAttr, ValidationError, validator, root_validator from app.db_sql import get_id_random, redis_lookup_id_random from app.lib_general import log, logging @@ -16,22 +16,15 @@ class Contact_Base(BaseModel): log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL log.debug(locals()) - id_random: Optional[str] = Field( - **base_fields['contact_id_random'], - alias = 'contact_id_random', - # default_factory = lambda:secrets.token_urlsafe(default_num_bytes), - ) - id: Optional[int] = Field( - alias = 'contact_id' - ) - account_id_random: Optional[str] - account_id: Optional[int] - - address_id_random: Optional[str] - address_id: Optional[int] - - linked_address_id_random: Optional[str] - linked_address_id: Optional[int] + # --- Standardized Vision IDs (Strings) --- + id: Optional[str] = Field(None, **base_fields['contact_id_random']) + contact_id: Optional[str] = Field(None, **base_fields['contact_id_random']) + + account_id: Optional[str] = Field(None, **base_fields['account_id_random']) + address_id: Optional[str] = Field(None, **base_fields['address_id_random']) + + # NOTE: Linked Address ID is actually the old contact.address_id (Legacy?) + linked_address_id: Optional[str] = Field(None, **base_fields['address_id_random']) for_type: Optional[str] for_id: Optional[int] @@ -88,63 +81,30 @@ class Contact_Base(BaseModel): _processed_at: datetime.datetime = PrivateAttr(default_factory=datetime.datetime.now) - #@validator('contact_id_random', always=True) - def contact_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 contact_id_lookup(cls, v, values, **kwargs): - log.setLevel(logging.WARNING) - log.debug(locals()) - - if isinstance(v, int) and v > 0: return v - elif id_random := values.get('id_random'): - return redis_lookup_id_random(record_id_random=id_random, table_name='contact') - return None - - @validator('account_id', always=True) - def account_id_lookup(cls, v, values, **kwargs): - log.setLevel(logging.WARNING) - log.debug(locals()) - - if isinstance(v, int) and v > 0: return v - elif id_random := values.get('account_id_random'): - return redis_lookup_id_random(record_id_random=id_random, table_name='account') - return None - - @validator('account_id_random', always=True) - def account_id_random_lookup(cls, v, values, **kwargs): - if isinstance(v, str) and len(v) >= 11: return v - elif account_id := values.get('account_id'): - return get_id_random(record_id=account_id, table_name='account') - return None - - @validator('address_id', always=True) - def address_id_lookup(cls, v, values, **kwargs): - log.setLevel(logging.WARNING) - log.debug(locals()) - - if isinstance(v, int) and v > 0: return v - elif id_random := values.get('address_id_random'): - return redis_lookup_id_random(record_id_random=id_random, table_name='address') - return None - - # NOTE: Linked Address ID is actually the old contact.address_id - # This should no longer be used... - @validator('linked_address_id', always=True) - def linked_address_id_lookup(cls, v, values, **kwargs): - log.setLevel(logging.WARNING) - log.debug(locals()) - - if isinstance(v, int) and v > 0: return v - elif id_random := values.get('linked_address_id_random'): - return redis_lookup_id_random(record_id_random=id_random, table_name='address') - return None + @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 + if rid := values.get('id_random') or values.get('contact_id_random'): + values['id'] = rid + values['contact_id'] = rid + + if a_rid := values.get('account_id_random'): + values['account_id'] = a_rid + if ad_rid := values.get('address_id_random'): + values['address_id'] = ad_rid + if lad_rid := values.get('linked_address_id_random'): + values['linked_address_id'] = lad_rid + + # 2. Prevent "Collision Population" + for k in ['id', 'contact_id', 'account_id', 'address_id', 'linked_address_id']: + if k in values and not isinstance(values[k], str) and values[k] is not None: + del values[k] + + return values @validator('for_id', pre=True, always=True) def for_id_lookup(cls, v, values, **kwargs): @@ -188,6 +148,6 @@ class Contact_Base(BaseModel): class Config: underscore_attrs_are_private = True - allow_population_by_field_name = True + allow_population_by_field_name = False fields = base_fields # ### END ### API Contact Models ### Contact_Base() ### diff --git a/app/models/user_models.py b/app/models/user_models.py index c302288..4b748b9 100644 --- a/app/models/user_models.py +++ b/app/models/user_models.py @@ -1,7 +1,7 @@ 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 pydantic import BaseModel, EmailStr, Field, Json, PrivateAttr, ValidationError, validator, root_validator from app.db_sql import get_id_random, redis_lookup_id_random from app.lib_general import log, logging, secure_hash_string @@ -18,29 +18,16 @@ class User_Base(BaseModel): log.setLevel(logging.INFO) # 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: Optional[str] - account_id: Optional[int] + # --- Standardized Vision IDs (Strings) --- + id: Optional[str] = Field(None, **base_fields['user_id_random']) + user_id: Optional[str] = Field(None, **base_fields['user_id_random']) + account_id: Optional[str] = Field(None, **base_fields['account_id_random']) + contact_id: Optional[str] = Field(None, **base_fields['contact_id_random']) + organization_id: Optional[str] = Field(None, **base_fields['organization_id_random']) + person_id: Optional[str] = Field(None, **base_fields['person_id_random']) account_name: Optional[str] - 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] @@ -93,61 +80,32 @@ class User_Base(BaseModel): _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 isinstance(v, int) and v > 0: return v - elif id_random := values.get('id_random'): - return redis_lookup_id_random(record_id_random=id_random, table_name='user') - return None - - @validator('account_id', always=True) - def account_id_lookup(cls, v, values, **kwargs): - if isinstance(v, int) and v > 0: return v - elif id_random := values.get('account_id_random'): - return redis_lookup_id_random(record_id_random=id_random, table_name='account') - return None - - @validator('account_id_random', always=True) - def account_id_random_lookup(cls, v, values, **kwargs): - if isinstance(v, str) and len(v) >= 11: return v - elif account_id := values.get('account_id'): - return get_id_random(record_id=account_id, 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): - if isinstance(v, int) and v > 0: return v - elif id_random := values.get('organization_id_random'): - return redis_lookup_id_random(record_id_random=id_random, table_name='organization') - return None - - @validator('person_id', always=True) - def person_id_lookup(cls, v, values, **kwargs): - if isinstance(v, int) and v > 0: return v - elif id_random := values.get('person_id_random'): - return redis_lookup_id_random(record_id_random=id_random, table_name='person') - return None + @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 + if rid := values.get('id_random') or values.get('user_id_random'): + values['id'] = rid + values['user_id'] = rid + + if a_rid := values.get('account_id_random'): + values['account_id'] = a_rid + if c_rid := values.get('contact_id_random'): + values['contact_id'] = c_rid + if o_rid := values.get('organization_id_random'): + values['organization_id'] = o_rid + if p_rid := values.get('person_id_random'): + values['person_id'] = p_rid + + # 2. Prevent "Collision Population" + for k in ['id', 'user_id', 'account_id', 'contact_id', 'organization_id', 'person_id']: + if k in values and not isinstance(values[k], str): + del values[k] + + return values @validator('password', always=True) def hash_new_password(cls, v, values, **kwargs): @@ -170,27 +128,16 @@ 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: Optional[str] - account_id: Optional[int] + # --- Standardized Vision IDs (Strings) --- + id: Optional[str] = Field(None, **base_fields['user_id_random']) + user_id: Optional[str] = Field(None, **base_fields['user_id_random']) + account_id: Optional[str] = Field(None, **base_fields['account_id_random']) + contact_id: Optional[str] = Field(None, **base_fields['contact_id_random']) + organization_id: Optional[str] = Field(None, **base_fields['organization_id_random']) + person_id: Optional[str] = Field(None, **base_fields['person_id_random']) + account_name: Optional[str] - 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: str name: str email: str @@ -224,59 +171,32 @@ class User_New_Base(BaseModel): _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 isinstance(v, int) and v > 0: return v - elif id_random := values.get('id_random'): - return redis_lookup_id_random(record_id_random=id_random, table_name='user') - return None - - @validator('account_id', always=True) - def account_id_lookup(cls, v, values, **kwargs): - if isinstance(v, int) and v > 0: return v - elif id_random := values.get('account_id_random'): - return redis_lookup_id_random(record_id_random=id_random, table_name='account') - return None - - @validator('account_id_random', always=True) - def account_id_random_lookup(cls, v, values, **kwargs): - if isinstance(v, str) and len(v) >= 11: return v - elif account_id := values.get('account_id'): - return get_id_random(record_id=account_id, table_name='account') - return None - - @validator('contact_id', always=True) - def contact_id_lookup(cls, v, values, **kwargs): - if isinstance(v, int) and v > 0: return v - elif id_random := values.get('contact_id_random'): - return redis_lookup_id_random(record_id_random=id_random, table_name='contact') - return None - - @validator('organization_id', always=True) - def organization_id_lookup(cls, v, values, **kwargs): - if isinstance(v, int) and v > 0: return v - elif id_random := values.get('organization_id_random'): - return redis_lookup_id_random(record_id_random=id_random, table_name='organization') - return None - - @validator('person_id', always=True) - def person_id_lookup(cls, v, values, **kwargs): - if isinstance(v, int) and v > 0: return v - elif id_random := values.get('person_id_random'): - return redis_lookup_id_random(record_id_random=id_random, table_name='person') - return None + @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 + if rid := values.get('id_random') or values.get('user_id_random'): + values['id'] = rid + values['user_id'] = rid + + if a_rid := values.get('account_id_random'): + values['account_id'] = a_rid + if c_rid := values.get('contact_id_random'): + values['contact_id'] = c_rid + if o_rid := values.get('organization_id_random'): + values['organization_id'] = o_rid + if p_rid := values.get('person_id_random'): + values['person_id'] = p_rid + + # 2. Prevent "Collision Population" + for k in ['id', 'user_id', 'account_id', 'contact_id', 'organization_id', 'person_id']: + if k in values and not isinstance(values[k], str): + del values[k] + + return values @validator('password', always=True) def hash_new_password(cls, v, values, **kwargs): @@ -299,27 +219,16 @@ 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', - ) - id: Optional[int] = Field( - alias = 'user_id' - ) + # --- Standardized Vision IDs (Strings) --- + id: Optional[str] = Field(None, **base_fields['user_id_random']) + user_id: Optional[str] = Field(None, **base_fields['user_id_random']) + account_id: Optional[str] = Field(None, **base_fields['account_id_random']) + contact_id: Optional[str] = Field(None, **base_fields['contact_id_random']) + organization_id: Optional[str] = Field(None, **base_fields['organization_id_random']) + person_id: Optional[str] = Field(None, **base_fields['person_id_random']) - account_id_random: Optional[str] - account_id: Optional[int] account_name: Optional[str] - 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] @@ -377,26 +286,32 @@ class User_Out_Base(BaseModel): _processed_at: datetime.datetime = PrivateAttr(default_factory=datetime.datetime.now) - @validator('account_id', always=True) - def account_id_lookup(cls, v, values, **kwargs): - if isinstance(v, int) and v > 0: return v - elif id_random := values.get('account_id_random'): - return redis_lookup_id_random(record_id_random=id_random, table_name='account') - return None - - @validator('account_id_random', always=True) - def account_id_random_lookup(cls, v, values, **kwargs): - if isinstance(v, str) and len(v) >= 11: return v - elif account_id := values.get('account_id'): - return get_id_random(record_id=account_id, table_name='account') - return None - - @validator('person_id', always=True) - def person_id_lookup(cls, v, values, **kwargs): - if isinstance(v, int) and v > 0: return v - elif id_random := values.get('person_id_random'): - return redis_lookup_id_random(record_id_random=id_random, table_name='person') - return None + @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 + if rid := values.get('id_random') or values.get('user_id_random'): + values['id'] = rid + values['user_id'] = rid + + if a_rid := values.get('account_id_random'): + values['account_id'] = a_rid + if c_rid := values.get('contact_id_random'): + values['contact_id'] = c_rid + if o_rid := values.get('organization_id_random'): + values['organization_id'] = o_rid + if p_rid := values.get('person_id_random'): + values['person_id'] = p_rid + + # 2. Prevent "Collision Population" + for k in ['id', 'user_id', 'account_id', 'contact_id', 'organization_id', 'person_id']: + if k in values and not isinstance(values[k], str): + del values[k] + + return values class Config: underscore_attrs_are_private = True