Add batch() method to CustomerIO track client

customerio/track.py

@@ -234,6 +234,21 @@ def merge_customers(self, primary_id_type, primary_id, secondary_id_type, second
         }
         self.send_request("POST", url, post_data)
 
+    def batch(self, operations):
+        """Send multiple operations in a single request.
+
+        Each operation is a dict with at minimum 'type' and 'action' keys.
+        See https://customer.io/docs/api/track/#operation/batch
+        """
+        if not operations:
+            raise CustomerIOException("operations cannot be empty in batch")
+
+        if self.port == 443:
+            url = f"https://{self.host}/api/v2/batch"
+        else:
+            url = f"https://{self.host}:{self.port}/api/v2/batch"
+        self.send_request("POST", url, {"batch": operations})
+
     def _build_session(self):
         session = super()._build_session()
         session.auth = (self.site_id, self.api_key)

tests/test_customerio.py

@@ -469,6 +469,58 @@ def test_merge_customers_call(self):
                 secondary_id="",
             )
 
+    def test_batch_call(self):
+        self.cio.http.hooks = dict(
+            response=partial(
+                self._check_request,
+                rq={
+                    "method": "POST",
+                    "authorization": _basic_auth_str("siteid", "apikey"),
+                    "content_type": "application/json",
+                    "url_suffix": "/api/v2/batch",
+                    "body": {
+                        "batch": [
+                            {
+                                "type": "person",
+                                "action": "identify",
+                                "identifiers": {"id": "1"},
+                                "attributes": {"name": "Alice"},
+                            },
+                            {
+                                "type": "person",
+                                "action": "event",
+                                "identifiers": {"id": "1"},
+                                "name": "purchase",
+                            },
+                        ]
+                    },
+                },
+            )
+        )
+
+        self.cio.batch(
+            [
+                {
+                    "type": "person",
+                    "action": "identify",
+                    "identifiers": {"id": "1"},
+                    "attributes": {"name": "Alice"},
+                },
+                {
+                    "type": "person",
+                    "action": "event",
+                    "identifiers": {"id": "1"},
+                    "name": "purchase",
+                },
+            ]
+        )
+
+        with self.assertRaises(CustomerIOException):
+            self.cio.batch([])
+
+        with self.assertRaises(CustomerIOException):
+            self.cio.batch(None)
+
 
 if __name__ == "__main__":
     unittest.main()