feat: implement batch upsert operation and related methods

Author: Samson Gebre

src/PowerPlatform/Dataverse/data/_batch.py

@@ -21,6 +21,7 @@
     CascadeConfiguration,
 )
 from ..models.labels import Label, LocalizedLabel
+from ..models.upsert import UpsertItem
 from ..common.constants import CASCADE_BEHAVIOR_REMOVE_LINK
 from ._raw_request import _RawRequest
 
@@ -71,6 +72,12 @@ class _RecordGet:
     select: Optional[List[str]] = None
 
 
+@dataclass
+class _RecordUpsert:
+    table: str
+    items: List[UpsertItem]  # always non-empty; normalised by BatchRecordOperations
+
+
 # --- Table intent types ---
 
 
@@ -287,6 +294,8 @@ def _resolve_item(self, item: Any) -> List[_RawRequest]:
             return self._resolve_record_delete(item)
         if isinstance(item, _RecordGet):
             return self._resolve_record_get(item)
+        if isinstance(item, _RecordUpsert):
+            return self._resolve_record_upsert(item)
         if isinstance(item, _TableCreate):
             return self._resolve_table_create(item)
         if isinstance(item, _TableDelete):
@@ -361,6 +370,15 @@ def _resolve_record_delete(self, op: _RecordDelete) -> List[_RawRequest]:
     def _resolve_record_get(self, op: _RecordGet) -> List[_RawRequest]:
         return [self._od._build_get(op.table, op.record_id, select=op.select)]
 
+    def _resolve_record_upsert(self, op: _RecordUpsert) -> List[_RawRequest]:
+        entity_set = self._od._entity_set_from_schema_name(op.table)
+        if len(op.items) == 1:
+            item = op.items[0]
+            return [self._od._build_upsert(entity_set, op.table, item.alternate_key, item.record)]
+        alternate_keys = [i.alternate_key for i in op.items]
+        records = [i.record for i in op.items]
+        return [self._od._build_upsert_multiple(entity_set, op.table, alternate_keys, records)]
+
     # ------------------------------------------------------------------
     # Table resolvers — delegate to _ODataClient._build_* methods
     # (pre-resolution GETs for MetadataId remain here; they are batch-

src/PowerPlatform/Dataverse/data/_odata.py

@@ -1704,6 +1704,60 @@ def _build_update_multiple(
             raise ValidationError("changes must be a dict or list[dict].", subcode="invalid_changes_type")
         return self._build_update_multiple_from_records(entity_set, table, records)
 
+    def _build_upsert(
+        self,
+        entity_set: str,
+        table: str,
+        alternate_key: Dict[str, Any],
+        record: Dict[str, Any],
+    ) -> _RawRequest:
+        """Build a single-record PATCH upsert request without sending it.
+
+        Unlike :meth:`_build_update`, no ``If-Match: *`` header is added so the
+        server creates the record when it does not yet exist.
+        """
+        body = self._lowercase_keys(record)
+        body = self._convert_labels_to_ints(table, body)
+        key_str = self._build_alternate_key_str(alternate_key)
+        url = f"{self.api}/{entity_set}({key_str})"
+        return _RawRequest(
+            method="PATCH",
+            url=url,
+            body=json.dumps(body, ensure_ascii=False),
+        )
+
+    def _build_upsert_multiple(
+        self,
+        entity_set: str,
+        table: str,
+        alternate_keys: List[Dict[str, Any]],
+        records: List[Dict[str, Any]],
+    ) -> _RawRequest:
+        """Build an UpsertMultiple POST request without sending it."""
+        if len(alternate_keys) != len(records):
+            raise ValidationError(
+                f"alternate_keys and records must have the same length "
+                f"({len(alternate_keys)} != {len(records)})",
+                subcode="upsert_length_mismatch",
+            )
+        logical_name = table.lower()
+        targets: List[Dict[str, Any]] = []
+        for alt_key, record in zip(alternate_keys, records):
+            alt_key_lower = self._lowercase_keys(alt_key)
+            record_processed = self._lowercase_keys(record)
+            record_processed = self._convert_labels_to_ints(table, record_processed)
+            combined: Dict[str, Any] = {**alt_key_lower, **record_processed}
+            if "@odata.type" not in combined:
+                combined["@odata.type"] = f"Microsoft.Dynamics.CRM.{logical_name}"
+            key_str = self._build_alternate_key_str(alt_key)
+            combined["@odata.id"] = f"{entity_set}({key_str})"
+            targets.append(combined)
+        return _RawRequest(
+            method="POST",
+            url=f"{self.api}/{entity_set}/Microsoft.Dynamics.CRM.UpsertMultiple",
+            body=json.dumps({"Targets": targets}, ensure_ascii=False),
+        )
+
     def _build_delete(
         self,
         table: str,

src/PowerPlatform/Dataverse/operations/batch.py

@@ -15,6 +15,7 @@
     _RecordUpdate,
     _RecordDelete,
     _RecordGet,
+    _RecordUpsert,
     _TableCreate,
     _TableDelete,
     _TableGet,
@@ -29,6 +30,7 @@
     _QuerySql,
 )
 from ..models.batch import BatchResult
+from ..models.upsert import UpsertItem
 from ..models.relationship import (
     LookupAttributeMetadata,
     OneToManyRelationshipMetadata,
@@ -241,6 +243,59 @@ def get(
         """
         self._batch._items.append(_RecordGet(table=table, record_id=record_id, select=select))
 
+    def upsert(
+        self,
+        table: str,
+        items: List[Union[UpsertItem, Dict[str, Any]]],
+    ) -> None:
+        """
+        Add an upsert operation to the batch.
+
+        Mirrors :meth:`~PowerPlatform.Dataverse.operations.records.RecordOperations.upsert`
+        exactly: a single item becomes a PATCH request using the alternate key; multiple
+        items become one ``UpsertMultiple`` POST.
+
+        Each item must be a :class:`~PowerPlatform.Dataverse.models.upsert.UpsertItem`
+        or a plain ``dict`` with ``"alternate_key"`` and ``"record"`` keys (both dicts).
+
+        :param table: Table schema name (e.g. ``"account"``).
+        :param items: Non-empty list of :class:`~PowerPlatform.Dataverse.models.upsert.UpsertItem`
+            instances or equivalent dicts.
+
+        Example::
+
+            from PowerPlatform.Dataverse.models.upsert import UpsertItem
+
+            batch.records.upsert("account", [
+                UpsertItem(
+                    alternate_key={"accountnumber": "ACC-001"},
+                    record={"name": "Contoso Ltd"},
+                ),
+                UpsertItem(
+                    alternate_key={"accountnumber": "ACC-002"},
+                    record={"name": "Fabrikam Inc"},
+                ),
+            ])
+        """
+        if not isinstance(items, list) or not items:
+            raise ValidationError("items must be a non-empty list", subcode="VALIDATION_UPSERT_EMPTY")
+        normalized: List[UpsertItem] = []
+        for i in items:
+            if isinstance(i, UpsertItem):
+                normalized.append(i)
+            elif (
+                isinstance(i, dict)
+                and isinstance(i.get("alternate_key"), dict)
+                and isinstance(i.get("record"), dict)
+            ):
+                normalized.append(UpsertItem(alternate_key=i["alternate_key"], record=i["record"]))
+            else:
+                raise ValidationError(
+                    "Each upsert item must be a UpsertItem or a dict with 'alternate_key' and 'record' keys.",
+                    subcode="VALIDATION_UPSERT_ITEM",
+                )
+        self._batch._items.append(_RecordUpsert(table=table, items=normalized))
+
 
 class BatchTableOperations:
     """

tests/unit/data/test_batch_serialization.py

@@ -14,6 +14,7 @@
     _RecordCreate,
     _RecordDelete,
     _RecordGet,
+    _RecordUpsert,
     _TableGet,
     _TableList,
     _QuerySql,
@@ -23,6 +24,7 @@
     _parse_http_response_part,
     _CRLF,
 )
+from PowerPlatform.Dataverse.models.upsert import UpsertItem
 from PowerPlatform.Dataverse.data._raw_request import _RawRequest
 from PowerPlatform.Dataverse.models.batch import BatchItemResponse, BatchResult
 
@@ -337,5 +339,124 @@ def test_operations_in_order(self):
         self.assertIsInstance(cs.operations[1], _RecordDelete)
 
 
+class TestResolveBatchUpsert(unittest.TestCase):
+    """Tests that _BatchClient._resolve_record_upsert calls the correct _build_* methods."""
+
+    def _client_and_od(self):
+        od = _make_od()
+        od._entity_set_from_schema_name.return_value = "accounts"
+        client = _BatchClient(od)
+        return client, od
+
+    def test_resolve_single_item_calls_build_upsert(self):
+        client, od = self._client_and_od()
+        mock_req = MagicMock()
+        od._build_upsert.return_value = mock_req
+
+        item = UpsertItem(alternate_key={"accountnumber": "ACC-001"}, record={"name": "Contoso"})
+        op = _RecordUpsert(table="account", items=[item])
+        result = client._resolve_record_upsert(op)
+
+        od._build_upsert.assert_called_once_with(
+            "accounts", "account", {"accountnumber": "ACC-001"}, {"name": "Contoso"}
+        )
+        self.assertEqual(result, [mock_req])
+
+    def test_resolve_multiple_items_calls_build_upsert_multiple(self):
+        client, od = self._client_and_od()
+        mock_req = MagicMock()
+        od._build_upsert_multiple.return_value = mock_req
+
+        items = [
+            UpsertItem(alternate_key={"accountnumber": "ACC-001"}, record={"name": "Contoso"}),
+            UpsertItem(alternate_key={"accountnumber": "ACC-002"}, record={"name": "Fabrikam"}),
+        ]
+        op = _RecordUpsert(table="account", items=items)
+        result = client._resolve_record_upsert(op)
+
+        od._build_upsert_multiple.assert_called_once_with(
+            "accounts",
+            "account",
+            [{"accountnumber": "ACC-001"}, {"accountnumber": "ACC-002"}],
+            [{"name": "Contoso"}, {"name": "Fabrikam"}],
+        )
+        self.assertEqual(result, [mock_req])
+
+    def test_resolve_item_dispatch_routes_to_upsert(self):
+        client, od = self._client_and_od()
+        mock_req = MagicMock()
+        od._build_upsert.return_value = mock_req
+
+        item = UpsertItem(alternate_key={"accountnumber": "ACC-001"}, record={"name": "Contoso"})
+        op = _RecordUpsert(table="account", items=[item])
+        result = client._resolve_item(op)
+
+        self.assertEqual(result, [mock_req])
+
+
+class TestBatchRecordOperationsUpsert(unittest.TestCase):
+    """Tests for BatchRecordOperations.upsert (operations/batch.py)."""
+
+    def _make_batch(self):
+        from PowerPlatform.Dataverse.operations.batch import BatchRecordOperations
+
+        batch = MagicMock()
+        batch._items = []
+        return BatchRecordOperations(batch), batch
+
+    def test_upsert_single_upsert_item_appended(self):
+        from PowerPlatform.Dataverse.operations.batch import BatchRecordOperations
+
+        rec_ops, batch = self._make_batch()
+        item = UpsertItem(alternate_key={"accountnumber": "ACC-001"}, record={"name": "Contoso"})
+        rec_ops.upsert("account", [item])
+
+        self.assertEqual(len(batch._items), 1)
+        intent = batch._items[0]
+        self.assertIsInstance(intent, _RecordUpsert)
+        self.assertEqual(intent.table, "account")
+        self.assertEqual(len(intent.items), 1)
+        self.assertEqual(intent.items[0].alternate_key, {"accountnumber": "ACC-001"})
+
+    def test_upsert_plain_dict_normalised_to_upsert_item(self):
+        from PowerPlatform.Dataverse.operations.batch import BatchRecordOperations
+
+        rec_ops, batch = self._make_batch()
+        rec_ops.upsert("account", [{"alternate_key": {"accountnumber": "X"}, "record": {"name": "Y"}}])
+
+        intent = batch._items[0]
+        self.assertIsInstance(intent.items[0], UpsertItem)
+        self.assertEqual(intent.items[0].record, {"name": "Y"})
+
+    def test_upsert_empty_list_raises(self):
+        from PowerPlatform.Dataverse.operations.batch import BatchRecordOperations
+        from PowerPlatform.Dataverse.core.errors import ValidationError
+
+        rec_ops, _ = self._make_batch()
+        with self.assertRaises(ValidationError):
+            rec_ops.upsert("account", [])
+
+    def test_upsert_invalid_item_raises(self):
+        from PowerPlatform.Dataverse.operations.batch import BatchRecordOperations
+        from PowerPlatform.Dataverse.core.errors import ValidationError
+
+        rec_ops, _ = self._make_batch()
+        with self.assertRaises(ValidationError):
+            rec_ops.upsert("account", ["not_a_valid_item"])
+
+    def test_upsert_multiple_items_all_normalised(self):
+        from PowerPlatform.Dataverse.operations.batch import BatchRecordOperations
+
+        rec_ops, batch = self._make_batch()
+        rec_ops.upsert("account", [
+            UpsertItem(alternate_key={"accountnumber": "A"}, record={"name": "Alpha"}),
+            UpsertItem(alternate_key={"accountnumber": "B"}, record={"name": "Beta"}),
+        ])
+
+        intent = batch._items[0]
+        self.assertEqual(len(intent.items), 2)
+        self.assertEqual(intent.items[1].alternate_key, {"accountnumber": "B"})
+
+
 if __name__ == "__main__":
     unittest.main()