Scott.Idem/OSIT-AE-API-FastAPI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Vision ID: Standardize Site Domain and Journal objects with string-only IDs and searchable mapping

Browse Source
This commit is contained in:
Scott Idem
2026-01-19 15:57:00 -05:00
parent 2dbf47d874
commit 7db937f8af
10 changed files with 272 additions and 305 deletions
Download Patch File Download Diff File Expand all files Collapse all files

83
app/models/journal_entry_models.py
View File

@@ -1,7 +1,7 @@
import datetime, pytz
from typing import Dict, List, Optional, Set, Union, ClassVar
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 redis_lookup_id_random
from app.lib_general import log, logging
@@ -14,23 +14,18 @@ class Journal_Entry_Base(BaseModel):
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
id_random: Optional[str] = Field(
**base_fields['journal_entry_id_random'],
alias = 'journal_entry_id_random',
)
id: Optional[int] = Field(
alias = 'journal_entry_id'
)
journal_id_random: Optional[str]
journal_id: Optional[int]
# --- Standardized Vision IDs (Strings) ---
id: Optional[str] = Field(None, **base_fields['journal_entry_id_random'])
journal_entry_id: Optional[str] = Field(None, **base_fields['journal_entry_id_random'])
journal_id: Optional[str] = Field(None, **base_fields['journal_id_random'])
account_id: Optional[str] = Field(None, **base_fields['account_id_random'])
external_id: Optional[str] # ID generated by or for external systems (should be stable and not change)
import_id: Optional[str] # Used for import purposes to track the source of the data
code: Optional[str]
for_type: Optional[str] # 'person', 'user', 'account', etc
for_id: Optional[int]
for_id: Optional[int] = Field(None, exclude=True)
for_id_random: Optional[str]
name: Optional[str]
@@ -76,10 +71,11 @@ class Journal_Entry_Base(BaseModel):
personal: Optional[bool] = True
professional: Optional[bool] = False
parent_id_random: Optional[str]
parent_id: Optional[int]
parent_id: Optional[str] = Field(None, **base_fields['journal_entry_id_random'])
# parent_id_random: Optional[str]
related_entry_id_random: Optional[List[str]]
related_entry_id_li: Optional[List[int]]
related_entry_id_li: Optional[List[int]] = Field(None, exclude=True)
due_datetime: Optional[datetime.datetime]
due_alert: Optional[bool]
@@ -113,41 +109,38 @@ class Journal_Entry_Base(BaseModel):
_processed_at: datetime.datetime = PrivateAttr(default_factory=datetime.datetime.now)
@validator('id', always=True)
def journal_entry_id_lookup(cls, v, values, **kwargs):
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='journal_entry')
return None
@validator('journal_id', always=True)
def journal_id_lookup(cls, v, values, **kwargs):
if isinstance(v, int) and v > 0: return v
elif id_random := values.get('journal_id_random'):
return redis_lookup_id_random(record_id_random=id_random, table_name='journal')
return None
@validator('parent_id', always=True)
def parent_id_lookup(cls, v, values, **kwargs):
if isinstance(v, int) and v > 0: return v
elif id_random := values.get('parent_id_random'):
return redis_lookup_id_random(record_id_random=id_random, table_name='journal_entry')
return None
@validator('for_id', always=True)
def for_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values.get('for_id_random') and values.get('for_type'):
return redis_lookup_id_random(record_id_random=values['for_id_random'], table_name=values['for_type'])
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('journal_entry_id_random'):
values['id'] = rid
values['journal_entry_id'] = rid
if j_rid := values.get('journal_id_random'):
values['journal_id'] = j_rid
if a_rid := values.get('account_id_random'):
values['account_id'] = a_rid
if p_rid := values.get('parent_id_random'):
values['parent_id'] = p_rid
# 2. Prevent "Collision Population"
for k in ['id', 'journal_id', 'account_id', 'parent_id']:
if k in values and not isinstance(values[k], str):
del values[k]
return values
# 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] = ['file_count']
class Config:
underscore_attrs_are_private = True
allow_population_by_field_name = True
allow_population_by_field_name = False
fields = base_fields
# ### END ### API Journal Entry Models ### Journal_Entry_Base() ###

96
app/models/journal_models.py
View File

@@ -1,7 +1,7 @@
import datetime, pytz
from typing import Dict, List, Optional, Set, Union, ClassVar
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 redis_lookup_id_random
from app.lib_general import log, logging
@@ -16,28 +16,19 @@ class Journal_Base(BaseModel):
log.setLevel(logging.INFO) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
id_random: Optional[str] = Field(
**base_fields['journal_id_random'],
alias = 'journal_id_random',
)
id: Optional[int] = Field(
alias = 'journal_id'
)
account_id_random: Optional[str]
account_id: Optional[int]
person_id_random: Optional[str]
person_id: Optional[int]
user_id_random: Optional[str]
user_id: Optional[int]
# --- Standardized Vision IDs (Strings) ---
id: Optional[str] = Field(None, **base_fields['journal_id_random'])
journal_id: Optional[str] = Field(None, **base_fields['journal_id_random'])
account_id: Optional[str] = Field(None, **base_fields['account_id_random'])
person_id: Optional[str] = Field(None, **base_fields['person_id_random'])
user_id: Optional[str] = Field(None, **base_fields['user_id_random'])
external_id: Optional[str] # ID generated by or for external systems (should be stable and not change)
import_id: Optional[str] # Used for import purposes to track the source of the data
code: Optional[str]
for_type: Optional[str] # 'person', 'user', 'account', etc
for_id: Optional[int]
for_id: Optional[int] = Field(None, exclude=True)
for_id_random: Optional[str]
name: Optional[str]
@@ -137,51 +128,30 @@ class Journal_Base(BaseModel):
_processed_at: datetime.datetime = PrivateAttr(default_factory=datetime.datetime.now)
@validator('id', always=True)
def journal_id_lookup(cls, v, values, **kwargs):
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='journal')
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('person_id', always=True)
def person_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('person_id_random'):
return redis_lookup_id_random(record_id_random=id_random, table_name='person')
return None
@validator('user_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('user_id_random'):
return redis_lookup_id_random(record_id_random=id_random, table_name='user')
return None
@validator('for_id', always=True)
def for_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values.get('for_id_random') and values.get('for_type'):
return redis_lookup_id_random(record_id_random=values['for_id_random'], table_name=values['for_type'])
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('journal_id_random'):
values['id'] = rid
values['journal_id'] = rid
if a_rid := values.get('account_id_random'):
values['account_id'] = a_rid
if p_rid := values.get('person_id_random'):
values['person_id'] = p_rid
if u_rid := values.get('user_id_random'):
values['user_id'] = u_rid
# 2. Prevent "Collision Population"
for k in ['id', 'account_id', 'person_id', 'user_id']:
if k in values and not isinstance(values[k], str):
del values[k]
return values
# 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] = [
@@ -192,6 +162,6 @@ class Journal_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 Journal Models ### Journal_Base() ###

183
app/models/site_domain_models.py
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 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
@@ -14,16 +14,11 @@ class Site_Domain_Base(BaseModel):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
id_random: Optional[str] = Field(
**base_fields['site_domain_id_random'],
alias = 'site_domain_id_random',
)
id: Optional[int] = Field(
alias = 'site_domain_id'
)
site_id_random: Optional[str]
site_id: Optional[int]
# --- Standardized Vision IDs (Strings) ---
id: Optional[str] = Field(None, **base_fields['site_domain_id_random'])
site_domain_id: Optional[str] = Field(None, **base_fields['site_domain_id_random'])
site_id: Optional[str] = Field(None, **base_fields['site_id_random'])
account_id: Optional[str] = Field(None, **base_fields['account_id_random'])
fqdn: Optional[str]
@@ -38,59 +33,45 @@ class Site_Domain_Base(BaseModel):
cfg_json: Optional[Union[Json, None]] # In use 2024-03-04
hide: Optional[bool]
priority: Optional[bool]
sort: Optional[int]
group: Optional[str]
hide: Optional[bool] = None # Field missing in physical table but common in views
# priority: Optional[bool] # MISSING in physical table
# sort: Optional[int] # MISSING in physical table
# group: Optional[str] # MISSING in physical table
notes: Optional[str]
notes: Optional[str] = None # MISSING in physical table
created_on: Optional[datetime.datetime] = None
updated_on: Optional[datetime.datetime] = None
# Including convenience data
# This is only for convenience. Probably going to keep unless it causes a problem.
account_id: Optional[int]
account_id_random: Optional[str]
_processed_at: datetime.datetime = PrivateAttr(default_factory=datetime.datetime.now)
@validator('id', always=True)
def site_domain_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='site_domain')
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('site_id', always=True)
def site_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['site_id_random']:
return redis_lookup_id_random(record_id_random=values['site_id_random'], table_name='site')
return None
@root_validator(pre=True)
def map_v3_ids(cls, values):
"""
Vision Transformer:
Map DB-centric keys to clean API keys and strip internal integers.
"""
# 1. Map Random Strings to Clean Names
# We prioritize the random strings to ensure the Vision is string-based.
if rid := values.get('id_random') or values.get('site_domain_id_random'):
values['id'] = rid
values['site_domain_id'] = rid
if s_rid := values.get('site_id_random'):
values['site_id'] = s_rid
if a_rid := values.get('account_id_random'):
values['account_id'] = a_rid
# 2. Prevent "Collision Population"
for k in ['id', 'site_id', 'account_id']:
if k in values and not isinstance(values[k], str):
del values[k]
return values
class Config:
underscore_attrs_are_private = True
fields = base_fields
allow_population_by_field_name = True
allow_population_by_field_name = False
# ### END ### API Site Domain Models ### Site_Domain_Base() ###
@@ -99,19 +80,11 @@ class Site_Domain_FQDN_ID_Base(BaseModel):
log.setLevel(logging.WARNING) # DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL
log.debug(locals())
id_random: Optional[str] = Field(
**base_fields['site_domain_id_random'],
alias = 'site_domain_id_random',
)
id: Optional[int] = Field(
alias = 'site_domain_id'
)
site_id_random: Optional[str]
site_id: Optional[int]
site_domain_id_random: Optional[str]
site_domain_id: Optional[int]
# --- Standardized Vision IDs (Strings) ---
id: Optional[str] = Field(None, **base_fields['site_domain_id_random'])
site_domain_id: Optional[str] = Field(None, **base_fields['site_domain_id_random'])
site_id: Optional[str] = Field(None, **base_fields['site_id_random'])
account_id: Optional[str] = Field(None, **base_fields['account_id_random'])
fqdn: Optional[str]
@@ -124,19 +97,13 @@ class Site_Domain_FQDN_ID_Base(BaseModel):
valid_for: Optional[int] # number of hours
enable: Optional[bool]
hide: Optional[bool]
priority: Optional[bool]
sort: Optional[int]
group: Optional[str]
notes: Optional[str]
hide: Optional[bool] = None
notes: Optional[str] = None
created_on: Optional[datetime.datetime] = None
updated_on: Optional[datetime.datetime] = None
# Including convenience data
# This is only for convenience. Probably going to keep unless it causes a problem.
account_id: Optional[int]
account_id_random: Optional[str]
account_code: Optional[str] # Useful for export file naming
account_name: Optional[str] # Generally useful for display
account_enable: Optional[bool]
@@ -157,54 +124,22 @@ class Site_Domain_FQDN_ID_Base(BaseModel):
_processed_at: datetime.datetime = PrivateAttr(default_factory=datetime.datetime.now)
@validator('id', always=True)
def id_lookup(cls, v, values, **kwargs):
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='site_domain')
return None
@validator('site_domain_id_random', always=True)
def site_domain_id_random_lookup(cls, v, values, **kwargs):
if isinstance(v, str) and len(v) >= 11: return v
elif id_random := values.get('id_random'):
return id_random
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('site_id', always=True)
def site_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['site_id_random']:
return redis_lookup_id_random(record_id_random=values['site_id_random'], table_name='site')
return None
@validator('site_domain_id', always=True)
def site_domain_id_lookup(cls, v, values, **kwargs):
log.setLevel(logging.WARNING)
log.debug(locals())
if values['site_domain_id_random']:
return redis_lookup_id_random(record_id_random=values['site_domain_id_random'], table_name='site')
return None
@root_validator(pre=True)
def map_v3_ids(cls, values):
if rid := values.get('id_random') or values.get('site_domain_id_random'):
values['id'] = rid
values['site_domain_id'] = rid
if s_rid := values.get('site_id_random'):
values['site_id'] = s_rid
if a_rid := values.get('account_id_random'):
values['account_id'] = a_rid
for k in ['id', 'site_id', 'account_id']:
if k in values and not isinstance(values[k], str):
del values[k]
return values
class Config:
underscore_attrs_are_private = True
fields = base_fields
allow_population_by_field_name = True
allow_population_by_field_name = False
# ### END ### API Site Domain Models ### Site_Domain_FQDN_ID_Base() ###