Error Bubbling: Implement machine-readable rich error objects for CRUD operations

app/lib_api_crud_v3.py

@@ -1,27 +1,58 @@
-from typing import Any, Dict, Optional
+from typing import Any, Dict, Optional, Union
 import json
 import logging
 import re
 
 from app.lib_general_v3 import AccountContext, StatusFilterParams
+from app.models.error_models import StandardError
 
 log = logging.getLogger(__name__)
 
-def format_db_error(raw_error: str) -> str:
+def format_db_error(raw_error: str) -> StandardError:
     """
-    Parses raw SQLAlchemy/MariaDB errors into user-friendly strings.
+    Parses raw SQLAlchemy/MariaDB errors into structured StandardError objects.
     """
     if not raw_error:
-        return ""
+        return StandardError(
+            category="unknown",
+            message="An unspecified database error occurred."
+        )
     
+    # 1. Extract Error Code and Message using regex
     # Standard MariaDB pattern: (code, "message")
-    match = re.search(r'\(\d+,\s*["\'](.*?)["\']\s*\)', raw_error)
+    code = None
-    if match:
+    message = raw_error
-        return match.group(1).strip()
+    recoverable = False
     
-    # Fallback: remove all (parenthesized) blocks which often contain codes
+    match = re.search(r'\((\d+),\s*["\'](.*?)["\']\s*\)', raw_error)
-    clean = re.sub(r'\(.*?\)', '', raw_error)
+    if match:
-    return clean.strip()
+        code = int(match.group(1))
+        message = match.group(2).strip()
+    else:
+        # Fallback: remove all (parenthesized) blocks which often contain codes
+        message = re.sub(r'\(.*?\)', '', raw_error).strip()
+
+    # 2. Categorize based on known MariaDB codes
+    # Ref: https://mariadb.com/kb/en/mariadb-error-codes/
+    if code in [1062]: # Duplicate Entry
+        category = "database_duplicate"
+    elif code in [1451, 1452]: # Foreign Key Constraint
+        category = "database_constraint"
+    elif code in [1045, 2002, 2003, 2006]: # Connection / Auth issues
+        category = "database_connection"
+        recoverable = True
+    elif code in [1054, 1146]: # Unknown column / Table
+        category = "database_schema"
+    else:
+        category = "database"
+
+    return StandardError(
+        category=category,
+        code=code,
+        message=message,
+        recoverable=recoverable,
+        details=raw_error if category == "database" else None # Only include raw details for uncategorized errors
+    )
 
 def check_account_access(sql_result: Any, account: AccountContext, obj_name: str = None) -> bool:
     """

app/models/error_models.py

@@ -0,0 +1,23 @@
+from typing import Optional, Any, Dict
+from pydantic import BaseModel, Field
+
+class StandardError(BaseModel):
+    """
+    Standardized machine-readable error structure for Aether.
+    Helps the frontend decide how to handle failures.
+    """
+    category: str = Field(..., description="Error category (e.g., 'database', 'validation', 'security')")
+    code: Optional[int] = Field(None, description="Specific error code (e.g., MariaDB error code)")
+    message: str = Field(..., description="Developer-friendly error message")
+    recoverable: bool = Field(False, description="If True, the frontend might want to retry or ask for user input")
+    details: Optional[Any] = Field(None, description="Raw technical details or traceback (if permitted)")
+
+    class Config:
+        schema_extra = {
+            "example": {
+                "category": "database",
+                "code": 1062,
+                "message": "Duplicate entry for key 'id_random'",
+                "recoverable": False
+            }
+        }

app/routers/api_crud_v3.py

@@ -432,8 +432,9 @@ async def post_obj(
                 return mk_resp(data=resp_data, response=response)
         return mk_resp(data={"obj_id": new_obj_id, "obj_id_random": new_obj_id_random}, response=response)
     else:
+        # Standardized rich error bubbling
         db_err = format_db_error(get_last_sql_error())
-        return mk_resp(data=False, status_code=400, response=response, status_message="Failed to create object.", details=db_err)
+        return mk_resp(data=False, status_code=400, response=response, status_message="Failed to create object.", details=db_err.dict())
 
 
 @router.patch('/{obj_type_l1}/{obj_id}', response_model=Resp_Body_Base)
@@ -494,8 +495,9 @@ async def patch_obj(
                 return mk_resp(data=resp_data, response=response)
         return mk_resp(data=True, response=response, status_message="Object updated successfully.")
     else:
+        # Standardized rich error bubbling
         db_err = format_db_error(get_last_sql_error())
-        return mk_resp(data=False, status_code=400, response=response, status_message="Failed to update object.", details=db_err)
+        return mk_resp(data=False, status_code=400, response=response, status_message="Failed to update object.", details=db_err.dict())
 
 
 @router.delete('/{obj_type_l1}/{obj_id}', response_model=Resp_Body_Base)

app/routers/api_crud_v3_nested.py

@@ -204,8 +204,9 @@ async def post_child_obj(
                 return mk_resp(data=resp_data, response=response)
         return mk_resp(data={"obj_id": new_obj_id, "obj_id_random": new_obj_id_random}, response=response)
     else:
+        # Standardized rich error bubbling
         db_err = format_db_error(get_last_sql_error())
-        return mk_resp(data=False, status_code=400, response=response, status_message="Failed to create child object.", details=db_err)
+        return mk_resp(data=False, status_code=400, response=response, status_message="Failed to create child object.", details=db_err.dict())
 
 
 @router.get('/{parent_obj_type}/{parent_obj_id}/{child_obj_type}/{child_obj_id}', response_model=Resp_Body_Base)