added custom exception for DUPLICATES_DETECTED error

src/aiosalesforce/__init__.py

@@ -1,4 +1,4 @@
-__version__ = "0.6.1"
+__version__ = "0.6.2"
 
 __all__ = [
     "ClientCredentialsFlow",

src/aiosalesforce/exceptions.py

@@ -1,4 +1,4 @@
-from typing import NoReturn
+from typing import NoReturn, TypedDict
 
 from httpx import Response
 
@@ -33,6 +33,34 @@ class MoreThanOneRecordError(SalesforceError):
     """Raised when more than one record is found by external ID."""
 
 
+class _DuplicateRecord(TypedDict):
+    id: str
+    sobject: str
+    match_confidence: float
+    match_engine: str
+    rule: str
+
+
+class DuplicatesDetectedError(SalesforceError):
+    """Raised when duplicates are detected, e.g. when creating a new record."""
+
+    @property
+    def duplicates(self) -> list[_DuplicateRecord]:
+        assert self.response is not None
+        response_json = json_loads(self.response.content)
+        return [
+            _DuplicateRecord(
+                id=match_record["record"]["Id"],
+                sobject=match_record["record"]["attributes"]["type"],
+                match_confidence=match_record["matchConfidence"],
+                match_engine=match_result["matchEngine"],
+                rule=match_result["rule"],
+            )
+            for match_result in response_json[0]["duplicateResult"]["matchResults"]
+            for match_record in match_result["matchRecords"]
+        ]
+
+
 class AuthenticationError(SalesforceError):
     """Raised when authentication fails."""
 
@@ -73,6 +101,24 @@ def raise_salesforce_error(response: Response) -> NoReturn:
         response_json = json_loads(response.content)
         error_code = response_json[0]["errorCode"]
         error_message = response_json[0]["message"]
+        if error_code == "DUPLICATES_DETECTED":
+            error_message = "\n".join(
+                [
+                    error_message,
+                    *[
+                        "  %s (%s, %s, %s)"
+                        % (
+                            duplicate["id"],
+                            f"{duplicate['match_confidence']:.1f}%",
+                            duplicate["match_engine"],
+                            duplicate["rule"],
+                        )
+                        for duplicate in DuplicatesDetectedError(
+                            "", response
+                        ).duplicates
+                    ],
+                ]
+            )
     except Exception:
         error_code = None
         error_message = response.text
@@ -95,6 +141,8 @@ def raise_salesforce_error(response: Response) -> NoReturn:
                 )
         case (_, "REQUEST_LIMIT_EXCEEDED"):
             exc_class = RequestLimitExceededError
+        case (400, "DUPLICATES_DETECTED"):
+            exc_class = DuplicatesDetectedError
         case (403, _):
             exc_class = AuthorizationError
         case (404, _):

tests/test_exceptions.py

@@ -1,8 +1,11 @@
+import re
+
 import orjson
 import pytest
 
 from aiosalesforce.exceptions import (
     AuthorizationError,
+    DuplicatesDetectedError,
     MoreThanOneRecordError,
     NotFoundError,
     RequestLimitExceededError,
@@ -35,6 +38,64 @@ def test_more_than_one_record_error():
         )
 
 
+def test_duplicates_detected_error():
+    with pytest.raises(
+        DuplicatesDetectedError,
+        match=re.compile(
+            (
+                r"^\[DUPLICATES_DETECTED\] Use one of these records\?$\n"
+                r"^  003aj000002oTldAAE "
+                r"\(100.0%, FuzzyMatchEngine, Standard_Contact_Match_Rule_v1_1\)"
+            ),
+            flags=re.MULTILINE,
+        ),
+    ):
+        raise_salesforce_error(
+            Response(
+                400,
+                content=orjson.dumps(
+                    [
+                        {
+                            "duplicateResult": {
+                                "allowSave": True,
+                                "duplicateRule": "Standard_Contact_Duplicate_Rule",
+                                "duplicateRuleEntityType": "Contact",
+                                "errorMessage": "Use one of these records?",
+                                "matchResults": [
+                                    {
+                                        "entityType": "Contact",
+                                        "errors": [],
+                                        "matchEngine": "FuzzyMatchEngine",
+                                        "matchRecords": [
+                                            {
+                                                "additionalInformation": [],
+                                                "fieldDiffs": [],
+                                                "matchConfidence": 100.0,
+                                                "record": {
+                                                    "attributes": {
+                                                        "type": "Contact",
+                                                        "url": "...",
+                                                    },
+                                                    "Id": "003aj000002oTldAAE",
+                                                },
+                                            }
+                                        ],
+                                        "rule": "Standard_Contact_Match_Rule_v1_1",
+                                        "size": 1,
+                                        "success": True,
+                                    }
+                                ],
+                            },
+                            "errorCode": "DUPLICATES_DETECTED",
+                            "message": "Use one of these records?",
+                        }
+                    ]
+                ),
+                request=Request("POST", "https://example.com/Contact"),
+            )
+        )
+
+
 def test_more_than_one_record_error_bad_response():
     with pytest.raises(
         MoreThanOneRecordError,