V3 Migration Phase 1: Stabilize Hosted File models, IDs, and whitelisting. Added comprehensive verification tests.

This commit is contained in:
Scott Idem
2026-01-22 18:30:34 -05:00
parent df0ce7f910
commit 1837b442cf
2 changed files with 106 additions and 29 deletions

View File

@@ -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 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
@@ -14,22 +14,13 @@ class Hosted_File_Link_Base(BaseModel):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
# id_random: Optional[str] = Field(
# **base_fields['hosted_file_link_id_random'],
# alias = 'hosted_file_link_id_random',
# )
id: Optional[int] = Field(
#alias = 'hosted_file_link_id'
)
account_id_random: Optional[str]
account_id: Optional[int]
id: Optional[Union[int, str]] = Field(None)
account_id: Optional[Union[int, str]] = Field(None, **base_fields['account_id_random'])
hosted_file_id_random: Optional[str]
hosted_file_id: Optional[int]
hosted_file_id: Optional[Union[int, str]] = Field(None, **base_fields['hosted_file_id_random'])
link_to_type: Optional[str] # Should this be renamed to "link_to_obj_type" for clarity?
link_to_id_random: Optional[str] # Should this be renamed to "link_to_obj_id_random" for clarity?
link_to_id: Optional[int] # Should this be renamed to "link_to_obj_id" for clarity?
link_to_id: Optional[Union[int, str]] = Field(None) # Random string or integer
# notes: Optional[str]
created_on: Optional[datetime.datetime] = None
@@ -40,21 +31,33 @@ class Hosted_File_Link_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
@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 account_id
if a_rid := values.get('account_id_random'):
if not isinstance(values.get('account_id'), int):
values['account_id'] = a_rid
# 2. Map hosted_file_id
if f_rid := values.get('hosted_file_id_random'):
if not isinstance(values.get('hosted_file_id'), int):
values['hosted_file_id'] = f_rid
# 3. Map link_to_id
if l_rid := values.get('link_to_id_random'):
if not isinstance(values.get('link_to_id'), int):
values['link_to_id'] = l_rid
return values
@validator('link_to_id', always=True)
def link_to_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['link_to_id_random'] and values['link_to_type']:
return redis_lookup_id_random(record_id_random=values['link_to_id_random'], table_name=values['link_to_type'])
return None
# Fields that are part of the model (for reading) but should not be saved to the DB table
fields_to_exclude_from_db: ClassVar[list] = [
'link_to'
]
class Config:
underscore_attrs_are_private = True