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

fix(legacy): resolve 422 error on site domain lookup and enhance V3 filtering

Browse Source
This commit is contained in:
Scott Idem
2026-01-06 16:29:09 -05:00
parent 459bd89198
commit d584457997
3 changed files with 60 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
app/models/site_domain_models.py
View File

@@ -100,8 +100,8 @@ class Site_Domain_FQDN_ID_Base(BaseModel):
log.debug(locals())
id_random: Optional[str] = Field(
# **base_fields['site_domain_id_random'],
# alias = 'site_domain_id_random',
**base_fields['site_domain_id_random'],
alias = 'site_domain_id_random',
)
id: Optional[int] = Field(
alias = 'site_domain_id'
@@ -157,6 +157,20 @@ 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

22
app/routers/api_crud.py
View File

@@ -699,8 +699,28 @@ async def get_obj_l2(
# exclude: Optional[list] = [],
# exclude_none: Optional[bool] = True,
commons: Common_Route_Params = Depends(common_route_params),
commons: Common_Route_Params = Depends(common_route_params_min),
):
# ### SECTION ### Special Case: site/domain lookup by FQDN
if obj_type_l1 == 'site' and obj_type_l2 == 'domain':
log.info(f'Special Case: Site Domain lookup by FQDN: {obj_id}')
table_name = 'v_site_domain_fqdn_id' if use_alt_table else 'v_site_domain'
base_name = Site_Domain_FQDN_ID_Base if use_alt_base else Site_Domain_Base
sql_result = sql_select(
table_name=table_name,
field_name='fqdn',
field_value=obj_id,
as_list=False
)
if sql_result:
resp_data = base_name(**sql_result).dict(by_alias=commons.by_alias, exclude_unset=commons.exclude_unset)
return mk_resp(data=resp_data, response=commons.response)
else:
return mk_resp(data=False, status_code=404, response=commons.response)
# ### SECTION ### Call generic function to get the object
return handle_get_obj_id(
obj_type_l1=obj_type_l1,

31
app/routers/api_crud_v3.py
View File

@@ -35,11 +35,10 @@ def safe_json_loads(json_str: Optional[str]) -> Any:
log.warning(f"Failed to parse JSON string: {json_str}. Error: {e}")
return None
def filter_order_by(order_by_li: Any, model: Any) -> Optional[Dict[str, str]]:
def filter_order_by(order_by_li: Any, model: Any, table_name: str = None) -> Optional[Dict[str, str]]:
"""
Filters the order_by_li dictionary to only include fields present in the Pydantic model.
This prevents SQL errors when the frontend requests ordering by fields that don't exist
on specific objects (e.g., 'priority' or 'sort' on 'account').
Filters the order_by_li dictionary to only include fields present in the Pydantic model
AND actually present in the database table/view.
"""
if not order_by_li or not isinstance(order_by_li, dict) or not model:
return order_by_li
@@ -47,12 +46,28 @@ def filter_order_by(order_by_li: Any, model: Any) -> Optional[Dict[str, str]]:
if not hasattr(model, '__fields__'):
return order_by_li
# Get all field names and aliases from the model
# 1. Filter by Pydantic Model Fields/Aliases
model_fields = set(model.__fields__.keys())
model_fields.update({f.alias for f in model.__fields__.values() if f.alias})
filtered = {k: v for k, v in order_by_li.items() if k in model_fields}
# 2. Filter by actual DB Column existence (Dry run)
if table_name and filtered:
from app.db_sql import db
from sqlalchemy import text
final_filtered = {}
for column in filtered:
try:
# Use a lightweight query to check if column exists
db.execute(text(f"SELECT `{column}` FROM `{table_name}` LIMIT 0"))
final_filtered[column] = filtered[column]
except Exception:
log.warning(f"Column '{column}' does not exist in '{table_name}'. Removing from order_by_li.")
continue
filtered = final_filtered
if len(filtered) != len(order_by_li):
log.info(f"Filtered order_by_li. Removed fields: {set(order_by_li.keys()) - set(filtered.keys())}")
@@ -279,7 +294,7 @@ async def get_obj_li(
if not table_name or not base_name:
return mk_resp(data=False, status_code=500, response=response, status_message=f"Configuration for object type '{obj_name}' (view: {view}) is incomplete.")
order_by_li = filter_order_by(order_by_li, base_name)
order_by_li = filter_order_by(order_by_li, base_name, table_name)
status_filter = get_supported_filters(base_name, status_filter)
if for_obj_type and for_obj_id:
@@ -376,7 +391,7 @@ async def search_obj_li(
if not table_name or not base_name:
return mk_resp(data=False, status_code=500, response=response, status_message=f"Configuration for object type '{obj_name}' (view: {view}) is incomplete.")
order_by_li = filter_order_by(order_by_li, base_name)
order_by_li = filter_order_by(order_by_li, base_name, table_name)
status_filter = get_supported_filters(base_name, status_filter)
searchable_fields = obj_cfg.get('searchable_fields')
@@ -640,7 +655,7 @@ async def get_child_obj_li(
if not table_name or not base_name:
return mk_resp(data=False, status_code=500, response=response, status_message=f"Configuration for object type '{obj_name}' is incomplete.")
order_by_li = filter_order_by(order_by_li, base_name)
order_by_li = filter_order_by(order_by_li, base_name, table_name)
status_filter = get_supported_filters(base_name, status_filter)
resolved_parent_id = redis_lookup_id_random(record_id_random=parent_obj_id, table_name=parent_obj_type)