Deprecate cloudevents messages

dapr/common/pubsub/subscription.py

@@ -1,13 +1,21 @@
 import json
-from typing import Optional, Union
+from typing import Mapping, Optional, Union
 
 from google.protobuf.json_format import MessageToDict
 
 from dapr.proto.runtime.v1.appcallback_pb2 import TopicEventRequest
 
 
 class SubscriptionMessage:
-    def __init__(self, msg: TopicEventRequest):
+    """Message delivered to Dapr pub/sub subscribers.
+
+    Delivered by both the streaming subscription API (``DaprClient.subscribe``) and the
+    App callback API (``@app.subscribe``). ``metadata`` carries delivery metadata (gRPC
+    invocation metadata and, for bulk deliveries, per-entry publish metadata); it is only
+    populated for App callback deliveries.
+    """
+
+    def __init__(self, msg: TopicEventRequest, metadata: Optional[Mapping[str, str]] = None):
         self._id: str = msg.id
         self._source: str = msg.source
         self._type: str = msg.type
@@ -17,6 +25,7 @@ def __init__(self, msg: TopicEventRequest):
         self._pubsub_name: str = msg.pubsub_name
         self._raw_data: bytes = msg.data
         self._data: Optional[Union[dict, str]] = None
+        self._metadata: dict[str, str] = dict(metadata) if metadata else {}
 
         try:
             self._extensions = MessageToDict(msg.extensions)
@@ -58,6 +67,9 @@ def extensions(self):
     def data(self):
         return self._data
 
+    def metadata(self) -> dict[str, str]:
+        return self._metadata
+
     def _parse_data_content(self):
         try:
             if self._data_content_type == 'application/json':

dapr/ext/grpc/AGENTS.md

@@ -14,7 +14,7 @@ dapr/ext/grpc/
 
 tests/ext/grpc/
 ├── test_app.py                    # Decorator registration tests
-├── test_servicier.py              # Routing, handlers, bulk events
+├── test_servicer.py               # Routing, handlers, bulk events
 ├── test_health_servicer.py        # Health check tests
 └── test_topic_event_response.py   # Response status tests
 ```
@@ -27,6 +27,7 @@ Installed via the `grpc` extra on core dapr: `pip install "dapr[grpc]"`.
 from dapr.ext.grpc import (
     App,                    # Main entry point — decorator-based gRPC server
     Rule,                   # CEL-based topic rule with priority
+    SubscriptionMessage,    # Event type received by pub/sub topic handlers (preferred)
     InvokeMethodRequest,    # Request object for service invocation handlers
     InvokeMethodResponse,   # Response object for service invocation handlers
     BindingRequest,         # Request object for input binding handlers
@@ -56,7 +57,7 @@ def handle_method(request: InvokeMethodRequest) -> InvokeMethodResponse:
 
 @app.subscribe(pubsub_name='pubsub', topic='orders', metadata={}, dead_letter_topic=None,
                 rule=Rule('event.type == "order"', priority=1), disable_topic_validation=False)
-def handle_event(event: v1.Event) -> Optional[TopicEventResponse]:
+def handle_event(event: SubscriptionMessage) -> Optional[TopicEventResponse]:
     ...
 
 @app.binding('binding_name')
@@ -115,7 +116,7 @@ app.register_health_check(lambda: None)  # Not a decorator — direct registrati
 ## Dependencies (declared via the `grpc` extra in the root `pyproject.toml`)
 
 - `dapr` (core, same wheel as this extension)
-- `cloudevents >= 1.0.0, < 2.0.0`
+- `cloudevents >= 1.0.0, < 2.0.0` (deprecated, only used for the legacy handler event type)
 
 ## Testing
 
@@ -125,14 +126,14 @@ uv run python -m unittest discover -v ./tests/ext/grpc
 
 Test patterns:
 - `test_app.py` — decorator registration, health check registration
-- `test_servicier.py` — handler invocation with mock gRPC context, return type handling (str, bytes, proto, response object), topic subscriptions, bulk events, bindings, duplicate registration errors
+- `test_servicer.py` — handler invocation with mock gRPC context, return type handling (str, bytes, proto, response object), topic subscriptions, bulk events, bindings, duplicate registration errors
 - `test_health_servicer.py` — health check callback invocation, missing callback (UNIMPLEMENTED)
 - `test_topic_event_response.py` — response creation from enum and string values
 
 ## Key details
 
 - **Synchronous only**: Uses `grpc.server()` with `ThreadPoolExecutor(10)`. No async handler support.
 - **Default port**: 3010 (from `dapr.conf.global_settings.GRPC_APP_PORT`)
-- **CloudEvents**: Requires `cloudevents >= 1.0.0` for pub/sub event handling
+- **Topic handler event type**: inferred from the handler annotation. Annotating the event parameter with `dapr.ext.grpc.SubscriptionMessage` — the same SDK-owned type the streaming subscription API (`DaprClient.subscribe`) delivers, with `metadata()` populated from the gRPC invocation metadata — delivers that type. Unannotated or otherwise-annotated handlers receive the DEPRECATED `cloudevents.sdk.event.v1.Event` and `subscribe()` emits a `DeprecationWarning` at registration. Deprecation timeline: 1.20 delivers `SubscriptionMessage` to unannotated handlers (legacy only via explicit `v1.Event` annotation), 1.21 drops `cloudevents` from the `grpc` extra (import becomes conditional), 1.22 removes the legacy path entirely (same release the `flask_dapr` shim goes away). New code must annotate with `SubscriptionMessage`. (Internally the choice is plumbed through `_CallbackServicer.register_topic(legacy_cloudevent=...)`.)
 - **Duplicate registration**: Registering the same method/topic/binding name twice raises `ValueError`
 - **Missing handlers**: Calling an unregistered method/topic/binding raises `NotImplementedError` (gRPC UNIMPLEMENTED)

dapr/ext/grpc/__init__.py

@@ -16,6 +16,7 @@
 from dapr.clients.grpc._jobs import ConstantFailurePolicy, DropFailurePolicy, FailurePolicy, Job
 from dapr.clients.grpc._request import BindingRequest, InvokeMethodRequest, JobEvent
 from dapr.clients.grpc._response import InvokeMethodResponse, TopicEventResponse
+from dapr.common.pubsub.subscription import SubscriptionMessage
 
 try:
     from dapr.ext.grpc.app import App, Rule  # type:ignore
@@ -30,6 +31,7 @@
 __all__ = [
     'App',
     'Rule',
+    'SubscriptionMessage',
     'InvokeMethodRequest',
     'InvokeMethodResponse',
     'BindingRequest',

dapr/ext/grpc/_servicer.py

@@ -25,6 +25,7 @@
 from dapr.clients._constants import DEFAULT_JSON_CONTENT_TYPE
 from dapr.clients.grpc._request import BindingRequest, InvokeMethodRequest, JobEvent
 from dapr.clients.grpc._response import InvokeMethodResponse, TopicEventResponse
+from dapr.common.pubsub.subscription import SubscriptionMessage
 from dapr.proto import appcallback_service_v1, appcallback_v1, common_v1
 from dapr.proto.common.v1.common_pb2 import InvokeRequest
 from dapr.proto.runtime.v1.appcallback_pb2 import (
@@ -36,7 +37,9 @@
 )
 
 InvokeMethodCallable = Callable[[InvokeMethodRequest], Union[str, bytes, InvokeMethodResponse]]
-TopicSubscribeCallable = Callable[[v1.Event], Optional[TopicEventResponse]]
+TopicSubscribeCallable = Callable[
+    [Union[v1.Event, SubscriptionMessage]], Optional[TopicEventResponse]
+]
 BindingCallable = Callable[[BindingRequest], None]
 JobEventCallable = Callable[[JobEvent], None]
 
@@ -74,6 +77,7 @@ class _CallbackServicer(
     def __init__(self):
         self._invoke_method_map: Dict[str, InvokeMethodCallable] = {}
         self._topic_map: Dict[str, TopicSubscribeCallable] = {}
+        self._topic_legacy_event: Dict[TopicSubscribeCallable, bool] = {}
         self._binding_map: Dict[str, BindingCallable] = {}
         self._job_event_map: Dict[str, JobEventCallable] = {}
 
@@ -120,8 +124,15 @@ def register_topic(
         dead_letter_topic: Optional[str] = None,
         rule: Optional[Rule] = None,
         disable_topic_validation: Optional[bool] = False,
+        legacy_cloudevent: bool = True,
     ) -> None:
-        """Registers topic subscription for pubsub."""
+        """Registers topic subscription for pubsub.
+
+        Args:
+            legacy_cloudevent (bool): when True (deprecated default), the handler receives a
+                ``cloudevents.sdk.event.v1.Event``; when False, it receives a
+                :class:`dapr.common.pubsub.subscription.SubscriptionMessage`.
+        """
         topic_key = pubsub_name + DELIMITER + topic
         pubsub_topic = topic_key + DELIMITER
         if rule is not None:
@@ -130,6 +141,7 @@ def register_topic(
         if pubsub_topic in self._topic_map:
             raise ValueError(f'{topic} is already registered with {pubsub_name}')
         self._topic_map[pubsub_topic] = cb
+        self._topic_legacy_event[cb] = legacy_cloudevent
         routing_path = path if rule is not None else topic
         self._route_map[(pubsub_name, routing_path)] = cb
 
@@ -233,21 +245,27 @@ def OnTopicEvent(self, request: TopicEventRequest, context):
             context.set_code(grpc.StatusCode.UNIMPLEMENTED)  # type: ignore
             raise NotImplementedError(f'topic {request.topic} is not implemented!')
 
-        customdata: Struct = request.extensions
-        extensions = dict()
-        for k, v in customdata.items():
-            extensions[k] = v
-        for k, v in context.invocation_metadata():
-            extensions['_metadata_' + k] = v
-
-        event = v1.Event()
-        event.SetEventType(request.type)
-        event.SetEventID(request.id)
-        event.SetSource(request.source)
-        event.SetData(request.data)
-        event.SetContentType(request.data_content_type)
-        event.SetSubject(request.topic)
-        event.SetExtensions(extensions)
+        invocation_metadata = dict(context.invocation_metadata())
+
+        event: Union[v1.Event, SubscriptionMessage]
+        if self._topic_legacy_event.get(cb, True):
+            customdata: Struct = request.extensions
+            extensions = dict()
+            for k, v in customdata.items():
+                extensions[k] = v
+            for k, v in invocation_metadata.items():
+                extensions['_metadata_' + k] = v
+
+            event = v1.Event()
+            event.SetEventType(request.type)
+            event.SetEventID(request.id)
+            event.SetSource(request.source)
+            event.SetData(request.data)
+            event.SetContentType(request.data_content_type)
+            event.SetSubject(request.topic)
+            event.SetExtensions(extensions)
+        else:
+            event = SubscriptionMessage(request, invocation_metadata)
 
         response = cb(event)
         if isinstance(response, TopicEventResponse):
@@ -323,35 +341,20 @@ def _handle_bulk_topic_event(
         if cb is None:
             return None  # we don't have a handler
 
+        use_legacy_event = self._topic_legacy_event.get(cb, True)
+        invocation_metadata = dict(context.invocation_metadata())
+
         statuses = []
         for entry in request.entries:
             entry_id = entry.entry_id
             try:
-                # Build event from entry & send req with many entries
-                event = v1.Event()
-                extensions = dict()
-                if entry.HasField('cloud_event') and entry.cloud_event:
-                    ce = entry.cloud_event
-                    event.SetEventType(ce.type)
-                    event.SetEventID(ce.id)
-                    event.SetSource(ce.source)
-                    event.SetData(ce.data)
-                    event.SetContentType(ce.data_content_type)
-                    if ce.extensions:
-                        for k, v in ce.extensions.items():
-                            extensions[k] = v
+                event: Union[v1.Event, SubscriptionMessage]
+                if use_legacy_event:
+                    event = self._bulk_entry_legacy_event(entry, request, invocation_metadata)
                 else:
-                    event.SetEventID(entry_id)
-                    event.SetData(entry.bytes if entry.HasField('bytes') else b'')
-                    event.SetContentType(entry.content_type or '')
-                event.SetSubject(request.topic)
-                if entry.metadata:
-                    for k, v in entry.metadata.items():
-                        extensions[k] = v
-                for k, v in context.invocation_metadata():
-                    extensions['_metadata_' + k] = v
-                if extensions:
-                    event.SetExtensions(extensions)
+                    event = self._bulk_entry_subscription_message(
+                        entry, request, invocation_metadata
+                    )
 
                 response = cb(event)  # invoke app registered handler and send event
                 if isinstance(response, TopicEventResponse):
@@ -365,6 +368,70 @@ def _handle_bulk_topic_event(
             )
         return appcallback_v1.TopicEventBulkResponse(statuses=statuses)
 
+    def _bulk_entry_legacy_event(
+        self,
+        entry,
+        request: TopicEventBulkRequest,
+        invocation_metadata: Dict[str, str],
+    ) -> v1.Event:
+        """Builds the deprecated cloudevents v1.Event for a bulk entry."""
+        event = v1.Event()
+        extensions = dict()
+        if entry.HasField('cloud_event') and entry.cloud_event:
+            ce = entry.cloud_event
+            event.SetEventType(ce.type)
+            event.SetEventID(ce.id)
+            event.SetSource(ce.source)
+            event.SetData(ce.data)
+            event.SetContentType(ce.data_content_type)
+            if ce.extensions:
+                for k, v in ce.extensions.items():
+                    extensions[k] = v
+        else:
+            event.SetEventID(entry.entry_id)
+            event.SetData(entry.bytes if entry.HasField('bytes') else b'')
+            event.SetContentType(entry.content_type or '')
+        event.SetSubject(request.topic)
+        if entry.metadata:
+            for k, v in entry.metadata.items():
+                extensions[k] = v
+        for k, v in invocation_metadata.items():
+            extensions['_metadata_' + k] = v
+        if extensions:
+            event.SetExtensions(extensions)
+        return event
+
+    def _bulk_entry_subscription_message(
+        self,
+        entry,
+        request: TopicEventBulkRequest,
+        invocation_metadata: Dict[str, str],
+    ) -> SubscriptionMessage:
+        """Builds a SubscriptionMessage for a bulk entry via a synthesized TopicEventRequest."""
+        if entry.HasField('cloud_event') and entry.cloud_event:
+            ce = entry.cloud_event
+            entry_request = TopicEventRequest(
+                id=ce.id,
+                source=ce.source,
+                type=ce.type,
+                spec_version=ce.spec_version,
+                data_content_type=ce.data_content_type,
+                data=ce.data,
+                topic=request.topic,
+                pubsub_name=request.pubsub_name,
+                extensions=ce.extensions,
+            )
+        else:
+            entry_request = TopicEventRequest(
+                id=entry.entry_id,
+                data=entry.bytes if entry.HasField('bytes') else b'',
+                data_content_type=entry.content_type or '',
+                topic=request.topic,
+                pubsub_name=request.pubsub_name,
+            )
+        metadata = {**invocation_metadata, **dict(entry.metadata)}
+        return SubscriptionMessage(entry_request, metadata)
+
     def OnBulkTopicEvent(self, request: TopicEventBulkRequest, context):
         """Subscribes bulk events from Pubsub"""
         response = self._handle_bulk_topic_event(request, context)