Enhance V3 CRUD: Implement Error Bubbling and Dry-Run Validation.

- Updated app/db_sql.py to capture SQL exceptions in thread-local storage for later retrieval.
- Implemented format_db_error() in app/lib_api_crud_v3.py to clean up raw MariaDB error strings.
- Added POST /v3/crud/{obj_type}/validate endpoint for dry-run payload validation.
- Updated main and nested routers to bubble up validation and database errors into the response 'meta.details' field.
- Added tests/test_v3_error_bubbling.py to verify formatting logic.
This commit is contained in:
Scott Idem
2026-01-09 16:57:54 -05:00
parent 3885cc6aba
commit 4b86432381
5 changed files with 122 additions and 10 deletions

View File

@@ -1,4 +1,4 @@
import datetime, json, pytz, random, redis, secrets
import datetime, json, pytz, random, redis, secrets, threading
from typing import Any, List, Optional
from timeit import default_timer as timer
@@ -10,6 +10,19 @@ from sqlalchemy import create_engine, text, Time
from sqlalchemy.exc import IntegrityError, OperationalError, ProgrammingError
from sqlalchemy.pool import NullPool
# Thread-local storage for capturing last SQL error message
_sql_error_state = threading.local()
def get_last_sql_error() -> Optional[str]:
"""Retrieves and clears the last captured SQL error message."""
error = getattr(_sql_error_state, 'last_error', None)
_sql_error_state.last_error = None
return error
def set_last_sql_error(error: Any):
"""Sets the last captured SQL error message."""
_sql_error_state.last_error = str(error)
from app.lib_sql_search import (
sql_limit_offset_part as _sql_limit_offset_part,
sql_and_like_part as _sql_and_like_part,
@@ -114,11 +127,13 @@ def sql_insert(
trans.rollback()
log.error('Integrity error (likely duplicate). Returning None')
log.debug(e)
set_last_sql_error(e)
return None
except Exception as e:
trans.rollback()
log.error('Unknown exception in sql_insert. Returning False')
log.exception(e)
set_last_sql_error(e)
return False
else:
if result_insert.rowcount == 1 and result_insert.lastrowid > 0:
@@ -186,11 +201,13 @@ def sql_update(
try:
result_update = db.execute(sql_update_stmt, data)
trans.commit()
except Exception:
except Exception as e:
set_last_sql_error(e)
return False
except Exception as e:
trans.rollback()
log.exception(e)
set_last_sql_error(e)
return False
else:
if result_update.rowcount >= 1: