feat(core): add JSON Batch format support

src/cloudevents/core/bindings/http.py

@@ -26,7 +26,7 @@
     TIME_ATTR,
     get_event_factory_for_version,
 )
-from cloudevents.core.formats.base import Format
+from cloudevents.core.formats.base import BatchFormat, Format
 from cloudevents.core.formats.json import JSONFormat
 from cloudevents.core.spec import SPECVERSION_V1_0
 
@@ -260,6 +260,42 @@ def from_structured(
     return event_format.read(event_factory, message.body)
 
 
+def to_batch(events: list[BaseCloudEvent], event_format: BatchFormat) -> HTTPMessage:
+    """
+    Convert a list of CloudEvents to an HTTP batched content mode message.
+
+    The entire batch is serialized into the HTTP body as a JSON array and the
+    Content-Type header is set to the format's batch media type.
+
+    :param events: The CloudEvents to convert
+    :param event_format: Batch format implementation for batch serialization
+    :return: HTTPMessage with the batch in the body
+    """
+    headers = {CONTENT_TYPE_HEADER: event_format.get_batch_content_type()}
+    body = event_format.write_batch(events)
+    return HTTPMessage(headers=headers, body=body)
+
+
+def from_batch(
+    message: HTTPMessage,
+    event_format: BatchFormat,
+    event_factory: EventFactory | None = None,
+) -> list[BaseCloudEvent]:
+    """
+    Parse an HTTP batched content mode message to a list of CloudEvents.
+
+    Deserializes the batch from the HTTP body using the specified format. When
+    ``event_factory`` is None, each event's version is auto-detected independently.
+
+    :param message: HTTPMessage to parse
+    :param event_format: Batch format implementation for batch deserialization
+    :param event_factory: Factory to create CloudEvent instances (auto-detected if None)
+    :return: List of CloudEvent instances
+    """
+    events: list[BaseCloudEvent] = event_format.read_batch(event_factory, message.body)
+    return events
+
+
 def from_http(
     message: HTTPMessage,
     event_format: Format,
@@ -300,6 +336,22 @@ def from_http(
     :param event_factory: Factory function to create CloudEvent instances (auto-detected if None)
     :return: CloudEvent instance
     """
+    content_type = ""
+    for key, value in message.headers.items():
+        if key.lower() == CONTENT_TYPE_HEADER:
+            content_type = value
+            break
+
+    received_content_type = content_type.split(";")[0].strip().lower()
+    if isinstance(event_format, BatchFormat):
+        batch_content_type = event_format.get_batch_content_type().strip().lower()
+        if received_content_type == batch_content_type:
+            raise ValueError(
+                f"Received a batch payload: content type '{received_content_type}' "
+                f"matches the batch content type '{batch_content_type}'. "
+                "Use from_batch()/from_events_batch() to parse batched CloudEvents."
+            )
+
     if any(key.lower().startswith(CE_PREFIX) for key in message.headers.keys()):
         return from_binary(message, event_format, event_factory)
 
@@ -402,6 +454,38 @@ def from_structured_event(
     return from_structured(message, event_format, None)
 
 
+def to_events_batch(
+    events: list[BaseCloudEvent],
+    event_format: BatchFormat | None = None,
+) -> HTTPMessage:
+    """
+    Convenience wrapper for to_batch with JSON format as default.
+
+    :param events: The CloudEvents to convert
+    :param event_format: Batch format implementation (defaults to JSONFormat)
+    :return: HTTPMessage with the batch in the body
+    """
+    if event_format is None:
+        event_format = JSONFormat()
+    return to_batch(events, event_format)
+
+
+def from_events_batch(
+    message: HTTPMessage,
+    event_format: BatchFormat | None = None,
+) -> list[BaseCloudEvent]:
+    """
+    Convenience wrapper for from_batch with JSON format and auto-detection.
+
+    :param message: HTTPMessage to parse
+    :param event_format: Batch format implementation (defaults to JSONFormat)
+    :return: List of CloudEvent instances (version auto-detected per event)
+    """
+    if event_format is None:
+        event_format = JSONFormat()
+    return from_batch(message, event_format, None)
+
+
 def from_http_event(
     message: HTTPMessage,
     event_format: Format | None = None,

src/cloudevents/core/formats/base.py

@@ -12,7 +12,7 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-from typing import Any, Protocol
+from typing import Any, Protocol, runtime_checkable
 
 from cloudevents.core.base import BaseCloudEvent, EventFactory
 
@@ -88,3 +88,50 @@ def get_content_type(self) -> str:
         :return: Content type string for CloudEvents structured content mode
         """
         ...
+
+
+@runtime_checkable
+class BatchFormat(Format, Protocol):
+    """
+    Protocol for formats that additionally support the CloudEvents Batch mode.
+
+    Batch support is kept separate from :class:`Format` so a format can implement
+    single-event (de)serialization without being forced to implement batching, and
+    so adding batch support to the SDK does not break existing ``Format``
+    implementations. A format that supports batches implements both protocols.
+
+    The protocol is ``runtime_checkable`` so callers that accept a plain ``Format``
+    (e.g. ``from_http``) can detect batch capability with ``isinstance``.
+    """
+
+    def write_batch(self, events: list[BaseCloudEvent]) -> bytes:
+        """
+        Serialize a list of CloudEvents into a batch wire representation.
+
+        :param events: The CloudEvents to serialize.
+        :return: The batch serialized as bytes.
+        """
+        ...
+
+    def read_batch(
+        self,
+        event_factory: EventFactory | None,
+        data: str | bytes,
+    ) -> list[BaseCloudEvent]:
+        """
+        Deserialize a batch wire representation into a list of CloudEvents.
+
+        :param event_factory: Factory to create CloudEvent instances, or None to
+                              auto-detect each element's version.
+        :param data: The serialized batch as a string or bytes.
+        :return: The list of CloudEvent instances.
+        """
+        ...
+
+    def get_batch_content_type(self) -> str:
+        """
+        Get the Content-Type header value for batch mode.
+
+        :return: Content type string for CloudEvents batch content mode.
+        """
+        ...