Add batch message type.

pyleco/json_utils/json_objects.py

@@ -32,8 +32,7 @@
 from __future__ import annotations
 from dataclasses import asdict, dataclass
 import json
-from typing import Any, Optional, Union
-
+from typing import Any, Generic, Iterable, Optional, TypeVar, Union
 
 ErrorType = Union["DataError", "Error"]
 NotificationType = Union["Notification", "ParamsNotification"]
@@ -130,3 +129,31 @@ def model_dump(self) -> dict[str, Any]:
         pre_dict = asdict(self)
         pre_dict["error"] = asdict(self.error)
         return pre_dict
+
+"""
+Batch Handling.
+
+Not included in jsonrpc2-objects, but defined by JSONRPC 2.0
+"""
+BatchType = TypeVar("BatchType", RequestType, ResponseType)
+
+
+class BatchObject(list, Generic[BatchType]):
+    """A batch of requests or responses."""
+    # Not defined by jsonrpc2-objects
+
+    def __init__(self, iterable: Optional[Iterable[BatchType]] = None):
+        if iterable:
+            super().__init__(item for item in iterable)
+        else:
+            super().__init__()
+
+    def model_dump(self) -> list[dict[str, Any]]:
+        return [obj.model_dump() for obj in self]
+
+    def model_dump_json(self) -> str:
+        return json.dumps(self.model_dump(), separators=(",", ":"))
+
+
+RequestBatch = BatchObject[RequestType]
+ResponseBatch = BatchObject[ResponseType]

tests/json_utils/test_json_objects.py

@@ -22,6 +22,8 @@
 # THE SOFTWARE.
 #
 
+import pytest
+
 from pyleco.json_utils import json_objects
 
 
@@ -83,3 +85,35 @@ def test_generate_data_error_from_error():
     assert data_error.code == error.code
     assert data_error.message == error.message
     assert data_error.data == "data"
+
+
+class Test_BatchObject:
+    element = json_objects.Request(5, "start")
+
+    @pytest.fixture
+    def batch_obj(self):
+        return json_objects.BatchObject([self.element])
+
+    def test_init_with_values(self):
+        obj = json_objects.BatchObject([self.element])
+        assert obj == [self.element]
+
+    def test_bool_value_with_element(self):
+        obj = json_objects.BatchObject([self.element])
+        assert bool(obj) is True
+
+    def test_bool_value_without_element(self):
+        obj = json_objects.BatchObject()
+        assert bool(obj) is False
+
+    def test_append(self, batch_obj: json_objects.BatchObject):
+        el2 = json_objects.Request(5, "start")
+        batch_obj.append(el2)
+        assert batch_obj[-1] == el2
+
+    def test_model_dump(self, batch_obj: json_objects.BatchObject):
+        assert batch_obj.model_dump() == [self.element.model_dump()]
+
+    def test_model_dump_json(self, batch_obj: json_objects.BatchObject):
+        result = '[{"id":5,"method":"start","jsonrpc":"2.0"}]'
+        assert batch_obj.model_dump_json() == result