Fix: Make forced account filtering schema-aware to prevent crashes on specialized views

app/lib_api_crud_v3.py

@@ -79,22 +79,40 @@ def check_account_access(sql_result: Any, account: AccountContext, obj_name: str
         return False
     return True
 
-def apply_forced_account_filter(and_qry_dict: Optional[Dict], account: AccountContext, model: Any, obj_name: str) -> Dict:
+def apply_forced_account_filter(and_qry_dict: Optional[Dict], account: AccountContext, model: Any, obj_name: str, table_name: str = None) -> Dict:
     """
     Secure Search Filtering.
     
     Automatically appends an `account_id` filter to database queries to ensure 
     users only retrieve records associated with their own account.
+    
+    Now schema-aware: checks if the column actually exists in the DB before applying.
     """
     forced = and_qry_dict or {}
     if account.super or account.auth_method == 'bypass':
         return forced
     
+    # 1. Determine the target column
+    target_col = 'account_id'
     if obj_name == 'account':
-        forced['id'] = account.account_id
+        target_col = 'id'
-    elif model and hasattr(model, '__fields__') and 'account_id' in model.__fields__:
+    
-        forced['account_id'] = account.account_id
+    # 2. Check if the model even supports it
+    if model and hasattr(model, '__fields__') and target_col not in model.__fields__:
+        return forced
+
+    # 3. If we have a table name, verify the column exists in the physical DB schema
+    # (Important for Views that might exclude account_id for performance/privacy)
+    if table_name:
+        from app import lib_sql_core
+        from sqlalchemy import text
+        try:
+            lib_sql_core.db.execute(text(f"SELECT `{target_col}` FROM `{table_name}` LIMIT 0"))
+        except Exception:
+            log.warning(f"Forced account filter skipped: Column '{target_col}' not found in '{table_name}'.")
+            return forced
         
+    forced[target_col] = account.account_id
     return forced
 
 def filter_order_by(order_by_li: Any, model: Any, table_name: str = None) -> Optional[Dict[str, str]]:

app/routers/api_crud_v3.py

@@ -216,7 +216,7 @@ async def get_obj_li(
 
     order_by_li = filter_order_by(order_by_li, base_name, table_name)
     status_filter = get_supported_filters(base_name, status_filter)
-    and_qry_dict_obj = apply_forced_account_filter(and_qry_dict_obj, account, base_name, obj_name)
+    and_qry_dict_obj = apply_forced_account_filter(and_qry_dict_obj, account, base_name, obj_name, table_name=table_name)
 
     if for_obj_type and for_obj_id:
         resolved_for_obj_id = redis_lookup_id_random(record_id_random=for_obj_id, table_name=for_obj_type)

app/routers/api_crud_v3_nested.py

@@ -98,7 +98,7 @@ async def get_child_obj_li(
     else:
         return mk_resp(data=False, status_code=404, response=response, status_message="Parent not found.")
 
-    and_qry_dict_obj = apply_forced_account_filter(and_qry_dict_obj, account, base_name, obj_name)
+    and_qry_dict_obj = apply_forced_account_filter(and_qry_dict_obj, account, base_name, obj_name, table_name=table_name)
 
     sql_result = sql_select(
         table_name=table_name,