From d5844579977f8dfce1c6ddd7f4e792c487a90761 Mon Sep 17 00:00:00 2001 From: Scott Idem Date: Tue, 6 Jan 2026 16:29:09 -0500 Subject: [PATCH] fix(legacy): resolve 422 error on site domain lookup and enhance V3 filtering --- app/models/site_domain_models.py | 18 ++++++++++++++++-- app/routers/api_crud.py | 22 +++++++++++++++++++++- app/routers/api_crud_v3.py | 31 +++++++++++++++++++++++-------- 3 files changed, 60 insertions(+), 11 deletions(-) diff --git a/app/models/site_domain_models.py b/app/models/site_domain_models.py index 84a3e79..7148a77 100644 --- a/app/models/site_domain_models.py +++ b/app/models/site_domain_models.py @@ -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 diff --git a/app/routers/api_crud.py b/app/routers/api_crud.py index e868abf..00f09bc 100644 --- a/app/routers/api_crud.py +++ b/app/routers/api_crud.py @@ -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, diff --git a/app/routers/api_crud_v3.py b/app/routers/api_crud_v3.py index d215a6f..c115e8e 100644 --- a/app/routers/api_crud_v3.py +++ b/app/routers/api_crud_v3.py @@ -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)