Preserve Workbench business rule payloads for clearer validation errors

specifyweb/backend/workbench/upload/tests/test_upload_results_json.py

@@ -4,6 +4,8 @@
 import unittest
 from jsonschema import validate, Draft7Validator # type: ignore
 
+from specifyweb.backend.businessrules.exceptions import BusinessRuleException
+
 from ..upload_result import *
 from ..upload_results_schema import schema
 
@@ -36,6 +38,114 @@ def testFailedBusinessRule(self, failedBusinessRule: FailedBusinessRule):
         j = json.dumps(failedBusinessRule.to_json())
         self.assertEqual(failedBusinessRule, FailedBusinessRule.from_json(json.loads(j)))
 
+    def testBusinessRuleExceptionPayload(self):
+        info = ReportInfo(
+            tableName="Collectionobject",
+            columns=["catalogNumber"],
+            treeInfo=None,
+        )
+        payload = {
+            "localizationKey": "childFieldNotUnique",
+            "table": "Collectionobject",
+            "fieldName": "catalognumber",
+            "fieldData": {"catalognumber": "0037481"},
+            "parentField": "collection",
+            "parentData": {"collection": "Collection object (360449)"},
+            "conflicting": [3347460],
+        }
+
+        self.assertEqual(
+            to_failed_business_rule(
+                BusinessRuleException(
+                    "Collectionobject must have unique catalognumber in collection",
+                    payload,
+                ),
+                info,
+            ),
+            FailedBusinessRule(
+                "Collectionobject must have unique catalognumber in collection",
+                payload,
+                info,
+            ),
+        )
+
+    def testBusinessRuleExceptionPayloadSanitization(self):
+        info = ReportInfo(
+            tableName="Collectionobject",
+            columns=["catalogNumber"],
+            treeInfo=None,
+        )
+
+        payload = {
+            "localizationKey": "childFieldNotUnique",
+            "table": "Collectionobject",
+            "fieldName": "catalognumber",
+            "goodNested": {"a": "b", "n": 1, "ok": True, "null": None},
+            "badNested": {"bad": info},
+            "goodList": [1, 2, 3],
+            "badList": [1, info],
+        }
+
+        failed_business_rule = to_failed_business_rule(
+            Exception(
+                "Collectionobject must have unique catalognumber in collection",
+                payload,
+            ),
+            info,
+        )
+
+        self.assertEqual(
+            failed_business_rule.payload,
+            {
+                "localizationKey": "childFieldNotUnique",
+                "table": "Collectionobject",
+                "fieldName": "catalognumber",
+                "goodNested": {"a": "b", "n": 1, "ok": True, "null": None},
+                "goodList": [1, 2, 3],
+            },
+        )
+
+        # Ensure sanitized payload always serializes in upload results.
+        json.dumps(failed_business_rule.to_json())
+
+    def testWrapperFallbackDoesNotMatchGenericTwoArgException(self):
+        info = ReportInfo(
+            tableName="Collectionobject",
+            columns=["catalogNumber"],
+            treeInfo=None,
+        )
+
+        exception = Exception(
+            "connection failed",
+            {"table": "Collectionobject", "reason": "timeout"},
+        )
+        failed_business_rule = to_failed_business_rule(exception, info)
+
+        self.assertEqual(failed_business_rule.payload, {})
+        self.assertEqual(failed_business_rule.message, str(exception))
+
+    def testBusinessRulePayloadPreservesTopLevelNone(self):
+        info = ReportInfo(
+            tableName="Collectionobject",
+            columns=["catalogNumber"],
+            treeInfo=None,
+        )
+
+        payload = {
+            "localizationKey": "childFieldNotUnique",
+            "table": "Collectionobject",
+            "fieldName": None,
+            "parentField": "collection",
+        }
+
+        failed_business_rule = to_failed_business_rule(
+            Exception("Business rule failed", payload),
+            info,
+        )
+
+        self.assertIn("fieldName", failed_business_rule.payload)
+        self.assertIsNone(failed_business_rule.payload["fieldName"])
+
     @given(noMatch=infer)
     def testNoMatch(self, noMatch: NoMatch):
         j = json.dumps(noMatch.to_json())

specifyweb/backend/workbench/upload/treerecord.py

@@ -47,6 +47,7 @@
     FailedBusinessRule,
     ReportInfo,
     TreeInfo,
+    to_failed_business_rule,
 )
 from .uploadable import (
     Row,
@@ -954,7 +955,7 @@ def _upload(
                         obj = self._do_insert(model, **new_attrs)
                 except (BusinessRuleException, IntegrityError) as e:
                     return UploadResult(
-                        FailedBusinessRule(str(e), {}, info), parent_result, {}
+                        to_failed_business_rule(e, info), parent_result, {}
                     )
 
             result = UploadResult(Uploaded(obj.id, info, []), parent_result, {})

specifyweb/backend/workbench/upload/upload_result.py

@@ -1,10 +1,21 @@
-from typing import Any, NamedTuple
+from typing import Any, NamedTuple, cast
 
 from typing import Literal
 
 from .parsing import WorkBenchParseFailure
 
 Failure = Literal["Failure"]
+BUSINESS_RULE_EXCEPTION_MODULE = "specifyweb.backend.businessrules.exceptions"
+BUSINESS_RULE_EXCEPTION_NAME = "BusinessRuleException"
+BusinessRulePayloadValue = (
+    str
+    | int
+    | bool
+    | None
+    | list[str | int | bool | None]
+    | dict[str, str | int | bool | None]
+)
+BusinessRulePayload = dict[str, BusinessRulePayloadValue]
 
 
 class TreeInfo(NamedTuple):
@@ -215,7 +226,7 @@ def from_json(json: dict) -> "Deleted":
 
 class FailedBusinessRule(NamedTuple):
     message: str
-    payload: dict[str, str | int | list[str] | list[int]]
+    payload: BusinessRulePayload
     info: ReportInfo
 
     def get_id(self) -> Failure:
@@ -238,6 +249,79 @@ def from_json(json: dict) -> "FailedBusinessRule":
         )
 
 
+def is_business_rule_exception_with_payload(exception: Exception) -> bool:
+    exception_class = exception.__class__
+    payload_like_exception = (
+        len(exception.args) >= 2
+        and isinstance(exception.args[0], str)
+        and isinstance(exception.args[1], dict)
+    )
+
+    if not payload_like_exception:
+        return False
+
+    # Some wrapped code paths can preserve a business-rule payload without
+    # preserving the original exception class identity.
+    has_business_rule_shape = isinstance(
+        exception.args[1].get("localizationKey"), str
+    )
+
+    return (
+        (
+            exception_class.__module__ == BUSINESS_RULE_EXCEPTION_MODULE
+            and exception_class.__name__ == BUSINESS_RULE_EXCEPTION_NAME
+        )
+        or has_business_rule_shape
+    )
+
+
+def _is_business_rule_scalar(value: Any) -> bool:
+    return isinstance(value, (str, int, bool)) or value is None
+
+
+_SANITIZE_FAILED = object()
+
+
+def _sanitize_business_rule_payload_value(value: Any) -> BusinessRulePayloadValue | object:
+    if _is_business_rule_scalar(value):
+        return value
+
+    if isinstance(value, list):
+        if all(_is_business_rule_scalar(item) for item in value):
+            return value
+        return _SANITIZE_FAILED
+
+    if isinstance(value, dict):
+        sanitized: dict[str, str | int | bool | None] = {}
+        for key, item in value.items():
+            if not isinstance(key, str) or not _is_business_rule_scalar(item):
+                return _SANITIZE_FAILED
+            sanitized[key] = item
+        return sanitized
+
+    return _SANITIZE_FAILED
+
+
+def _sanitize_business_rule_payload(payload: dict[Any, Any]) -> BusinessRulePayload:
+    sanitized: BusinessRulePayload = {}
+    for key, value in payload.items():
+        if not isinstance(key, str):
+            continue
+        sanitized_value = _sanitize_business_rule_payload_value(value)
+        if sanitized_value is _SANITIZE_FAILED:
+            continue
+        sanitized[key] = cast(BusinessRulePayloadValue, sanitized_value)
+    return sanitized
+
+
+def to_failed_business_rule(exception: Exception, info: ReportInfo) -> FailedBusinessRule:
+    if is_business_rule_exception_with_payload(exception):
+        payload = _sanitize_business_rule_payload(exception.args[1])
+        return FailedBusinessRule(exception.args[0], payload, info)
+
+    return FailedBusinessRule(str(exception), {}, info)
+
+
 class NoMatch(NamedTuple):
     info: ReportInfo
 

specifyweb/backend/workbench/upload/upload_table.py

@@ -39,6 +39,7 @@
     PicklistAddition,
     ParseFailures,
     PropagatedFailure,
+    to_failed_business_rule,
 )
 from .uploadable import (
     NULL_RECORD,
@@ -591,7 +592,7 @@ def _handle_row(self, skip_match: bool, allow_null: bool) -> UploadResult:
         except ContetRef as e:
             # Not sure if there is a better way for this. Consider moving this to binding.
             return UploadResult(
-                FailedBusinessRule(str(e), {}, info), to_one_results, {}
+                to_failed_business_rule(e, info), to_one_results, {}
             )
 
         attrs = {
@@ -760,7 +761,7 @@ def _do_upload(
                 picklist_additions = self._do_picklist_additions()
             except (BusinessRuleException, IntegrityError) as e:
                 return UploadResult(
-                    FailedBusinessRule(str(e), {}, info), to_one_results, {}
+                    to_failed_business_rule(e, info), to_one_results, {}
                 )
 
         record = Uploaded(uploaded.id, info, picklist_additions)
@@ -865,7 +866,7 @@ def delete_row(self, parent_obj=None) -> UploadResult:
                 reference_record.delete()
                 result = Deleted(self.current_id, info)
             except (BusinessRuleException, IntegrityError) as e:
-                result = FailedBusinessRule(str(e), {}, info)
+                result = to_failed_business_rule(e, info)
 
         to_one_deleted: dict[str, UploadResult] = {
             key: value.delete_row()
@@ -1066,7 +1067,7 @@ def _do_upload(
                     picklist_additions = self._do_picklist_additions()
                 except (BusinessRuleException, IntegrityError) as e:
                     return UploadResult(
-                        FailedBusinessRule(str(e), {}, info), to_one_results, {}
+                        to_failed_business_rule(e, info), to_one_results, {}
                     )
 
         record: Updated | NoChange = (

specifyweb/frontend/js_src/lib/components/LocalityUpdate/utils.ts

@@ -6,8 +6,8 @@ import { f } from '../../utils/functools';
 import type { IR, RA, RR } from '../../utils/types';
 import { tables } from '../DataModel/tables';
 import type { Tables } from '../DataModel/types';
-import { resolveBackendParsingMessage } from '../WorkBench/resultsParser';
 import type { LocalityUpdateHeader, LocalityUpdateTaskStatus } from './types';
+import { resolveBackendParsingMessage } from '../WorkBench/resultMessageResolvers';
 
 const localityUpdateAcceptedLocalityFields: RA<
   Lowercase<keyof Tables['Locality']['fields']>