From 40f5d1b582d4ffcafdcd72c6b5d156f9ebb48e55 Mon Sep 17 00:00:00 2001 From: Sai Kishore Punagani <63619246+saikishore-p@users.noreply.github.com> Date: Sun, 13 Sep 2026 00:04:00 -0400 Subject: [PATCH 1/2] refactor: extract shared bases from the gRPC callback and health servicers Splits _CallbackServicer into a base holding everything that does not invoke a user handler - the handler registries, topic routing, and the translation of incoming requests into the SDK types handlers receive - and a thin subclass supplying the synchronous gRPC entry points. _HealthCheckServicer is split the same way, with callback registration moving to the base. This is preparation for asyncio servicers, which need all of the former and none of the latter. Keeping that logic in one place avoids the duplication that left the previous attempt at async support (#829) unmergeable once SubscriptionMessage delivery, the _route_map rewrite and bulk topic events landed on the synchronous side only. Behaviour is unchanged. Nothing is removed from either class's reachable surface - attribute lookup walks the MRO, so callers reaching into internals such as app._servicer, which examples/pubsub-simple does, are unaffected. The existing tests pass unmodified. Also extracts _resolve_topic_event_type in app.py so the deprecation warning for unannotated topic handlers has a single definition to share. Signed-off-by: Sai Kishore Punagani <63619246+saikishore-p@users.noreply.github.com> --- dapr/ext/grpc/_health_servicer.py | 36 ++- dapr/ext/grpc/_servicer.py | 389 ++++++++++++++++++------------ dapr/ext/grpc/app.py | 33 ++- 3 files changed, 282 insertions(+), 176 deletions(-) diff --git a/dapr/ext/grpc/_health_servicer.py b/dapr/ext/grpc/_health_servicer.py index 8405fbf8e..db51785e9 100644 --- a/dapr/ext/grpc/_health_servicer.py +++ b/dapr/ext/grpc/_health_servicer.py @@ -1,33 +1,49 @@ -from typing import Callable, Optional +from typing import Awaitable, Callable, Optional import grpc +from dapr.ext.grpc._servicer import _ServicerContext from dapr.proto import appcallback_service_v1 from dapr.proto.runtime.v1.appcallback_pb2 import HealthCheckResponse HealthCheckCallable = Optional[Callable[[], None]] +# The asyncio servicer awaits its callback, so it also accepts an awaitable-returning one. +AsyncHealthCheckCallable = Callable[[], Optional[Awaitable[None]]] -class _HealthCheckServicer(appcallback_service_v1.AppCallbackHealthCheckServicer): - """The implementation of HealthCheck Server. +class _HealthCheckServicerBase(appcallback_service_v1.AppCallbackHealthCheckServicer): + """Health check registration shared by the sync and asyncio servicers. - :class:`App` provides useful decorators to register method, topic, input bindings. + Mirrors the :class:`_CallbackServicerBase` arrangement: everything that does not invoke + the callback lives here, and each servicer supplies only the gRPC entry point. """ def __init__(self): - self._health_check_cb: Optional[HealthCheckCallable] = None + self._health_check_cb: Optional[AsyncHealthCheckCallable] = None - def register_health_check(self, cb: HealthCheckCallable) -> None: + def register_health_check(self, cb: Optional[AsyncHealthCheckCallable]) -> None: if not cb: raise ValueError('health check callback must be defined') self._health_check_cb = cb - def HealthCheck(self, request, context): - """Health check.""" - + def _require_health_check_cb(self, context: _ServicerContext) -> AsyncHealthCheckCallable: + """Returns the registered callback, or marks the RPC UNIMPLEMENTED if there is none.""" if not self._health_check_cb: context.set_code(grpc.StatusCode.UNIMPLEMENTED) # type: ignore context.set_details('Method not implemented!') raise NotImplementedError('Method not implemented!') - self._health_check_cb() + return self._health_check_cb + + +class _HealthCheckServicer(_HealthCheckServicerBase): + """The synchronous HealthCheck servicer. + + Shares registration with the asyncio servicer via their common base; only this gRPC entry + point differs, calling the callback directly rather than awaiting it. + """ + + def HealthCheck(self, request, context): + """Health check.""" + health_check = self._require_health_check_cb(context) + health_check() return HealthCheckResponse() diff --git a/dapr/ext/grpc/_servicer.py b/dapr/ext/grpc/_servicer.py index 33d9171c0..78c0504a5 100644 --- a/dapr/ext/grpc/_servicer.py +++ b/dapr/ext/grpc/_servicer.py @@ -14,7 +14,7 @@ """ import warnings -from typing import Callable, Dict, List, Optional, Tuple, Union +from typing import Any, Callable, Dict, List, Optional, Protocol, Tuple, Union import grpc from cloudevents.sdk.event import v1 # type: ignore @@ -32,6 +32,7 @@ BindingEventRequest, JobEventRequest, TopicEventBulkRequest, + TopicEventBulkRequestEntry, TopicEventBulkResponse, TopicEventRequest, ) @@ -45,6 +46,23 @@ DELIMITER = ':' +HandlerResponse = Union[str, bytes, GrpcMessage, InvokeMethodResponse] + + +class _ServicerContext(Protocol): + """The subset of the gRPC servicer context the shared helpers rely on. + + Declared structurally because the synchronous and asyncio servicers are handed a + ``grpc.ServicerContext`` and a ``grpc.aio.ServicerContext`` respectively, which share no + common base class but expose these members identically. + """ + + def set_code(self, code: grpc.StatusCode) -> None: ... + + def set_details(self, details: str) -> None: ... + + def invocation_metadata(self) -> Any: ... + class Rule: def __init__(self, match: str, priority: int) -> None: @@ -62,16 +80,16 @@ def __init__( self.rules = rules -class _CallbackServicer( +class _CallbackServicerBase( appcallback_service_v1.AppCallbackServicer, appcallback_service_v1.AppCallbackAlphaServicer ): - """The implementation of AppCallback Server. - - This internal class implements application server and provides helpers to register - method, topic, and input bindings. It implements the routing handling logic to route - mulitple methods, topics, and bindings. + """Handler registration and request translation shared by the sync and asyncio servicers. - :class:`App` provides useful decorators to register method, topic, input bindings. + Holds every part of the AppCallback implementation that does not invoke a user handler: + the handler registries, the topic routing table, and the translation of incoming gRPC + requests into the SDK types handlers receive. :class:`_CallbackServicer` and + :class:`dapr.ext.grpc.aio._servicer._AioCallbackServicer` add the gRPC entry points on + top, which differ only in whether they await the handler. """ def __init__(self): @@ -199,117 +217,98 @@ def register_job_event(self, name: str, cb: JobEventCallable) -> None: raise ValueError(f'Job event handler for {name} is already registered') self._job_event_map[name] = cb - def OnInvoke(self, request: InvokeRequest, context): - """Invokes service method with InvokeRequest.""" - if request.method not in self._invoke_method_map: - context.set_code(grpc.StatusCode.UNIMPLEMENTED) # type: ignore - raise NotImplementedError(f'{request.method} method not implemented!') + def _unimplemented(self, context: _ServicerContext, message: str) -> NotImplementedError: + """Marks the RPC UNIMPLEMENTED and returns the error for the caller to raise.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) # type: ignore + return NotImplementedError(message) + def _build_invoke_request( + self, request: InvokeRequest, context: _ServicerContext + ) -> InvokeMethodRequest: + """Translates an InvokeRequest into the request object handlers receive.""" req = InvokeMethodRequest(request.data, request.content_type) req.metadata = context.invocation_metadata() - resp = self._invoke_method_map[request.method](req) + return req - if not resp: - return common_v1.InvokeResponse() + def _to_invoke_response_data( + self, resp: HandlerResponse, context: _ServicerContext, method: str + ) -> InvokeMethodResponse: + """Normalizes a method handler's return value into an InvokeMethodResponse.""" + if isinstance(resp, InvokeMethodResponse): + return resp resp_data = InvokeMethodResponse() if isinstance(resp, (bytes, str)): resp_data.set_data(resp) resp_data.content_type = DEFAULT_JSON_CONTENT_TYPE - elif isinstance(resp, GrpcMessage): + return resp_data + if isinstance(resp, GrpcMessage): resp_data.set_data(resp) - elif isinstance(resp, InvokeMethodResponse): - resp_data = resp - else: - context.set_code(grpc.StatusCode.OUT_OF_RANGE) - context.set_details(f'{type(resp)} is the invalid return type.') - raise NotImplementedError(f'{request.method} method not implemented!') + return resp_data - if len(resp_data.get_headers()) > 0: - context.send_initial_metadata(resp_data.get_headers()) + context.set_code(grpc.StatusCode.OUT_OF_RANGE) + context.set_details(f'{type(resp)} is the invalid return type.') + raise NotImplementedError(f'{method} method not implemented!') + def _to_invoke_response(self, resp_data: InvokeMethodResponse) -> common_v1.InvokeResponse: + """Packs a normalized handler response into the wire InvokeResponse.""" content_type = '' if resp_data.content_type: content_type = resp_data.content_type - return common_v1.InvokeResponse(data=resp_data.proto, content_type=content_type) - def ListTopicSubscriptions(self, request, context): - """Lists all topics subscribed by this app.""" - return appcallback_v1.ListTopicSubscriptionsResponse(subscriptions=self._registered_topics) - - def OnTopicEvent(self, request: TopicEventRequest, context): - """Subscribes events from Pubsub.""" - cb = self._get_topic_callback(request.pubsub_name, request.topic, request.path) - if cb is None: - context.set_code(grpc.StatusCode.UNIMPLEMENTED) # type: ignore - raise NotImplementedError(f'topic {request.topic} is not implemented!') + def _build_topic_event( + self, + cb: TopicSubscribeCallable, + request: TopicEventRequest, + invocation_metadata: Dict[str, str], + ) -> Union[v1.Event, SubscriptionMessage]: + """Translates a topic event request into the event type the handler expects.""" + if not self._topic_legacy_event.get(cb, True): + return SubscriptionMessage(request, invocation_metadata) - invocation_metadata = dict(context.invocation_metadata()) + 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: 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) + 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) + return event - response = cb(event) + def _to_topic_event_response( + self, response: Optional[TopicEventResponse] + ) -> Union[appcallback_v1.TopicEventResponse, empty_pb2.Empty]: + """Maps a topic handler's return value to the single-event wire response.""" if isinstance(response, TopicEventResponse): return appcallback_v1.TopicEventResponse(status=response.status.value) return empty_pb2.Empty() - def ListInputBindings(self, request, context): - """Lists all input bindings subscribed by this app.""" - return appcallback_v1.ListInputBindingsResponse(bindings=self._registered_bindings) - - def OnBindingEvent(self, request: BindingEventRequest, context): - """Listens events from the input bindings - User application can save the states or send the events to the output - bindings optionally by returning BindingEventResponse. - """ - if request.name not in self._binding_map: - context.set_code(grpc.StatusCode.UNIMPLEMENTED) # type: ignore - raise NotImplementedError(f'{request.name} binding not implemented!') - + def _build_binding_request( + self, request: BindingEventRequest, context: _ServicerContext + ) -> BindingRequest: + """Translates a binding event request into the request object handlers receive.""" req = BindingRequest(request.data, dict(request.metadata)) req.metadata = context.invocation_metadata() - self._binding_map[request.name](req) + return req - # TODO: support output bindings options - return appcallback_v1.BindingEventResponse() + def _build_job_event(self, request: JobEventRequest, context: _ServicerContext) -> JobEvent: + """Translates a job event request into the JobEvent handlers receive. - def _handle_job_event(self, request: JobEventRequest, context): - """Handles job events from Dapr runtime. - - This method is called by Dapr when a scheduled job is triggered. - It routes the job event to the appropriate registered handler based on the job name. - - Args: - request (JobEventRequest): The job event request from Dapr. - context: The gRPC context. - - Returns: - appcallback_v1.JobEventResponse: Empty response indicating successful handling. + Raises NotImplementedError (UNIMPLEMENTED) when no handler is registered for the job. """ - job_name = request.name - - if job_name not in self._job_event_map: - context.set_code(grpc.StatusCode.UNIMPLEMENTED) # type: ignore - raise NotImplementedError(f'Job event handler for {job_name} not implemented!') + if request.name not in self._job_event_map: + raise self._unimplemented( + context, f'Job event handler for {request.name} not implemented!' + ) # Create a JobEvent object matching Go SDK's common.JobEvent # Extract raw data bytes from the Any proto (matching Go implementation) @@ -317,60 +316,39 @@ def _handle_job_event(self, request: JobEventRequest, context): if request.HasField('data') and request.data.value: data_bytes = request.data.value - job_event = JobEvent(name=request.name, data=data_bytes) - - # Call the registered handler with the JobEvent object - self._job_event_map[job_name](job_event) - - # Return empty response - return appcallback_v1.JobEventResponse() - - def OnJobEvent(self, request: JobEventRequest, context): - """Handles job events on the stable AppCallback service.""" - return self._handle_job_event(request, context) - - def OnJobEventAlpha1(self, request: JobEventRequest, context): - """Handles job events on the deprecated AppCallbackAlpha service.""" - return self._handle_job_event(request, context) + return JobEvent(name=request.name, data=data_bytes) - def _handle_bulk_topic_event( - self, request: TopicEventBulkRequest, context - ) -> Optional[TopicEventBulkResponse]: - """Process bulk topic event request - routes each entry to the appropriate topic handler.""" - cb = self._get_topic_callback(request.pubsub_name, request.topic, request.path) - 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()) + def _warn_bulk_alpha1_deprecated(self) -> None: + """Emits the shared deprecation warning for the alpha bulk-topic entry point.""" + warnings.warn( + 'OnBulkTopicEventAlpha1 is deprecated. Use OnBulkTopicEvent instead.', + DeprecationWarning, + stacklevel=3, + ) - statuses = [] - for entry in request.entries: - entry_id = entry.entry_id - try: - event: Union[v1.Event, SubscriptionMessage] - if use_legacy_event: - event = self._bulk_entry_legacy_event(entry, request, invocation_metadata) - else: - event = self._bulk_entry_subscription_message( - entry, request, invocation_metadata - ) - - response = cb(event) # invoke app registered handler and send event - if isinstance(response, TopicEventResponse): - status = response.status.value - else: - status = appcallback_v1.TopicEventResponse.TopicEventResponseStatus.SUCCESS - except Exception: - status = appcallback_v1.TopicEventResponse.TopicEventResponseStatus.RETRY - statuses.append( - appcallback_v1.TopicEventBulkResponseEntry(entry_id=entry_id, status=status) - ) - return appcallback_v1.TopicEventBulkResponse(statuses=statuses) + def _bulk_entry_event( + self, + entry: TopicEventBulkRequestEntry, + request: TopicEventBulkRequest, + use_legacy_event: bool, + invocation_metadata: Dict[str, str], + ) -> Union[v1.Event, SubscriptionMessage]: + """Translates one bulk entry into the event type the handler expects.""" + if use_legacy_event: + return self._bulk_entry_legacy_event(entry, request, invocation_metadata) + return self._bulk_entry_subscription_message(entry, request, invocation_metadata) + + def _bulk_entry_status( + self, response: Optional[TopicEventResponse] + ) -> appcallback_v1.TopicEventResponse.TopicEventResponseStatus.ValueType: + """Maps a topic handler's return value to a bulk-response entry status.""" + if isinstance(response, TopicEventResponse): + return response.status.value + return appcallback_v1.TopicEventResponse.TopicEventResponseStatus.SUCCESS def _bulk_entry_legacy_event( self, - entry, + entry: TopicEventBulkRequestEntry, request: TopicEventBulkRequest, invocation_metadata: Dict[str, str], ) -> v1.Event: @@ -403,7 +381,7 @@ def _bulk_entry_legacy_event( def _bulk_entry_subscription_message( self, - entry, + entry: TopicEventBulkRequestEntry, request: TopicEventBulkRequest, invocation_metadata: Dict[str, str], ) -> SubscriptionMessage: @@ -432,25 +410,126 @@ def _bulk_entry_subscription_message( metadata = {**invocation_metadata, **dict(entry.metadata)} return SubscriptionMessage(entry_request, metadata) + +class _CallbackServicer(_CallbackServicerBase): + """The implementation of AppCallback Server. + + This internal class implements application server and provides helpers to register + method, topic, and input bindings. It implements the routing handling logic to route + mulitple methods, topics, and bindings. + + :class:`App` provides useful decorators to register method, topic, input bindings. + """ + + def OnInvoke(self, request: InvokeRequest, context): + """Invokes service method with InvokeRequest.""" + if request.method not in self._invoke_method_map: + raise self._unimplemented(context, f'{request.method} method not implemented!') + + req = self._build_invoke_request(request, context) + resp = self._invoke_method_map[request.method](req) + + if not resp: + return common_v1.InvokeResponse() + + resp_data = self._to_invoke_response_data(resp, context, request.method) + + headers = resp_data.get_headers() + if len(headers) > 0: + context.send_initial_metadata(headers) + + return self._to_invoke_response(resp_data) + + def ListTopicSubscriptions(self, request, context): + """Lists all topics subscribed by this app.""" + return appcallback_v1.ListTopicSubscriptionsResponse(subscriptions=self._registered_topics) + + def OnTopicEvent(self, request: TopicEventRequest, context): + """Subscribes events from Pubsub.""" + cb = self._get_topic_callback(request.pubsub_name, request.topic, request.path) + if cb is None: + raise self._unimplemented(context, f'topic {request.topic} is not implemented!') + + event = self._build_topic_event(cb, request, dict(context.invocation_metadata())) + + return self._to_topic_event_response(cb(event)) + + def ListInputBindings(self, request, context): + """Lists all input bindings subscribed by this app.""" + return appcallback_v1.ListInputBindingsResponse(bindings=self._registered_bindings) + + def OnBindingEvent(self, request: BindingEventRequest, context): + """Listens events from the input bindings + User application can save the states or send the events to the output + bindings optionally by returning BindingEventResponse. + """ + if request.name not in self._binding_map: + raise self._unimplemented(context, f'{request.name} binding not implemented!') + + req = self._build_binding_request(request, context) + self._binding_map[request.name](req) + + # TODO: support output bindings options + return appcallback_v1.BindingEventResponse() + + def _handle_job_event(self, request: JobEventRequest, context): + """Handles job events from Dapr runtime. + + This method is called by Dapr when a scheduled job is triggered. + It routes the job event to the appropriate registered handler based on the job name. + + Args: + request (JobEventRequest): The job event request from Dapr. + context: The gRPC context. + + Returns: + appcallback_v1.JobEventResponse: Empty response indicating successful handling. + """ + job_event = self._build_job_event(request, context) + self._job_event_map[request.name](job_event) + return appcallback_v1.JobEventResponse() + + def OnJobEvent(self, request: JobEventRequest, context): + """Handles job events on the stable AppCallback service.""" + return self._handle_job_event(request, context) + + def OnJobEventAlpha1(self, request: JobEventRequest, context): + """Handles job events on the deprecated AppCallbackAlpha service.""" + return self._handle_job_event(request, context) + + def _handle_bulk_topic_event( + self, request: TopicEventBulkRequest, context: _ServicerContext + ) -> TopicEventBulkResponse: + """Process bulk topic event request - routes each entry to the appropriate topic handler.""" + cb = self._get_topic_callback(request.pubsub_name, request.topic, request.path) + if cb is None: + raise self._unimplemented(context, f'bulk topic {request.topic} is not implemented!') + + 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: + event = self._bulk_entry_event( + entry, request, use_legacy_event, invocation_metadata + ) + status = self._bulk_entry_status(cb(event)) + except Exception: + status = appcallback_v1.TopicEventResponse.TopicEventResponseStatus.RETRY + statuses.append( + appcallback_v1.TopicEventBulkResponseEntry(entry_id=entry_id, status=status) + ) + return appcallback_v1.TopicEventBulkResponse(statuses=statuses) + def OnBulkTopicEvent(self, request: TopicEventBulkRequest, context): """Subscribes bulk events from Pubsub""" - response = self._handle_bulk_topic_event(request, context) - if response is None: - context.set_code(grpc.StatusCode.UNIMPLEMENTED) # type: ignore - raise NotImplementedError(f'bulk topic {request.topic} is not implemented!') - return response + return self._handle_bulk_topic_event(request, context) def OnBulkTopicEventAlpha1(self, request: TopicEventBulkRequest, context): """Subscribes bulk events from Pubsub. Deprecated: Use OnBulkTopicEvent instead. """ - warnings.warn( - 'OnBulkTopicEventAlpha1 is deprecated. Use OnBulkTopicEvent instead.', - DeprecationWarning, - stacklevel=2, - ) - response = self._handle_bulk_topic_event(request, context) - if response is None: - context.set_code(grpc.StatusCode.UNIMPLEMENTED) # type: ignore - raise NotImplementedError(f'bulk topic {request.topic} is not implemented!') - return response + self._warn_bulk_alpha1_deprecated() + return self._handle_bulk_topic_event(request, context) diff --git a/dapr/ext/grpc/app.py b/dapr/ext/grpc/app.py index c59a05cdb..059447158 100644 --- a/dapr/ext/grpc/app.py +++ b/dapr/ext/grpc/app.py @@ -44,6 +44,27 @@ def _wants_subscription_message(func: Callable) -> bool: return isinstance(annotation, type) and issubclass(annotation, SubscriptionMessage) +def _resolve_topic_event_type(func: Callable) -> bool: + """True if the handler opts into SubscriptionMessage, warning about the legacy type if not. + + Shared by the synchronous and asyncio ``App.subscribe`` decorators so both emit the same + deprecation guidance. The stacklevel points past this helper and the decorator, at the + line applying ``@app.subscribe``. + """ + wants_subscription_message = _wants_subscription_message(func) + if not wants_subscription_message: + warnings.warn( + 'Topic handlers receive a deprecated cloudevents.sdk.event.v1.Event unless ' + 'their event parameter is annotated with dapr.ext.grpc.SubscriptionMessage. ' + 'Annotate the handler to adopt SubscriptionMessage and silence this warning; ' + 'a future release will deliver SubscriptionMessage to all handlers and drop ' + 'the cloudevents dependency.', + DeprecationWarning, + stacklevel=3, + ) + return wants_subscription_message + + class App: """App object implements a Dapr application callback which can interact with Dapr runtime. Once its object is initiated, it will act as a central registry for service invocation, @@ -205,17 +226,7 @@ def topic(event: SubscriptionMessage) -> None: """ def decorator(func): - handler_wants_subscription_message = _wants_subscription_message(func) - if not handler_wants_subscription_message: - warnings.warn( - 'Topic handlers receive a deprecated cloudevents.sdk.event.v1.Event unless ' - 'their event parameter is annotated with dapr.ext.grpc.SubscriptionMessage. ' - 'Annotate the handler to adopt SubscriptionMessage and silence this warning; ' - 'a future release will deliver SubscriptionMessage to all handlers and drop ' - 'the cloudevents dependency.', - DeprecationWarning, - stacklevel=2, - ) + handler_wants_subscription_message = _resolve_topic_event_type(func) self._servicer.register_topic( pubsub_name, topic, From feb4373f2d3c5dacd5ecc63273a4483104d45dfb Mon Sep 17 00:00:00 2001 From: Sai Kishore Punagani <63619246+saikishore-p@users.noreply.github.com> Date: Sun, 13 Sep 2026 00:04:00 -0400 Subject: [PATCH 2/2] feat: add asyncio support to the gRPC app extension (dapr.ext.grpc.aio) dapr.ext.grpc.App runs on grpc.server() with a thread pool, so handlers must be synchronous. Applications built on asyncio have had to vendor their own callback server to use `async def` handlers. Add dapr.ext.grpc.aio, backed by grpc.aio.server(). It exports the same names as dapr.ext.grpc; swapping the import is the only change an app needs, beyond `async def` handlers and awaiting run()/stop(). _AioCallbackServicer is a sibling of _CallbackServicer over the base extracted in the previous commit - neither subclasses the other, so there is no sync/async mixing in one MRO and no misleading isinstance relationship. It supplies only the gRPC entry points, which await the handler result and the grpc.aio context coroutines. The asyncio servicer covers the full current surface: service invocation, pub/sub including bulk events, input bindings, job events on both the stable and alpha services, and health checks. AppCallbackAlphaServicer is registered on the server, and send_initial_metadata is awaited, since it is a coroutine on the aio context rather than a plain method. Handlers may be plain functions - results are awaited only when awaitable - so register_health_check(lambda: None) keeps working. The server is built on the first run()/start() call rather than in __init__, because grpc.aio.server() binds to whichever event loop is current when it is called; add_external_service() therefore queues its registration and replays it at creation. Adds unit tests including an end-to-end suite over a real grpc.aio server and parity guards asserting every sync RPC is mirrored as a coroutine, plus the invoke-simple-async and pubsub-simple-async examples. The async client (dapr.aio.clients.DaprClient) already exists; this closes the remaining server-side gap. Closes #695 Signed-off-by: Sai Kishore Punagani <63619246+saikishore-p@users.noreply.github.com> --- AGENTS.md | 2 +- dapr/ext/grpc/AGENTS.md | 68 ++- dapr/ext/grpc/README.md | 29 +- dapr/ext/grpc/aio/__init__.py | 38 ++ dapr/ext/grpc/aio/_health_servicer.py | 34 ++ dapr/ext/grpc/aio/_servicer.py | 185 +++++++ dapr/ext/grpc/aio/app.py | 461 ++++++++++++++++++ examples/AGENTS.md | 2 + examples/invoke-simple-async/README.md | 109 +++++ examples/invoke-simple-async/invoke-caller.py | 41 ++ .../invoke-simple-async/invoke-receiver.py | 34 ++ examples/invoke-simple-async/requirements.txt | 1 + examples/pubsub-simple-async/README.md | 112 +++++ examples/pubsub-simple-async/publisher.py | 69 +++ examples/pubsub-simple-async/requirements.txt | 1 + examples/pubsub-simple-async/subscriber.py | 60 +++ examples/pubsub-streaming-async/README.md | 4 + tests/examples/test_invoke_simple_async.py | 32 ++ tests/examples/test_pubsub_simple_async.py | 37 ++ tests/ext/grpc/aio/__init__.py | 0 tests/ext/grpc/aio/test_app.py | 383 +++++++++++++++ tests/ext/grpc/aio/test_health_servicer.py | 55 +++ tests/ext/grpc/aio/test_server.py | 415 ++++++++++++++++ tests/ext/grpc/aio/test_servicer.py | 457 +++++++++++++++++ 24 files changed, 2620 insertions(+), 9 deletions(-) create mode 100644 dapr/ext/grpc/aio/__init__.py create mode 100644 dapr/ext/grpc/aio/_health_servicer.py create mode 100644 dapr/ext/grpc/aio/_servicer.py create mode 100644 dapr/ext/grpc/aio/app.py create mode 100644 examples/invoke-simple-async/README.md create mode 100644 examples/invoke-simple-async/invoke-caller.py create mode 100644 examples/invoke-simple-async/invoke-receiver.py create mode 100644 examples/invoke-simple-async/requirements.txt create mode 100644 examples/pubsub-simple-async/README.md create mode 100644 examples/pubsub-simple-async/publisher.py create mode 100644 examples/pubsub-simple-async/requirements.txt create mode 100644 examples/pubsub-simple-async/subscriber.py create mode 100644 tests/examples/test_invoke_simple_async.py create mode 100644 tests/examples/test_pubsub_simple_async.py create mode 100644 tests/ext/grpc/aio/__init__.py create mode 100644 tests/ext/grpc/aio/test_app.py create mode 100644 tests/ext/grpc/aio/test_health_servicer.py create mode 100644 tests/ext/grpc/aio/test_server.py create mode 100644 tests/ext/grpc/aio/test_servicer.py diff --git a/AGENTS.md b/AGENTS.md index 6c174db60..336de35c7 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -57,7 +57,7 @@ Extensions are bundled into the core `dapr` wheel and exposed as installable ext | Extra | Import path | Purpose | Active development | |-------|-------------|---------|--------------------| | `dapr[workflow]` | `dapr.ext.workflow` | Durable workflow orchestration (durabletask vendored internally) | **High**, major focus area | -| `dapr[grpc]` | `dapr.ext.grpc` | gRPC server for Dapr callbacks (methods, pub/sub, bindings, jobs) | Moderate | +| `dapr[grpc]` | `dapr.ext.grpc`, `dapr.ext.grpc.aio` | gRPC server for Dapr callbacks (methods, pub/sub, bindings, jobs), sync and asyncio | Moderate | | `dapr[fastapi]` | `dapr.ext.fastapi` | FastAPI integration for pub/sub and actors | Moderate | | `dapr[flask]` | `dapr.ext.flask` | Flask integration for pub/sub and actors (legacy `flask_dapr` import path is a deprecated shim) | Low | | `dapr[langgraph]` | `dapr.ext.langgraph` | LangGraph checkpoint persistence to Dapr state store | Moderate | diff --git a/dapr/ext/grpc/AGENTS.md b/dapr/ext/grpc/AGENTS.md index 6f29f0a29..e3ac5485b 100644 --- a/dapr/ext/grpc/AGENTS.md +++ b/dapr/ext/grpc/AGENTS.md @@ -7,16 +7,26 @@ The gRPC extension provides a **server-side callback framework** for Dapr applic ``` dapr/ext/grpc/ ├── __init__.py # Public API exports -├── app.py # App class — main entry point -├── _servicer.py # _CallbackServicer — internal routing -├── _health_servicer.py # _HealthCheckServicer +├── app.py # App class — main entry point (sync) +├── _servicer.py # _CallbackServicerBase (shared) + _CallbackServicer (sync) +├── _health_servicer.py # _HealthCheckServicerBase (shared) + _HealthCheckServicer +├── aio/ # asyncio-native variant, same public surface +│ ├── __init__.py # Public API exports +│ ├── app.py # App class backed by grpc.aio +│ ├── _servicer.py # _AioCallbackServicer — async gRPC entry points +│ └── _health_servicer.py # _AioHealthCheckServicer └── py.typed tests/ext/grpc/ ├── test_app.py # Decorator registration tests ├── test_servicer.py # Routing, handlers, bulk events ├── test_health_servicer.py # Health check tests -└── test_topic_event_response.py # Response status tests +├── test_topic_event_response.py # Response status tests +└── aio/ + ├── test_app.py # Decorator registration, lazy server creation, lifecycle + ├── test_servicer.py # Async handlers, sync/aio parity guards + ├── test_health_servicer.py # Async and sync health check callbacks + └── test_server.py # End-to-end over a real grpc.aio server ``` Installed via the `grpc` extra on core dapr: `pip install "dapr[grpc]"`. @@ -42,6 +52,8 @@ from dapr.ext.grpc import ( Note: `InvokeMethodRequest`, `InvokeMethodResponse`, `BindingRequest`, `TopicEventResponse`, `Job`, `JobEvent`, and failure policies are actually defined in the core SDK (`dapr/clients/grpc/`) and re-exported here. +`dapr.ext.grpc.aio` exports the same names. Swapping the import is the only change an app needs, beyond `async def` handlers and awaiting `run()`/`stop()`. + ## App class (`app.py`) The central entry point. Creates a gRPC server and provides decorators for handler registration. @@ -88,9 +100,44 @@ app.register_health_check(lambda: None) # Not a decorator — direct registrati - `TopicEventResponse('success'|'retry'|'drop')` → explicit status - `None` → defaults to SUCCESS +## Asyncio app (`aio/`) + +`dapr.ext.grpc.aio.App` is the asyncio counterpart, backed by `grpc.aio.server()`. Same decorators, same registration semantics, same wire behavior. + +```python +import asyncio +from dapr.ext.grpc.aio import App + +app = App() + +@app.subscribe(pubsub_name='pubsub', topic='orders') +async def handle_event(event: SubscriptionMessage) -> TopicEventResponse: + ... + +asyncio.run(app.run(3010)) +``` + +Differences from the synchronous `App`, all of them forced by the async runtime: + +- **`run()` and `stop()` are coroutines.** `stop(grace=None)` takes a grace period (the sync `stop()` is always immediate). There is no `__del__` hook, because a coroutine cannot be awaited from one — `run()` instead stops the server in a `finally`, so a cancelled app does not leave its port bound. +- **`start()` exists** as a non-blocking alternative to `run()`, for serving the app alongside other work on the same loop (e.g. from an ASGI lifespan handler). `run()` is `start()` plus `wait_for_termination()`. +- **The server is built lazily**, on the first `run()`/`start()` call, not in `__init__`. `grpc.aio.server()` binds to whichever event loop is current when it is called, so building it in `__init__` would attach it to the wrong loop. `add_external_service()` therefore queues its registration and replays it when the server is created, and raises if called once the app is running. +- **`start()` and `stop()` are serialised** by an `asyncio.Lock`. grpc.aio segfaults if a `stop()` call is *concurrently in flight* with a `start()` call, so the two must never overlap; the lock also means a restart waits for an in-progress drain rather than binding a second server to the same port. (Stopping a server whose own `start()` has already unwound — the cleanup path in `_start`'s `except` — is sequential, not concurrent, and is safe. `stop()` on a server that never started returns cleanly; on one whose `start()` was *cancelled part-way* it raises `InvalidStateError`, which that path suppresses.) +- **The lock is built per running loop**, not in `__init__`: an `asyncio.Lock` binds to the loop of its first *contended* acquire, so one built at construction raises "bound to a different event loop" on a second `asyncio.run()` — and silently stops excluding anything before that, because the uncontended path returns before the loop check. +- **Decorators return the handler**, so the decorated name stays bound. The sync decorators return `None`. +- **Handlers may be plain functions.** Results are awaited only when awaitable, so `register_health_check(lambda: None)` still works. A plain handler runs inline on the event loop and must not block. + +### Sharing with the sync implementation + +`_CallbackServicerBase` (in `_servicer.py`) holds everything that does not invoke a user handler: the handler registries, topic routing, and the request→event translation. `_CallbackServicer` and `_AioCallbackServicer` are **siblings** on top of it — neither subclasses the other — and each supplies only the gRPC entry points. + +This keeps the churn-prone routing logic (`_get_topic_callback`, `register_topic`, the bulk entry builders) in one place while leaving the two servicers free to differ where they must. `tests/ext/grpc/aio/test_servicer.py::AsyncParityTests` enforces the arrangement: every RPC the sync servicer implements must be mirrored on the aio servicer as a coroutine function, and the registration helpers must stay shared rather than be reimplemented. + +`_HealthCheckServicerBase` splits the health servicer the same way: registration in the base, the gRPC entry point in each sibling. + ## Internal routing (`_servicer.py`) -`_CallbackServicer` implements `AppCallbackServicer` + `AppCallbackAlphaServicer` gRPC service interfaces. It maintains internal registries: +`_CallbackServicerBase` implements `AppCallbackServicer` + `AppCallbackAlphaServicer` gRPC service interfaces; `_CallbackServicer` adds the synchronous entry points. It maintains internal registries: - `_invoke_method_map` — method name → handler - `_topic_map` — topic key → handler @@ -124,6 +171,13 @@ app.register_health_check(lambda: None) # Not a decorator — direct registrati uv run python -m unittest discover -v ./tests/ext/grpc ``` +`unittest discover` covers the whole tree including `aio/` — `IsolatedAsyncioTestCase` and +`subTest` are both native unittest. pytest runs the same tests: + +```bash +uv run pytest ./tests/ext/grpc +``` + Test patterns: - `test_app.py` — decorator registration, health check registration - `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 @@ -132,8 +186,8 @@ Test patterns: ## Key details -- **Synchronous only**: Uses `grpc.server()` with `ThreadPoolExecutor(10)`. No async handler support. +- **Sync app threading**: `dapr.ext.grpc.App` uses `grpc.server()` with `ThreadPoolExecutor(10)`. For `async def` handlers use `dapr.ext.grpc.aio.App`, which serves on a `grpc.aio` event loop instead. - **Default port**: 3010 (from `dapr.conf.global_settings.GRPC_APP_PORT`) -- **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=...)`.) +- **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 `_CallbackServicerBase.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) diff --git a/dapr/ext/grpc/README.md b/dapr/ext/grpc/README.md index ea92b1bf2..044412aff 100644 --- a/dapr/ext/grpc/README.md +++ b/dapr/ext/grpc/README.md @@ -12,5 +12,32 @@ pip install "dapr[grpc]" from dapr.ext.grpc import App ``` +An asyncio-native app backed by a `grpc.aio` server is available under +`dapr.ext.grpc.aio`. It exposes the same names and decorators; handlers may be +`async def` and are awaited, and `run()`/`stop()` are coroutines: + +```python +import asyncio + +from dapr.ext.grpc.aio import App, InvokeMethodRequest, InvokeMethodResponse + +app = App() + + +@app.method(name='my-method') +async def my_method(request: InvokeMethodRequest) -> InvokeMethodResponse: + ... + + +asyncio.run(app.run(50051)) +``` + +Plain (non-async) handlers are still accepted, but they run inline on the event +loop, so they must not block. See the [`invoke-simple-async`][invoke-async] and +[`pubsub-simple-async`][pubsub-async] examples. + See the root [README](../../../README.md) for migration steps from the legacy -`dapr-ext-grpc` distribution. \ No newline at end of file +`dapr-ext-grpc` distribution. + +[invoke-async]: ../../../examples/invoke-simple-async +[pubsub-async]: ../../../examples/pubsub-simple-async diff --git a/dapr/ext/grpc/aio/__init__.py b/dapr/ext/grpc/aio/__init__.py new file mode 100644 index 000000000..7b08dc988 --- /dev/null +++ b/dapr/ext/grpc/aio/__init__.py @@ -0,0 +1,38 @@ +# -*- coding: utf-8 -*- + +""" +Copyright 2025 The Dapr Authors +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +""" + +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 + +# No cloudevents ImportError guard here: importing this subpackage imports the parent +# dapr.ext.grpc first, which already raises the actionable error. +from dapr.ext.grpc.aio.app import App, Rule # type:ignore + +__all__ = [ + 'App', + 'Rule', + 'SubscriptionMessage', + 'InvokeMethodRequest', + 'InvokeMethodResponse', + 'BindingRequest', + 'TopicEventResponse', + 'Job', + 'JobEvent', + 'FailurePolicy', + 'DropFailurePolicy', + 'ConstantFailurePolicy', +] diff --git a/dapr/ext/grpc/aio/_health_servicer.py b/dapr/ext/grpc/aio/_health_servicer.py new file mode 100644 index 000000000..7f8dbe4b1 --- /dev/null +++ b/dapr/ext/grpc/aio/_health_servicer.py @@ -0,0 +1,34 @@ +# -*- coding: utf-8 -*- + +""" +Copyright 2025 The Dapr Authors +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +""" + +from dapr.ext.grpc._health_servicer import _HealthCheckServicerBase +from dapr.ext.grpc.aio._servicer import _needs_await +from dapr.proto.runtime.v1.appcallback_pb2 import HealthCheckResponse + + +class _AioHealthCheckServicer(_HealthCheckServicerBase): + """The asyncio-native implementation of HealthCheck Server. + + Shares registration with the synchronous servicer via their common base; only the gRPC + entry point differs, awaiting the callback result. + """ + + async def HealthCheck(self, request, context): + """Health check.""" + health_check = self._require_health_check_cb(context) + result = health_check() + if _needs_await(result): + await result + return HealthCheckResponse() diff --git a/dapr/ext/grpc/aio/_servicer.py b/dapr/ext/grpc/aio/_servicer.py new file mode 100644 index 000000000..51202f839 --- /dev/null +++ b/dapr/ext/grpc/aio/_servicer.py @@ -0,0 +1,185 @@ +# -*- coding: utf-8 -*- + +""" +Copyright 2025 The Dapr Authors +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +""" + +import inspect +from typing import Any, Awaitable, Protocol, TypeGuard + +from dapr.ext.grpc._servicer import _CallbackServicerBase, _ServicerContext +from dapr.proto import appcallback_v1, common_v1 +from dapr.proto.common.v1.common_pb2 import InvokeRequest +from dapr.proto.runtime.v1.appcallback_pb2 import ( + BindingEventRequest, + JobEventRequest, + TopicEventBulkRequest, + TopicEventBulkResponse, + TopicEventRequest, +) + + +class _AioServicerContext(_ServicerContext, Protocol): + """The ``grpc.aio`` servicer context surface this servicer relies on. + + Extends the shared protocol with ``send_initial_metadata``, which is a coroutine here but + a plain method on the synchronous context — the difference that makes a single shared + protocol impossible. + """ + + async def send_initial_metadata(self, initial_metadata: Any) -> None: ... + + +def _needs_await(result: Any) -> TypeGuard[Awaitable[Any]]: + """True if a handler's return value still has to be awaited. + + The test is on the returned value, not on the handler: an ``async def`` handler yields a + coroutine, while a plain function's value is already final. Plain handlers are accepted so + trivial ones (and ``register_health_check(lambda: None)``) keep working, but they run + inline on the event loop and must not block. Deliberately not a coroutine function - every + RPC calls this, and wrapping a two-line test would allocate a coroutine per request. + + Returns: + bool: True if ``result`` is awaitable, narrowed for the type checker. + """ + return inspect.isawaitable(result) + + +class _AioCallbackServicer(_CallbackServicerBase): + """The asyncio-native implementation of the AppCallback Server. + + Shares its handler registries, topic routing, and request translation with the + synchronous :class:`dapr.ext.grpc._servicer._CallbackServicer` via their common base. + Only the gRPC entry points are redefined here: they await the handler result and the + ``grpc.aio`` context coroutines. + """ + + async def OnInvoke(self, request: InvokeRequest, context: _AioServicerContext): + """Invokes service method with InvokeRequest.""" + if request.method not in self._invoke_method_map: + raise self._unimplemented(context, f'{request.method} method not implemented!') + + req = self._build_invoke_request(request, context) + resp = self._invoke_method_map[request.method](req) + if _needs_await(resp): + resp = await resp + + if not resp: + return common_v1.InvokeResponse() + + resp_data = self._to_invoke_response_data(resp, context, request.method) + + headers = resp_data.get_headers() + if len(headers) > 0: + # grpc.aio's ServicerContext.send_initial_metadata is a coroutine, unlike the + # synchronous context's method of the same name. + await context.send_initial_metadata(headers) + + return self._to_invoke_response(resp_data) + + async def ListTopicSubscriptions(self, request, context: _AioServicerContext): + """Lists all topics subscribed by this app.""" + return appcallback_v1.ListTopicSubscriptionsResponse(subscriptions=self._registered_topics) + + async def OnTopicEvent(self, request: TopicEventRequest, context: _AioServicerContext): + """Subscribes events from Pubsub.""" + cb = self._get_topic_callback(request.pubsub_name, request.topic, request.path) + if cb is None: + raise self._unimplemented(context, f'topic {request.topic} is not implemented!') + + event = self._build_topic_event(cb, request, dict(context.invocation_metadata())) + response = cb(event) + if _needs_await(response): + response = await response + + return self._to_topic_event_response(response) + + async def ListInputBindings(self, request, context: _AioServicerContext): + """Lists all input bindings subscribed by this app.""" + return appcallback_v1.ListInputBindingsResponse(bindings=self._registered_bindings) + + async def OnBindingEvent(self, request: BindingEventRequest, context: _AioServicerContext): + """Listens events from the input bindings + User application can save the states or send the events to the output + bindings optionally by returning BindingEventResponse. + """ + if request.name not in self._binding_map: + raise self._unimplemented(context, f'{request.name} binding not implemented!') + + req = self._build_binding_request(request, context) + binding_result = self._binding_map[request.name](req) + if _needs_await(binding_result): + await binding_result + + # TODO: support output bindings options + return appcallback_v1.BindingEventResponse() + + async def _handle_job_event(self, request: JobEventRequest, context: _AioServicerContext): + """Routes a job event to the handler registered for its job name.""" + job_event = self._build_job_event(request, context) + job_result = self._job_event_map[request.name](job_event) + if _needs_await(job_result): + await job_result + return appcallback_v1.JobEventResponse() + + async def OnJobEvent(self, request: JobEventRequest, context: _AioServicerContext): + """Handles job events on the stable AppCallback service.""" + return await self._handle_job_event(request, context) + + async def OnJobEventAlpha1(self, request: JobEventRequest, context: _AioServicerContext): + """Handles job events on the deprecated AppCallbackAlpha service.""" + return await self._handle_job_event(request, context) + + async def _handle_bulk_topic_event( + self, request: TopicEventBulkRequest, context: _AioServicerContext + ) -> TopicEventBulkResponse: + """Process bulk topic event request - routes each entry to the appropriate topic handler.""" + cb = self._get_topic_callback(request.pubsub_name, request.topic, request.path) + if cb is None: + raise self._unimplemented(context, f'bulk topic {request.topic} is not implemented!') + + use_legacy_event = self._topic_legacy_event.get(cb, True) + invocation_metadata = dict(context.invocation_metadata()) + + # Entries are handled one at a time, matching the synchronous servicer's delivery + # order. Gathering them would let a batch overtake itself, so concurrency here is a + # deliberate non-goal; the per-RPC path is where the event loop pays off. + statuses = [] + for entry in request.entries: + entry_id = entry.entry_id + try: + event = self._bulk_entry_event( + entry, request, use_legacy_event, invocation_metadata + ) + entry_response = cb(event) + if _needs_await(entry_response): + entry_response = await entry_response + status = self._bulk_entry_status(entry_response) + except Exception: + status = appcallback_v1.TopicEventResponse.TopicEventResponseStatus.RETRY + statuses.append( + appcallback_v1.TopicEventBulkResponseEntry(entry_id=entry_id, status=status) + ) + return appcallback_v1.TopicEventBulkResponse(statuses=statuses) + + async def OnBulkTopicEvent(self, request: TopicEventBulkRequest, context: _AioServicerContext): + """Subscribes bulk events from Pubsub""" + return await self._handle_bulk_topic_event(request, context) + + async def OnBulkTopicEventAlpha1( + self, request: TopicEventBulkRequest, context: _AioServicerContext + ): + """Subscribes bulk events from Pubsub. + Deprecated: Use OnBulkTopicEvent instead. + """ + self._warn_bulk_alpha1_deprecated() + return await self._handle_bulk_topic_event(request, context) diff --git a/dapr/ext/grpc/aio/app.py b/dapr/ext/grpc/aio/app.py new file mode 100644 index 000000000..4b0276e49 --- /dev/null +++ b/dapr/ext/grpc/aio/app.py @@ -0,0 +1,461 @@ +# -*- coding: utf-8 -*- + +""" +Copyright 2025 The Dapr Authors +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +""" + +import asyncio +import contextlib +import logging +from typing import Any, Callable, Dict, List, Optional, Tuple + +import grpc.aio + +from dapr.conf import settings +from dapr.ext.grpc._health_servicer import AsyncHealthCheckCallable +from dapr.ext.grpc._servicer import Rule # type: ignore +from dapr.ext.grpc.aio._health_servicer import _AioHealthCheckServicer # type: ignore +from dapr.ext.grpc.aio._servicer import _AioCallbackServicer # type: ignore +from dapr.ext.grpc.app import _resolve_topic_event_type +from dapr.proto import appcallback_service_v1 + +logger = logging.getLogger(__name__) + +ExternalServiceRegistration = Tuple[Callable[[Any, grpc.aio.Server], None], Any] + + +class App: + """App object implements a Dapr application callback which can interact with Dapr runtime. + + This is the asyncio-native counterpart of :class:`dapr.ext.grpc.App`: it is backed by a + ``grpc.aio`` server, so handlers may be ``async def`` and are awaited. The decorators take + the same arguments as the synchronous app's, but return the handler so the decorated name + stays bound (the synchronous decorators return ``None``). + + Differences from the synchronous app, all forced by the async runtime: + + * :meth:`run` and :meth:`stop` are coroutines, and :meth:`stop` takes a grace period. + * :meth:`start` is a non-blocking alternative to :meth:`run`, for serving alongside other + work on the same loop. + * The gRPC server is built on the first :meth:`run`/:meth:`start` rather than in + ``__init__``, because ``grpc.aio.server()`` binds to the loop current at creation. + :meth:`add_external_service` must therefore be called **before** the app is started; it + raises afterwards, where the synchronous app accepts it at any time. + + You can create a :class:`App` instance in your main module: + + import asyncio + from dapr.ext.grpc.aio import App + + app = App() + asyncio.run(app.run(50051)) + """ + + def __init__(self, max_grpc_message_length: Optional[int] = None, **kwargs): + """Inits App object. The gRPC server itself is created when the app is started. + + Args: + max_grpc_message_length (int, optional): The maximum grpc send and receive + message length in bytes. Only used when kwargs are not set. When this + argument is omitted, the env var + ``DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES`` is consulted to set the + receive limit (matches the Java SDK property of the same name). + kwargs: arguments to grpc.aio.server() + """ + self._servicer = _AioCallbackServicer() + self._health_check_servicer = _AioHealthCheckServicer() + self._server: Optional[grpc.aio.Server] = None + self._listen_port: Optional[int] = None + self._abandoned_servers: List[grpc.aio.Server] = [] + self._abandoned_ports: set[int] = set() + self._lifecycle_lock: Optional[asyncio.Lock] = None + self._lifecycle_loop: Optional[asyncio.AbstractEventLoop] = None + self._external_services: List[ExternalServiceRegistration] = [] + + if kwargs: + self._server_kwargs: Dict[str, Any] = kwargs + return + + options = [] + if max_grpc_message_length is not None: + options = [ + ('grpc.max_send_message_length', max_grpc_message_length), + ('grpc.max_receive_message_length', max_grpc_message_length), + ] + elif settings.DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES: + options = [ + ( + 'grpc.max_receive_message_length', + settings.DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES, + ), + ] + self._server_kwargs = {'options': options} + + def _lock_for_running_loop(self) -> asyncio.Lock: + """Returns the start/stop lock, rebuilding it if the running loop has changed. + + Serialises start and stop, which grpc.aio cannot have in flight at the same time. + An App is meant to be driven from one event loop; this guard catches the common + mistake, but it is best effort — two loops racing their very first ``start()`` from + different threads can each build a lock before either publishes one. + The lock cannot be built in ``__init__``: an :class:`asyncio.Lock` binds to the loop + of its first *contended* acquire, so one built there would raise "bound to a different + event loop" on a second ``asyncio.run()`` — and, because the uncontended path returns + before that check, would silently stop excluding anything in between. + """ + loop = asyncio.get_running_loop() + lock = self._lifecycle_lock + owning_loop = self._lifecycle_loop + if lock is not None and owning_loop is loop: + return lock + + # Handing out a fresh lock for a foreign loop would destroy the exclusion entirely, + # so a live server may only be driven from the loop that started it. + if self._server is not None: + raise RuntimeError( + 'app gRPC server was started on a different event loop. Stop it from that ' + 'loop, or — if that loop is already closed — call stop() from this one to ' + 'release it.' + ) + + lock = asyncio.Lock() + self._lifecycle_lock = lock + self._lifecycle_loop = loop + return lock + + def _abandon_if_owning_loop_is_closed(self) -> bool: + """Releases a server whose event loop is gone, warning that its port stays bound. + + Such a server cannot be shut down from here — grpc.aio needs its own loop to drain — + but refusing would trap the App, since stop() is the documented remedy. The handle is + dropped so the App becomes usable again, loudly, because the listener survives for + the life of the process and grpc enables SO_REUSEPORT: restarting on the same port + would bind a *second* server and the kernel would split callbacks between them. + + Returns: + bool: True if a server was abandoned and the caller should stop. + """ + owning_loop = self._lifecycle_loop + if self._server is None or owning_loop is None or not owning_loop.is_closed(): + return False + + logger.warning( + 'Cannot stop the app gRPC server: the event loop it was started on is closed. ' + 'Its listener stays bound until the process exits. Restarting on the same port ' + 'would bind a second server alongside it, so choose a different port.' + ) + # Parked rather than dropped: releasing the last reference fires grpc's + # Server.__del__ against the closed loop, printing an "Event loop is closed" + # traceback attributed to this file. Holding it defers that to interpreter exit. + self._abandoned_servers.append(self._server) + if self._listen_port is not None: + self._abandoned_ports.add(self._listen_port) + self._server = None + self._lifecycle_lock = None + self._lifecycle_loop = None + return True + + def _create_server(self) -> grpc.aio.Server: + """Builds the gRPC server and registers every servicer on it. + + ``grpc.aio.server()`` binds to whichever event loop is current when it is called, so + the server cannot be built in ``__init__`` the way the synchronous App does — it has + to be created inside the running loop that will serve requests. + """ + server = grpc.aio.server(**self._server_kwargs) + appcallback_service_v1.add_AppCallbackServicer_to_server(self._servicer, server) + appcallback_service_v1.add_AppCallbackAlphaServicer_to_server(self._servicer, server) + appcallback_service_v1.add_AppCallbackHealthCheckServicer_to_server( + self._health_check_servicer, server + ) + for servicer_callback, external_servicer in self._external_services: + servicer_callback(external_servicer, server) + return server + + async def _start( + self, app_port: Optional[int] = None, listen_address: Optional[str] = None + ) -> grpc.aio.Server: + async with self._lock_for_running_loop(): + if self._server is not None: + raise RuntimeError('app gRPC server is already running') + + if app_port is None: + app_port = settings.GRPC_APP_PORT + listen_addr = f'{listen_address if listen_address else "[::]"}:{app_port}' + + if app_port in self._abandoned_ports: + logger.warning( + 'Starting on port %s, which an abandoned server still holds. grpc enables ' + 'SO_REUSEPORT, so both will bind and callbacks will be split between them.', + app_port, + ) + + server = self._create_server() + try: + # add_insecure_port raises on grpc.aio when the port is taken, so it belongs + # inside the guard: otherwise a bind failure strands a fully built server. + server.add_insecure_port(listen_addr) + # Published before the await so add_external_service(), which is sync and + # takes no lock, rejects registrations for the whole of startup. + self._server = server + self._listen_port = app_port + await server.start() + except BaseException: + self._server = None + # start() has already unwound here, so this is sequential cleanup, not the + # concurrent overlap the lock prevents. stop() is a no-op on a server that + # never started, but raises InvalidStateError on one whose start() was + # cancelled part-way. CancelledError is suppressed alongside normal errors + # because the original failure is re-raised below; KeyboardInterrupt and + # SystemExit deliberately are not. + with contextlib.suppress(Exception, asyncio.CancelledError): + await server.stop(None) + raise + return server + + async def start( + self, app_port: Optional[int] = None, listen_address: Optional[str] = None + ) -> None: + """Starts app gRPC server and returns once it is accepting requests. + + Use this instead of :meth:`run` to serve the app alongside other work on the same + event loop, for example from an ASGI lifespan handler. + + Args: + app_port (int, optional): The port on which to listen for incoming gRPC calls. + Defaults to settings.GRPC_APP_PORT. + listen_address (str, optional): The IP address on which to listen for incoming gRPC + calls. Defaults to [::] (all IP addresses). + """ + await self._start(app_port, listen_address) + + async def run( + self, app_port: Optional[int] = None, listen_address: Optional[str] = None + ) -> None: + """Starts app gRPC server and waits until :class:`App`.stop() is called. + + Args: + app_port (int, optional): The port on which to listen for incoming gRPC calls. + Defaults to settings.GRPC_APP_PORT. + listen_address (str, optional): The IP address on which to listen for incoming gRPC + calls. Defaults to [::] (all IP addresses). + """ + server = await self._start(app_port, listen_address) + try: + await server.wait_for_termination() + finally: + # Guarded on identity: if the app was stopped and restarted while this was + # unwinding, self._server is a different server and is not ours to tear down. + # The synchronous App relies on __del__ to stop its server; a coroutine cannot be + # awaited from one, so cancellation (Ctrl-C, a wait_for timeout, a TaskGroup + # teardown) would otherwise leave the listener bound and the App unrestartable. + if self._server is server: + await self.stop() + + async def stop(self, grace: Optional[float] = None) -> None: + """Stops app server, letting in-flight requests finish within the grace period. + + Args: + grace (float, optional): Seconds to wait for in-flight requests before cancelling + them. Defaults to None, which cancels them immediately. + """ + if self._abandon_if_owning_loop_is_closed(): + return + + # Waits for an in-flight start() rather than racing it, and holds the lock for the + # whole drain so a restart cannot bind a second server to the same port. + async with self._lock_for_running_loop(): + server = self._server + if server is None: + return + try: + await server.stop(grace) + except asyncio.CancelledError: + # A graceful drain that is cut short still has to release the socket, or the + # next start() fails on bind. Force an immediate stop, then report it gone. + with contextlib.suppress(Exception, asyncio.CancelledError): + await server.stop(None) + self._server = None + raise + # Only cleared once the server is actually down. If stop() failed for any other + # reason the server's state is unknown, and claiming it stopped would make the + # next start() fail on bind rather than with an accurate "already running". + self._server = None + + def add_external_service( + self, + servicer_callback: Callable[[Any, grpc.aio.Server], None], + external_servicer: Any, + ) -> None: + """Adds an external gRPC service to the same server. + + The registration is replayed when the server is built by :meth:`run` or :meth:`start`, + so it must be called before the app is started. A ``grpc.aio`` server does not accept + new services once it is serving, so registering late raises rather than being dropped. + + Raises: + RuntimeError: if the app has already been started. + """ + if self._server is not None: + raise RuntimeError( + 'add_external_service must be called before the app is started; ' + 'a running gRPC server cannot accept new services' + ) + self._external_services.append((servicer_callback, external_servicer)) + + def register_health_check(self, health_check_callback: AsyncHealthCheckCallable) -> None: + """Adds a health check callback + + The below example adds a basic health check to check Dapr gRPC is running + + app.register_health_check(lambda: None) + """ + self._health_check_servicer.register_health_check(health_check_callback) + + def method(self, name: str) -> Callable: + """A decorator that is used to register the method for the service invocation. + + Return JSON formatted data response:: + + @app.method('start') + async def start(request: InvokeMethodRequest): + + ... + + return json.dumps() + + Return Protocol buffer response:: + + @app.method('start') + async def start(request: InvokeMethodRequest): + + ... + + return CustomProtoResponse(data='hello world') + + + Specify Response header:: + + @app.method('start') + async def start(request: InvokeMethodRequest): + + ... + + resp = InvokeMethodResponse('hello world', 'text/plain') + resp.headers = ('key', 'value') + + return resp + + Args: + name (str): name of invoked method + """ + + def decorator(func: Callable) -> Callable: + self._servicer.register_method(name, func) + return func + + return decorator + + def subscribe( + self, + pubsub_name: str, + topic: str, + metadata: Optional[Dict[str, str]] = None, + dead_letter_topic: Optional[str] = None, + rule: Optional[Rule] = None, + disable_topic_validation: Optional[bool] = False, + ) -> Callable: + """A decorator that is used to register the subscribing topic method. + + The event type the handler receives is inferred from its annotation: annotate the + event parameter with :class:`dapr.ext.grpc.SubscriptionMessage` to receive that type. + Unannotated (or otherwise-annotated) handlers receive the deprecated + ``cloudevents.sdk.event.v1.Event`` and trigger a :class:`DeprecationWarning`. + + The below example registers 'topic' subscription topic and pass custom + metadata to pubsub component:: + + from dapr.ext.grpc.aio import SubscriptionMessage + + @app.subscribe('pubsub_name', 'topic', metadata={'session-id': 'session-id-value'}) + async def topic(event: SubscriptionMessage) -> None: + ... + + Args: + pubsub_name (str): the name of the pubsub component + topic (str): the topic name which is subscribed + metadata (dict, optional): metadata which will be passed to pubsub component + during initialization + dead_letter_topic (str, optional): the dead letter topic name for the subscription + """ + + def decorator(func: Callable) -> Callable: + handler_wants_subscription_message = _resolve_topic_event_type(func) + self._servicer.register_topic( + pubsub_name, + topic, + func, + metadata, + dead_letter_topic, + rule, + disable_topic_validation, + legacy_cloudevent=not handler_wants_subscription_message, + ) + return func + + return decorator + + def binding(self, name: str) -> Callable: + """A decorator that is used to register input binding. + + The below registers input binding which this application subscribes: + + @app.binding('input') + async def input(request: BindingRequest) -> None: + ... + + Args: + name (str): the name of invoked method + """ + + def decorator(func: Callable) -> Callable: + self._servicer.register_binding(name, func) + return func + + return decorator + + def job_event(self, name: str) -> Callable: + """A decorator that is used to register job event handler. + + This decorator registers a handler for job events triggered by the Dapr scheduler. + The handler will be called when a job with the specified name is triggered. + + The below registers a job event handler for jobs named 'my-job': + + from dapr.ext.grpc.aio import JobEvent + + @app.job_event('my-job') + async def handle_my_job(job_event: JobEvent) -> None: + print(f"Job {job_event.name} triggered") + data_str = job_event.get_data_as_string() + print(f"Job data: {data_str}") + # Process the job... + + Args: + name (str): the name of the job to handle events for + """ + + def decorator(func: Callable) -> Callable: + self._servicer.register_job_event(name, func) + return func + + return decorator diff --git a/examples/AGENTS.md b/examples/AGENTS.md index f098a2b10..4d8cc47d9 100644 --- a/examples/AGENTS.md +++ b/examples/AGENTS.md @@ -76,6 +76,7 @@ Common component types used in examples: `state.redis`, `pubsub.redis`, `lock.re | Example | Pattern | SDK packages | Has components | |---------|---------|-------------|----------------| | `invoke-simple` | Client-server (receiver/caller) | `dapr[grpc]` | No | +| `invoke-simple-async` | Client-server, asyncio app + async client | `dapr[grpc]` | No | | `invoke-custom-data` | Client-server (protobuf) | `dapr[grpc]` | No | | `invoke-http` | Client-server (Flask) | `dapr`, Flask | No | | `invoke-binding` | Client with bindings | `dapr[grpc]` | Yes | @@ -85,6 +86,7 @@ Common component types used in examples: `state.redis`, `pubsub.redis`, `lock.re | Example | Pattern | SDK packages | Has components | |---------|---------|-------------|----------------| | `pubsub-simple` | Client-server (publisher/subscriber) | `dapr[grpc]` | No | +| `pubsub-simple-async` | Client-server, asyncio subscriber app | `dapr[grpc]` | No | | `pubsub-streaming` | Streaming pub/sub | `dapr` (base only) | No | | `pubsub-streaming-async` | Async streaming pub/sub | `dapr` (base only) | No | diff --git a/examples/invoke-simple-async/README.md b/examples/invoke-simple-async/README.md new file mode 100644 index 000000000..22c0741eb --- /dev/null +++ b/examples/invoke-simple-async/README.md @@ -0,0 +1,109 @@ +# Example - Invoke a service with an asyncio gRPC app + +This example is the asyncio counterpart of [`invoke-simple`](../invoke-simple). It uses +`dapr.ext.grpc.aio.App`, which runs a `grpc.aio` server so handlers can be `async def` and are +awaited. The caller uses the async `dapr.aio.clients.DaprClient` to issue three invocations +concurrently. + +The receiver sleeps for half a second inside the handler. Because the handlers are awaited on +one event loop rather than dispatched to a thread pool, all three invocations are served in +roughly that same half second. + +> **Note:** Make sure to use the latest proto bindings + +## Pre-requisites + +- [Dapr CLI and initialized environment](https://docs.dapr.io/getting-started) +- [Install Python 3.10+](https://www.python.org/downloads/) + +## Install Dapr python-SDK + + + +```bash +pip3 install "dapr[grpc]" +``` + +## Running in self-hosted mode + +Run the following command in a terminal/command-prompt: + + + +```bash +# 1. Start Receiver (expose gRPC server receiver on port 13551) +dapr run --app-id invoke-receiver --app-protocol grpc --app-port 13551 -- python3 invoke-receiver.py +``` + + + +In another terminal/command prompt run: + + + +```bash +# 2. Start Caller (runs three concurrent invocations, then exits) +dapr run --app-id invoke-caller --app-protocol grpc -- python3 invoke-caller.py +``` + + + +## Cleanup + + + +```bash +dapr stop --app-id invoke-receiver +``` + + + +## The difference from `invoke-simple` + +Only the import and the way the app is started change: + +```diff +-from dapr.ext.grpc import App, InvokeMethodRequest, InvokeMethodResponse ++from dapr.ext.grpc.aio import App, InvokeMethodRequest, InvokeMethodResponse + + app = App() + + @app.method(name='my-method') +-def mymethod(request: InvokeMethodRequest) -> InvokeMethodResponse: ++async def mymethod(request: InvokeMethodRequest) -> InvokeMethodResponse: + ... + +-app.run(13551) ++asyncio.run(app.run(13551)) +``` + +`App.run()` and `App.stop()` are coroutines on the asyncio app. Everything else — `@app.method`, +`@app.subscribe`, `@app.binding`, `@app.job_event`, `register_health_check` and +`add_external_service` — take the same arguments as on the synchronous app. The decorators +return the handler, so the decorated name stays bound (the synchronous decorators return +`None`), and may be called at any time. `add_external_service` is the one exception: it must +be called before the app is started. diff --git a/examples/invoke-simple-async/invoke-caller.py b/examples/invoke-simple-async/invoke-caller.py new file mode 100644 index 000000000..d7adc4b93 --- /dev/null +++ b/examples/invoke-simple-async/invoke-caller.py @@ -0,0 +1,41 @@ +# ------------------------------------------------------------ +# Copyright 2025 The Dapr Authors +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ------------------------------------------------------------ + +import asyncio +import json + +from dapr.aio.clients import DaprClient + + +async def invoke_receiver(client: DaprClient, request_id: int) -> None: + req_data = {'id': request_id, 'message': 'hello world'} + + resp = await client.invoke_method( + 'invoke-receiver', + 'my-method', + data=json.dumps(req_data), + ) + + print(resp.content_type, flush=True) + print(resp.text(), flush=True) + + +async def main() -> None: + async with DaprClient() as client: + # All three invocations are in flight at once. Each one sleeps for half a + # second on the receiver, yet the whole gather finishes in about that same + # half second because the async receiver interleaves them. + await asyncio.gather(*(invoke_receiver(client, request_id) for request_id in (1, 2, 3))) + + +asyncio.run(main()) diff --git a/examples/invoke-simple-async/invoke-receiver.py b/examples/invoke-simple-async/invoke-receiver.py new file mode 100644 index 000000000..8d0e3710a --- /dev/null +++ b/examples/invoke-simple-async/invoke-receiver.py @@ -0,0 +1,34 @@ +# ------------------------------------------------------------ +# Copyright 2025 The Dapr Authors +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ------------------------------------------------------------ + +import asyncio + +from dapr.ext.grpc.aio import App, InvokeMethodRequest, InvokeMethodResponse + +app = App() + + +@app.method(name='my-method') +async def mymethod(request: InvokeMethodRequest) -> InvokeMethodResponse: + print(request.metadata, flush=True) + print(request.text(), flush=True) + + # Handlers are awaited on the server's event loop, so awaiting I/O here + # (a database read, an outbound HTTP call) leaves the loop free to serve + # other in-flight invocations instead of tying up a worker thread. + await asyncio.sleep(0.5) + + return InvokeMethodResponse(b'INVOKE_RECEIVED', 'text/plain; charset=UTF-8') + + +asyncio.run(app.run(13551)) diff --git a/examples/invoke-simple-async/requirements.txt b/examples/invoke-simple-async/requirements.txt new file mode 100644 index 000000000..349ce048e --- /dev/null +++ b/examples/invoke-simple-async/requirements.txt @@ -0,0 +1 @@ +dapr[grpc] >= 1.19.0.dev diff --git a/examples/pubsub-simple-async/README.md b/examples/pubsub-simple-async/README.md new file mode 100644 index 000000000..2c806baeb --- /dev/null +++ b/examples/pubsub-simple-async/README.md @@ -0,0 +1,112 @@ +# Example - Publish and subscribe to messages with an asyncio gRPC app + +This example is the asyncio counterpart of [`pubsub-simple`](../pubsub-simple). The subscriber +uses `dapr.ext.grpc.aio.App`, which runs a `grpc.aio` server so `@app.subscribe` handlers can be +`async def` and are awaited. The publisher uses the async `dapr.aio.clients.DaprClient`. + +If you want async pub/sub without running a callback server, see +[`pubsub-streaming-async`](../pubsub-streaming-async) instead — that example uses the client-side +streaming subscription API, where your code pulls messages. This example is the server-side +callback model, where Dapr delivers messages to handlers you register. + +> **Note:** Make sure to use the latest proto bindings + +## Pre-requisites + +- [Dapr CLI and initialized environment](https://docs.dapr.io/getting-started) +- [Install Python 3.10+](https://www.python.org/downloads/) + +## Install Dapr python-SDK + + + +```bash +pip3 install "dapr[grpc]" +``` + +## Run the example + +Run the following command in a terminal/command prompt: + + + +```bash +# 1. Start Subscriber (expose gRPC server receiver on port 13551) +dapr run --app-id python-subscriber --app-protocol grpc --app-port 13551 --enable-app-health-check --app-health-probe-interval 1 -- python3 subscriber.py +``` + + + +In another terminal/command prompt run: + + + +```bash +# 2. Start Publisher +dapr run --app-id python-publisher --app-protocol grpc -- python3 publisher.py +``` + + + +## Cleanup + + + +```bash +dapr stop --app-id python-subscriber +``` + + + +## The difference from `pubsub-simple` + +Only the import and the way the app is started change: + +```diff +-from dapr.ext.grpc import App, SubscriptionMessage +-from dapr.clients.grpc._response import TopicEventResponse ++from dapr.ext.grpc.aio import App, SubscriptionMessage, TopicEventResponse + + app = App() + + @app.subscribe(pubsub_name='pubsub', topic='TOPIC_A') +-def mytopic(event: SubscriptionMessage) -> TopicEventResponse: ++async def mytopic(event: SubscriptionMessage) -> TopicEventResponse: + ... + +-app.run(13551) ++asyncio.run(app.run(13551)) +``` + +Handlers keep returning `TopicEventResponse('success' | 'retry' | 'drop')`, and topic rules, +dead letter topics, bulk delivery and `disable_topic_validation` all behave exactly as they do +on the synchronous app. diff --git a/examples/pubsub-simple-async/publisher.py b/examples/pubsub-simple-async/publisher.py new file mode 100644 index 000000000..8985f61f8 --- /dev/null +++ b/examples/pubsub-simple-async/publisher.py @@ -0,0 +1,69 @@ +# ------------------------------------------------------------ +# Copyright 2025 The Dapr Authors +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ------------------------------------------------------------ + +import asyncio +import json + +from dapr.aio.clients import DaprClient + + +async def main() -> None: + async with DaprClient() as client: + for id in range(1, 4): + req_data = {'id': id, 'message': 'hello world'} + + await client.publish_event( + pubsub_name='pubsub', + topic_name='TOPIC_A', + data=json.dumps(req_data), + data_content_type='application/json', + ) + + print(req_data, flush=True) + await asyncio.sleep(0.5) + + # A second topic, handled by a second async handler on the same subscriber. + req_data = {'id': 4, 'message': 'hello world'} + await client.publish_event( + pubsub_name='pubsub', + topic_name='TOPIC_B', + data=json.dumps(req_data), + data_content_type='application/json', + ) + print(req_data, flush=True) + + await asyncio.sleep(0.5) + + # Bulk publish multiple events at once using publish_events + bulk_events = [ + json.dumps({'id': 20, 'message': 'bulk event 1'}), + json.dumps({'id': 21, 'message': 'bulk event 2'}), + json.dumps({'id': 22, 'message': 'bulk event 3'}), + ] + + resp = await client.publish_events( + pubsub_name='pubsub', + topic_name='TOPIC_A', + data=bulk_events, + data_content_type='application/json', + ) + + print( + f'Bulk published {len(bulk_events)} events. Failed entries: {len(resp.failed_entries)}', + flush=True, + ) + + await asyncio.sleep(0.5) + + +asyncio.run(main()) diff --git a/examples/pubsub-simple-async/requirements.txt b/examples/pubsub-simple-async/requirements.txt new file mode 100644 index 000000000..349ce048e --- /dev/null +++ b/examples/pubsub-simple-async/requirements.txt @@ -0,0 +1 @@ +dapr[grpc] >= 1.19.0.dev diff --git a/examples/pubsub-simple-async/subscriber.py b/examples/pubsub-simple-async/subscriber.py new file mode 100644 index 000000000..d7c25d0d6 --- /dev/null +++ b/examples/pubsub-simple-async/subscriber.py @@ -0,0 +1,60 @@ +# ------------------------------------------------------------ +# Copyright 2025 The Dapr Authors +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ------------------------------------------------------------ + +import asyncio + +from dapr.ext.grpc.aio import App, SubscriptionMessage, TopicEventResponse + +app = App() + + +@app.subscribe(pubsub_name='pubsub', topic='TOPIC_A') +async def mytopic(event: SubscriptionMessage) -> TopicEventResponse: + # event.data() is already parsed based on the content type (dict for application/json) + data = event.data() + + # Awaiting inside the handler yields the event loop, so the subscriber keeps + # accepting deliveries while this one waits on I/O. + await asyncio.sleep(0) + + print( + f'Subscriber received: id={data["id"]}, message="{data["message"]}", ' + f'content_type="{event.data_content_type()}"', + flush=True, + ) + return TopicEventResponse('success') + + +@app.subscribe(pubsub_name='pubsub', topic='TOPIC_B') +async def myothertopic(event: SubscriptionMessage) -> TopicEventResponse: + data = event.data() + + await asyncio.sleep(0) + + print( + f'Other-Subscriber received: id={data["id"]}, message="{data["message"]}", ' + f'content_type="{event.data_content_type()}"', + flush=True, + ) + return TopicEventResponse('success') + + +async def healthy() -> None: + # Awaited by the app health check. Deliberately silent: daprd probes on a timer, so + # printing here would bury the subscriber output this example exists to show. + await asyncio.sleep(0) + + +app.register_health_check(healthy) + +asyncio.run(app.run(13551)) diff --git a/examples/pubsub-streaming-async/README.md b/examples/pubsub-streaming-async/README.md index 35c399b13..7801cf251 100644 --- a/examples/pubsub-streaming-async/README.md +++ b/examples/pubsub-streaming-async/README.md @@ -4,6 +4,10 @@ This example utilizes a publisher and a subscriber to show the bidirectional pub It creates a publisher and calls the `publish_event` method in the `DaprClient`. In the s`subscriber.py` file it creates a subscriber object that can call the `next_message` method to get new messages from the stream. After processing the new message, it returns a status to the stream. +This is the client-side streaming subscription API, where your code pulls messages. For the +server-side callback model — where Dapr delivers messages to `@app.subscribe` handlers on an +asyncio gRPC app — see [`pubsub-simple-async`](../pubsub-simple-async). + > **Note:** Make sure to use the latest proto bindings diff --git a/tests/examples/test_invoke_simple_async.py b/tests/examples/test_invoke_simple_async.py new file mode 100644 index 000000000..fb1bd03e2 --- /dev/null +++ b/tests/examples/test_invoke_simple_async.py @@ -0,0 +1,32 @@ +import pytest + +EXPECTED_CALLER = [ + 'text/plain', + 'INVOKE_RECEIVED', +] + +EXPECTED_RECEIVER = [ + '{"id": 1, "message": "hello world"}', + '{"id": 2, "message": "hello world"}', + '{"id": 3, "message": "hello world"}', +] + + +@pytest.mark.example_dir('invoke-simple-async') +def test_invoke_simple_async(dapr): + dapr.start( + '--app-id invoke-receiver --app-protocol grpc --app-port 13551 ' + '-- python3 invoke-receiver.py', + ) + + caller_output = dapr.run( + '--app-id invoke-caller --app-protocol grpc -- python3 invoke-caller.py', + timeout=30, + ) + for line in EXPECTED_CALLER: + assert line in caller_output, f'Missing in caller output: {line}' + assert caller_output.count('INVOKE_RECEIVED') == 3, 'Expected three concurrent invocations' + + receiver_output = dapr.stop() + for line in EXPECTED_RECEIVER: + assert line in receiver_output, f'Missing in receiver output: {line}' diff --git a/tests/examples/test_pubsub_simple_async.py b/tests/examples/test_pubsub_simple_async.py new file mode 100644 index 000000000..14b5f8eea --- /dev/null +++ b/tests/examples/test_pubsub_simple_async.py @@ -0,0 +1,37 @@ +import pytest + +EXPECTED_SUBSCRIBER = [ + 'Subscriber received: id=1, message="hello world", content_type="application/json"', + 'Subscriber received: id=2, message="hello world", content_type="application/json"', + 'Subscriber received: id=3, message="hello world", content_type="application/json"', + 'Other-Subscriber received: id=4, message="hello world", content_type="application/json"', + 'Subscriber received: id=20, message="bulk event 1", content_type="application/json"', + 'Subscriber received: id=21, message="bulk event 2", content_type="application/json"', + 'Subscriber received: id=22, message="bulk event 3", content_type="application/json"', +] + +EXPECTED_PUBLISHER = [ + "{'id': 1, 'message': 'hello world'}", + "{'id': 2, 'message': 'hello world'}", + "{'id': 3, 'message': 'hello world'}", + "{'id': 4, 'message': 'hello world'}", + 'Bulk published 3 events. Failed entries: 0', +] + + +@pytest.mark.example_dir('pubsub-simple-async') +def test_pubsub_simple_async(dapr): + dapr.start( + '--app-id python-subscriber --app-protocol grpc --app-port 13551 ' + '--enable-app-health-check --app-health-probe-interval 1 -- python3 subscriber.py', + ) + publisher_output = dapr.run( + '--app-id python-publisher --app-protocol grpc -- python3 publisher.py', + timeout=30, + ) + for line in EXPECTED_PUBLISHER: + assert line in publisher_output, f'Missing in publisher output: {line}' + + subscriber_output = dapr.stop() + for line in EXPECTED_SUBSCRIBER: + assert line in subscriber_output, f'Missing in subscriber output: {line}' diff --git a/tests/ext/grpc/aio/__init__.py b/tests/ext/grpc/aio/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/ext/grpc/aio/test_app.py b/tests/ext/grpc/aio/test_app.py new file mode 100644 index 000000000..cf72d7df5 --- /dev/null +++ b/tests/ext/grpc/aio/test_app.py @@ -0,0 +1,383 @@ +# -*- coding: utf-8 -*- + +""" +Copyright 2025 The Dapr Authors +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +""" + +import asyncio +import unittest +import warnings +from unittest.mock import MagicMock, patch + +from dapr.conf import settings +from dapr.ext.grpc.aio import ( + App, + BindingRequest, + InvokeMethodRequest, + JobEvent, + Rule, + SubscriptionMessage, +) + + +class AppTests(unittest.TestCase): + def setUp(self): + self._app = App() + + def test_method_decorator(self): + @self._app.method('Method1') + async def method1(request: InvokeMethodRequest): + pass + + @self._app.method('Method2') + async def method2(request: InvokeMethodRequest): + pass + + method_map = self._app._servicer._invoke_method_map + self.assertIn('AppTests.test_method_decorator..method1', str(method_map['Method1'])) + self.assertIn('AppTests.test_method_decorator..method2', str(method_map['Method2'])) + + def test_binding_decorator(self): + @self._app.binding('binding1') + async def binding1(request: BindingRequest): + pass + + binding_map = self._app._servicer._binding_map + self.assertIn( + 'AppTests.test_binding_decorator..binding1', str(binding_map['binding1']) + ) + + def test_subscribe_decorator(self): + @self._app.subscribe(pubsub_name='pubsub', topic='topic') + async def handle_default(event: SubscriptionMessage) -> None: + pass + + @self._app.subscribe( + pubsub_name='pubsub', topic='topic', rule=Rule('event.type == "test"', 1) + ) + async def handle_test_event(event: SubscriptionMessage) -> None: + pass + + @self._app.subscribe(pubsub_name='pubsub', topic='topic2', dead_letter_topic='topic2_dead') + async def handle_dead_letter(event: SubscriptionMessage) -> None: + pass + + subscription_map = self._app._servicer._topic_map + self.assertIn( + 'AppTests.test_subscribe_decorator..handle_default', + str(subscription_map['pubsub:topic:']), + ) + self.assertIn( + 'AppTests.test_subscribe_decorator..handle_test_event', + str(subscription_map['pubsub:topic:handle_test_event']), + ) + self.assertIn( + 'AppTests.test_subscribe_decorator..handle_dead_letter', + str(subscription_map['pubsub:topic2:']), + ) + + def test_job_event_decorator(self): + @self._app.job_event('job1') + async def handle_job(event: JobEvent) -> None: + pass + + job_map = self._app._servicer._job_event_map + self.assertIn('AppTests.test_job_event_decorator..handle_job', str(job_map['job1'])) + + def test_decorators_return_the_handler(self): + """Unlike the sync App, the aio decorators leave the decorated name bound.""" + + @self._app.method('Method1') + async def method1(request: InvokeMethodRequest): + pass + + @self._app.binding('binding1') + async def binding1(request: BindingRequest): + pass + + @self._app.subscribe(pubsub_name='pubsub', topic='topic') + async def handle_event(event: SubscriptionMessage) -> None: + pass + + @self._app.job_event('job1') + async def handle_job(event: JobEvent) -> None: + pass + + for handler in (method1, binding1, handle_event, handle_job): + self.assertTrue(callable(handler)) + + def test_subscribe_warns_for_unannotated_handler(self): + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + + @self._app.subscribe(pubsub_name='pubsub', topic='topic') + async def handler(event): + pass + + deprecations = [w for w in caught if issubclass(w.category, DeprecationWarning)] + self.assertEqual(1, len(deprecations)) + self.assertIn('SubscriptionMessage', str(deprecations[0].message)) + + def test_subscribe_annotated_handler_does_not_warn(self): + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + + @self._app.subscribe(pubsub_name='pubsub', topic='topic') + async def handler(event: SubscriptionMessage): + pass + + deprecations = [w for w in caught if issubclass(w.category, DeprecationWarning)] + self.assertEqual([], deprecations) + + def test_register_health_check(self): + async def health_check_cb(): + pass + + self._app.register_health_check(health_check_cb) + registered_cb = self._app._health_check_servicer._health_check_cb + self.assertIn( + 'AppTests.test_register_health_check..health_check_cb', str(registered_cb) + ) + + def test_no_health_check(self): + registered_cb = self._app._health_check_servicer._health_check_cb + self.assertIsNone(registered_cb) + + +class AppServerCreationTests(unittest.TestCase): + """The aio server is built lazily so it binds to the loop that will serve requests.""" + + def test_server_is_not_created_on_init(self): + app = App() + + self.assertIsNone(app._server) + + @patch('dapr.ext.grpc.aio.app.grpc.aio.server') + def test_create_server_registers_every_servicer(self, mock_server): + mock_server.return_value = MagicMock() + app = App() + + server = app._create_server() + + registered_services = { + handler.service_name() + for call in server.add_generic_rpc_handlers.call_args_list + for handler in call[0][0] + } + self.assertEqual( + { + 'dapr.proto.runtime.v1.AppCallback', + 'dapr.proto.runtime.v1.AppCallbackAlpha', + 'dapr.proto.runtime.v1.AppCallbackHealthCheck', + }, + registered_services, + ) + self.assertIs(mock_server.return_value, server) + + @patch('dapr.ext.grpc.aio.app.grpc.aio.server') + def test_create_server_passes_kwargs_through(self, mock_server): + mock_server.return_value = MagicMock() + app = App(max_grpc_message_length=32 * 1024 * 1024) + + app._create_server() + + _, kwargs = mock_server.call_args + options = dict(kwargs.get('options') or []) + self.assertEqual(32 * 1024 * 1024, options.get('grpc.max_send_message_length')) + self.assertEqual(32 * 1024 * 1024, options.get('grpc.max_receive_message_length')) + + +class AppGrpcOptionsTests(unittest.TestCase): + """Exercises options passed to grpc.aio.server() based on env var / constructor arg.""" + + def _options(self, app): + return dict(app._server_kwargs.get('options') or []) + + @patch.object(settings, 'DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES', 0) + def test_default_no_size_options(self): + options = self._options(App()) + + self.assertNotIn('grpc.max_send_message_length', options) + self.assertNotIn('grpc.max_receive_message_length', options) + + @patch.object(settings, 'DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES', 8 * 1024 * 1024) + def test_env_var_sets_receive_only(self): + options = self._options(App()) + + self.assertEqual(8 * 1024 * 1024, options.get('grpc.max_receive_message_length')) + self.assertNotIn('grpc.max_send_message_length', options) + + @patch.object(settings, 'DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES', 8 * 1024 * 1024) + def test_constructor_arg_overrides_env(self): + options = self._options(App(max_grpc_message_length=32 * 1024 * 1024)) + + self.assertEqual(32 * 1024 * 1024, options.get('grpc.max_send_message_length')) + self.assertEqual(32 * 1024 * 1024, options.get('grpc.max_receive_message_length')) + + def test_explicit_kwargs_replace_options(self): + app = App(migration_thread_pool=None, maximum_concurrent_rpcs=7) + + self.assertEqual(7, app._server_kwargs['maximum_concurrent_rpcs']) + self.assertNotIn('options', app._server_kwargs) + + +class AppExternalServiceTests(unittest.TestCase): + @patch('dapr.ext.grpc.aio.app.grpc.aio.server') + def test_external_service_is_registered_when_server_is_created(self, mock_server): + mock_server.return_value = MagicMock() + app = App() + servicer_callback = MagicMock() + external_servicer = MagicMock() + + app.add_external_service(servicer_callback, external_servicer) + servicer_callback.assert_not_called() + + server = app._create_server() + + servicer_callback.assert_called_once_with(external_servicer, server) + + @patch('dapr.ext.grpc.aio.app.grpc.aio.server') + def test_external_service_after_start_raises(self, mock_server): + """A running grpc.aio server cannot take new services, so late registration raises.""" + mock_server.return_value = MagicMock() + app = App() + app._server = mock_server.return_value # simulate a started app + + with self.assertRaises(RuntimeError) as exception_context: + app.add_external_service(MagicMock(), MagicMock()) + + self.assertIn('before the app is started', str(exception_context.exception)) + + +async def _suspending_start() -> None: + """Stands in for grpc.aio's start(), yielding so concurrent callers really interleave.""" + await asyncio.sleep(0.05) + + +class AppLifecycleTests(unittest.IsolatedAsyncioTestCase): + async def test_stop_before_start_is_a_noop(self): + app = App() + + await app.stop() + + self.assertIsNone(app._server) + + @patch('dapr.ext.grpc.aio.app.grpc.aio.server') + async def test_double_start_raises(self, mock_server): + mock_server.return_value = MagicMock() + mock_server.return_value.start = unittest.mock.AsyncMock() + app = App() + + await app.start(app_port=50055, listen_address='127.0.0.1') + + with self.assertRaises(RuntimeError): + await app.start(app_port=50055, listen_address='127.0.0.1') + + @patch('dapr.ext.grpc.aio.app.grpc.aio.server') + async def test_concurrent_starts_raise(self, mock_server): + """Only one of two concurrent start() calls may win. + + The mocked start() suspends, so the two calls genuinely interleave — without that the + first would run to completion before the second began and the race would go untested. + """ + mock_server.return_value = MagicMock() + mock_server.return_value.start = unittest.mock.AsyncMock(side_effect=_suspending_start) + app = App() + + results = await asyncio.gather( + app.start(app_port=50055, listen_address='127.0.0.1'), + app.start(app_port=50056, listen_address='127.0.0.1'), + return_exceptions=True, + ) + + errors = [r for r in results if isinstance(r, RuntimeError)] + self.assertEqual(1, len(errors), f'expected exactly one rejection, got {results}') + self.assertEqual(1, mock_server.call_count, 'a second server must not be built') + + @patch('dapr.ext.grpc.aio.app.grpc.aio.server') + async def test_external_service_during_start_raises(self, mock_server): + """Registration is rejected while a start() is still in flight, not just after it.""" + mock_server.return_value = MagicMock() + mock_server.return_value.start = unittest.mock.AsyncMock(side_effect=_suspending_start) + app = App() + + start_task = asyncio.ensure_future(app.start(app_port=50055, listen_address='127.0.0.1')) + await asyncio.sleep(0) # let start() reach its await with the slot claimed + + with self.assertRaises(RuntimeError): + app.add_external_service(MagicMock(), MagicMock()) + + await start_task + + @patch('dapr.ext.grpc.aio.app.grpc.aio.server') + async def test_stop_clears_the_server_so_the_app_can_restart(self, mock_server): + mock_server.return_value = MagicMock() + mock_server.return_value.start = unittest.mock.AsyncMock() + mock_server.return_value.stop = unittest.mock.AsyncMock() + app = App() + + await app.start(app_port=50055, listen_address='127.0.0.1') + await app.stop(grace=0) + + self.assertIsNone(app._server) + mock_server.return_value.stop.assert_awaited_once_with(0) + + +class AppLoopGuardTests(unittest.IsolatedAsyncioTestCase): + """The lifecycle lock binds to a loop, so loop identity is part of the contract.""" + + async def test_foreign_loop_with_a_live_server_raises(self): + app = App() + app._server = MagicMock() + app._lifecycle_lock = asyncio.Lock() + app._lifecycle_loop = asyncio.new_event_loop() + self.addCleanup(app._lifecycle_loop.close) + + with self.assertRaises(RuntimeError) as exception_context: + app._lock_for_running_loop() + + self.assertIn('different event loop', str(exception_context.exception)) + + def _app_with_a_dead_owning_loop(self) -> App: + dead_loop = asyncio.new_event_loop() + dead_loop.close() + app = App() + app._server = MagicMock() + app._lifecycle_lock = asyncio.Lock() + app._lifecycle_loop = dead_loop + return app + + async def test_stop_abandons_a_server_whose_loop_is_closed(self): + """stop() is the documented remedy, so it must not be refused - but it must warn. + + The server cannot actually be shut down without its loop, and its listener survives. + """ + app = self._app_with_a_dead_owning_loop() + + with self.assertLogs('dapr.ext.grpc.aio.app', level='WARNING') as logs: + await app.stop(grace=0) + + self.assertIsNone(app._server, 'the unreachable server must be released') + self.assertIn('second server', '\n'.join(logs.output)) + + async def test_start_refuses_while_an_unreachable_server_is_still_bound(self): + """Rebinding would quietly add a second listener, since grpc enables SO_REUSEPORT.""" + app = self._app_with_a_dead_owning_loop() + + with self.assertRaises(RuntimeError) as exception_context: + await app.start(app_port=50055, listen_address='127.0.0.1') + + self.assertIn('different event loop', str(exception_context.exception)) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/ext/grpc/aio/test_health_servicer.py b/tests/ext/grpc/aio/test_health_servicer.py new file mode 100644 index 000000000..a51e1d822 --- /dev/null +++ b/tests/ext/grpc/aio/test_health_servicer.py @@ -0,0 +1,55 @@ +# -*- coding: utf-8 -*- + +""" +Copyright 2025 The Dapr Authors +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +""" + +import unittest +from unittest.mock import AsyncMock, MagicMock + +from dapr.ext.grpc.aio._health_servicer import _AioHealthCheckServicer + + +class HealthCheckTests(unittest.IsolatedAsyncioTestCase): + def setUp(self): + self._health_servicer = _AioHealthCheckServicer() + + async def test_async_healthcheck_cb_awaited(self): + health_cb = AsyncMock() + self._health_servicer.register_health_check(health_cb) + + await self._health_servicer.HealthCheck(None, MagicMock()) + + health_cb.assert_awaited_once() + + async def test_sync_healthcheck_cb_called(self): + """`register_health_check(lambda: None)` keeps working on the aio app.""" + health_cb = MagicMock() + self._health_servicer.register_health_check(health_cb) + + await self._health_servicer.HealthCheck(None, MagicMock()) + + health_cb.assert_called_once() + + async def test_no_healthcheck_cb(self): + with self.assertRaises(NotImplementedError) as exception_context: + await self._health_servicer.HealthCheck(None, MagicMock()) + + self.assertIn('Method not implemented!', exception_context.exception.args[0]) + + def test_falsy_healthcheck_cb_rejected(self): + with self.assertRaises(ValueError): + self._health_servicer.register_health_check(None) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/ext/grpc/aio/test_server.py b/tests/ext/grpc/aio/test_server.py new file mode 100644 index 000000000..e6bee444a --- /dev/null +++ b/tests/ext/grpc/aio/test_server.py @@ -0,0 +1,415 @@ +# -*- coding: utf-8 -*- + +""" +Copyright 2025 The Dapr Authors +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. + +End-to-end coverage over a real grpc.aio server, exercising the wiring the unit tests mock +out: servicer registration, the aio context coroutines, and graceful shutdown. +""" + +import asyncio +import contextlib +import socket +import unittest + +import grpc.aio +from google.protobuf.any_pb2 import Any as GrpcAny +from google.protobuf.empty_pb2 import Empty + +from dapr.ext.grpc.aio import ( + App, + BindingRequest, + InvokeMethodRequest, + InvokeMethodResponse, + JobEvent, + SubscriptionMessage, + TopicEventResponse, +) +from dapr.proto import appcallback_service_v1, appcallback_v1, common_v1 + +RPC_TIMEOUT_SECONDS = 10 + + +def _free_port() -> int: + """Reserves an ephemeral port for the test server to bind.""" + with socket.socket() as probe: + probe.bind(('127.0.0.1', 0)) + return probe.getsockname()[1] + + +async def _wait_until_serving(port: int, run_task: 'asyncio.Task') -> None: + """Waits until the port actually accepts connections, not merely until it is assigned. + + ``App._server`` is set *before* ``server.start()`` is awaited, so polling it would return + while the server is still coming up — and a cancellation aimed at the serving app could + land inside startup instead. ``_free_port`` is also racy, so a bind can fail outright; + surfacing ``run_task``'s exception beats spinning on a server that will never appear. + """ + + # One channel, closed by this function rather than inside poll(): a close awaited in + # poll()'s finally would run while wait_for is cancelling it, and could surface as a bare + # CancelledError instead of the TimeoutError that actually explains the failure. + channel = grpc.aio.insecure_channel(f'127.0.0.1:{port}') + + async def poll() -> None: + while True: + if run_task.done(): + await run_task # re-raises whatever stopped the server coming up + raise AssertionError('run() returned before the server was serving') + try: + await asyncio.wait_for(channel.channel_ready(), timeout=0.25) + return + except asyncio.TimeoutError: + continue + + try: + await asyncio.wait_for(poll(), timeout=RPC_TIMEOUT_SECONDS) + except BaseException: + # Otherwise run_task keeps its listener for the rest of the session and the loop is + # torn down under a live task. + run_task.cancel() + with contextlib.suppress(Exception, asyncio.CancelledError): + await run_task + raise + finally: + await channel.close() + + +class AioAppServerTests(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self): + self.received_events = [] + self.received_bindings = [] + self.received_jobs = [] + self.health_checks = [] + + self._app = App() + self._register_handlers(self._app) + + self._port = _free_port() + await self._app.start(app_port=self._port, listen_address='127.0.0.1') + self.addAsyncCleanup(self._app.stop, 0) + + self._channel = grpc.aio.insecure_channel(f'127.0.0.1:{self._port}') + self.addAsyncCleanup(self._channel.close) + self.stub = appcallback_service_v1.AppCallbackStub(self._channel) + self.health_stub = appcallback_service_v1.AppCallbackHealthCheckStub(self._channel) + + def _register_handlers(self, app): + @app.method('echo') + async def echo(request: InvokeMethodRequest): + await asyncio.sleep(0) + response = InvokeMethodResponse(b'async-pong', 'text/plain') + response.headers = (('x-custom', 'header-value'),) + return response + + @app.subscribe(pubsub_name='pubsub', topic='orders') + async def on_order(event: SubscriptionMessage) -> TopicEventResponse: + await asyncio.sleep(0) + self.received_events.append(event.data()) + return TopicEventResponse('success') + + @app.binding('input') + async def on_binding(request: BindingRequest) -> None: + await asyncio.sleep(0) + self.received_bindings.append(request.text()) + + @app.job_event('nightly') + async def on_job(event: JobEvent) -> None: + await asyncio.sleep(0) + self.received_jobs.append(event.get_data_as_string()) + + async def health_check() -> None: + self.health_checks.append(True) + + app.register_health_check(health_check) + + async def test_async_method_handler_returns_data_and_headers(self): + call = self.stub.OnInvoke( + common_v1.InvokeRequest(method='echo', data=GrpcAny()), + timeout=RPC_TIMEOUT_SECONDS, + ) + response = await call + initial_metadata = dict(await call.initial_metadata()) + + self.assertEqual(b'async-pong', response.data.value) + self.assertEqual('text/plain', response.content_type) + self.assertEqual('header-value', initial_metadata.get('x-custom')) + + async def test_unregistered_method_is_unimplemented(self): + with self.assertRaises(grpc.aio.AioRpcError) as exception_context: + await self.stub.OnInvoke( + common_v1.InvokeRequest(method='missing', data=GrpcAny()), + timeout=RPC_TIMEOUT_SECONDS, + ) + + self.assertEqual(grpc.StatusCode.UNIMPLEMENTED, exception_context.exception.code()) + + async def test_list_topic_subscriptions(self): + response = await self.stub.ListTopicSubscriptions(Empty(), timeout=RPC_TIMEOUT_SECONDS) + + self.assertEqual( + [('pubsub', 'orders')], + [(sub.pubsub_name, sub.topic) for sub in response.subscriptions], + ) + + async def test_async_topic_handler_receives_event(self): + request = appcallback_v1.TopicEventRequest( + id='event-1', + pubsub_name='pubsub', + topic='orders', + data=b'{"id": 1}', + data_content_type='application/json', + ) + + response = await self.stub.OnTopicEvent(request, timeout=RPC_TIMEOUT_SECONDS) + + self.assertEqual([{'id': 1}], self.received_events) + self.assertEqual( + appcallback_v1.TopicEventResponse.TopicEventResponseStatus.SUCCESS, response.status + ) + + async def test_list_input_bindings(self): + response = await self.stub.ListInputBindings(Empty(), timeout=RPC_TIMEOUT_SECONDS) + + self.assertEqual(['input'], list(response.bindings)) + + async def test_async_binding_handler_receives_event(self): + request = appcallback_v1.BindingEventRequest(name='input', data=b'binding-payload') + + await self.stub.OnBindingEvent(request, timeout=RPC_TIMEOUT_SECONDS) + + self.assertEqual(['binding-payload'], self.received_bindings) + + async def test_async_job_handler_receives_event(self): + request = appcallback_v1.JobEventRequest(name='nightly', data=GrpcAny(value=b'job-payload')) + + await self.stub.OnJobEvent(request, timeout=RPC_TIMEOUT_SECONDS) + + self.assertEqual(['job-payload'], self.received_jobs) + + async def test_async_health_check(self): + await self.health_stub.HealthCheck(Empty(), timeout=RPC_TIMEOUT_SECONDS) + + self.assertEqual([True], self.health_checks) + + async def test_concurrent_requests_are_not_serialized(self): + """A request must complete while an earlier, still-blocked one is in flight. + + The slow handler is confirmed to be inside its await before the fast request is + sent, so a serialized dispatcher would deadlock rather than pass. + """ + slow_entered = asyncio.Event() + release_slow = asyncio.Event() + completion_order = [] + app = App() + + @app.method('slow') + async def slow(request: InvokeMethodRequest): + slow_entered.set() + await release_slow.wait() + completion_order.append('slow') + return b'slow-done' + + @app.method('fast') + async def fast(request: InvokeMethodRequest): + completion_order.append('fast') + release_slow.set() + return b'fast-done' + + port = _free_port() + await app.start(app_port=port, listen_address='127.0.0.1') + try: + async with grpc.aio.insecure_channel(f'127.0.0.1:{port}') as channel: + stub = appcallback_service_v1.AppCallbackStub(channel) + + slow_task = asyncio.ensure_future( + stub.OnInvoke( + common_v1.InvokeRequest(method='slow', data=GrpcAny()), + timeout=RPC_TIMEOUT_SECONDS, + ) + ) + # Only send the second request once the first is provably blocked. + await asyncio.wait_for(slow_entered.wait(), timeout=RPC_TIMEOUT_SECONDS) + + fast_response = await asyncio.wait_for( + stub.OnInvoke( + common_v1.InvokeRequest(method='fast', data=GrpcAny()), + timeout=RPC_TIMEOUT_SECONDS, + ), + timeout=RPC_TIMEOUT_SECONDS, + ) + slow_response = await asyncio.wait_for(slow_task, timeout=RPC_TIMEOUT_SECONDS) + + self.assertEqual(['fast', 'slow'], completion_order) + self.assertEqual(b'slow-done', slow_response.data.value) + self.assertEqual(b'fast-done', fast_response.data.value) + finally: + await app.stop(grace=0) + + +class AioAppRunTests(unittest.IsolatedAsyncioTestCase): + async def test_run_serves_until_stopped(self): + app = App() + + @app.method('ping') + async def ping(request: InvokeMethodRequest): + return b'pong' + + port = _free_port() + run_task = asyncio.create_task(app.run(app_port=port, listen_address='127.0.0.1')) + await _wait_until_serving(port, run_task) + + try: + async with grpc.aio.insecure_channel(f'127.0.0.1:{port}') as channel: + stub = appcallback_service_v1.AppCallbackStub(channel) + response = await asyncio.wait_for( + stub.OnInvoke( + common_v1.InvokeRequest(method='ping', data=GrpcAny()), + timeout=RPC_TIMEOUT_SECONDS, + ), + timeout=RPC_TIMEOUT_SECONDS, + ) + self.assertEqual(b'pong', response.data.value) + finally: + await app.stop(grace=0) + + await asyncio.wait_for(run_task, timeout=RPC_TIMEOUT_SECONDS) + + async def test_run_releases_the_server_when_cancelled(self): + """Cancelling run() must stop the server rather than leave the port bound. + + The synchronous App stops its server from __del__; a coroutine cannot be awaited + from one, so run() has to clean up after itself. + """ + app = App() + + @app.method('ping') + async def ping(request: InvokeMethodRequest): + return b'pong' + + port = _free_port() + run_task = asyncio.create_task(app.run(app_port=port, listen_address='127.0.0.1')) + await _wait_until_serving(port, run_task) + + run_task.cancel() + with self.assertRaises(asyncio.CancelledError): + await run_task + + self.assertIsNone(app._server) + + channel = grpc.aio.insecure_channel(f'127.0.0.1:{port}') + try: + with self.assertRaises(asyncio.TimeoutError): + await asyncio.wait_for(channel.channel_ready(), timeout=2) + finally: + await channel.close() + + # The app must be reusable, not wedged by a stale _server reference. + await app.start(app_port=_free_port(), listen_address='127.0.0.1') + await app.stop(grace=0) + + +class AppLoopBindingTests(unittest.TestCase): + """The lifecycle lock binds to a loop, so loop identity is part of the contract. + + These run their own ``asyncio.run`` blocks rather than using IsolatedAsyncioTestCase, + because the behaviour under test is precisely what happens across separate loops. + """ + + def test_reuse_on_a_second_loop_after_a_clean_stop(self): + app = App() + port_one, port_two = _free_port(), _free_port() + + async def cycle(port): + await app.start(app_port=port, listen_address='127.0.0.1') + await app.stop(grace=0) + + asyncio.run(cycle(port_one)) + asyncio.run(cycle(port_two)) # must not raise "bound to a different event loop" + + def test_foreign_loop_while_owning_loop_is_alive_raises(self): + app = App() + loop = asyncio.new_event_loop() + try: + loop.run_until_complete(app.start(app_port=_free_port(), listen_address='127.0.0.1')) + + async def stop_from_elsewhere(): + await app.stop(grace=0) + + with self.assertRaises(RuntimeError) as ctx: + asyncio.run(stop_from_elsewhere()) + self.assertIn('different event loop', str(ctx.exception)) + finally: + # In the finally: if the guard ever regresses the assertion fails here, and an + # un-stopped server would leak a bound port into every later test. + if app._server is not None: + loop.run_until_complete(app.stop(grace=0)) + loop.close() + + +class AppStartFailureTests(unittest.IsolatedAsyncioTestCase): + async def test_bind_failure_leaves_no_dangling_server(self): + """A taken port must not strand the server built for it.""" + holder = socket.socket() + holder.bind(('127.0.0.1', 0)) + holder.listen() + self.addCleanup(holder.close) + taken_port = holder.getsockname()[1] + + app = App() + with self.assertRaises(RuntimeError): + await app.start(app_port=taken_port, listen_address='127.0.0.1') + self.assertIsNone(app._server) + + # and the App is still usable afterwards + await app.start(app_port=_free_port(), listen_address='127.0.0.1') + await app.stop(grace=0) + + async def test_cancelled_drain_clears_the_server_and_closes_the_port(self): + """A graceful stop that is cut short must leave the App stopped and the port free. + + grpc closes the listener as soon as the drain begins, so this pins the observable + contract rather than the forced ``stop(None)`` specifically; that call is defensive. + """ + app = App() + + @app.method('slow') + async def slow(request: InvokeMethodRequest): + await asyncio.sleep(30) + return b'never' + + port = _free_port() + await app.start(app_port=port, listen_address='127.0.0.1') + + async with grpc.aio.insecure_channel(f'127.0.0.1:{port}') as channel: + stub = appcallback_service_v1.AppCallbackStub(channel) + in_flight = asyncio.ensure_future( + stub.OnInvoke(common_v1.InvokeRequest(method='slow', data=GrpcAny())) + ) + await asyncio.sleep(0.2) + + with self.assertRaises(asyncio.TimeoutError): + await asyncio.wait_for(app.stop(grace=30), timeout=1) + + in_flight.cancel() + + self.assertIsNone(app._server) + probe = grpc.aio.insecure_channel(f'127.0.0.1:{port}') + try: + with self.assertRaises(asyncio.TimeoutError): + await asyncio.wait_for(probe.channel_ready(), timeout=2) + finally: + await probe.close() + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/ext/grpc/aio/test_servicer.py b/tests/ext/grpc/aio/test_servicer.py new file mode 100644 index 000000000..498d9ded1 --- /dev/null +++ b/tests/ext/grpc/aio/test_servicer.py @@ -0,0 +1,457 @@ +# -*- coding: utf-8 -*- + +""" +Copyright 2025 The Dapr Authors +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +""" + +import inspect +import unittest +from unittest.mock import AsyncMock, MagicMock, Mock + +from cloudevents.sdk.event import v1 +from google.protobuf.any_pb2 import Any as GrpcAny + +from dapr.clients.grpc._request import InvokeMethodRequest +from dapr.clients.grpc._response import InvokeMethodResponse, TopicEventResponse +from dapr.common.pubsub.subscription import SubscriptionMessage +from dapr.ext.grpc._health_servicer import _HealthCheckServicer +from dapr.ext.grpc._servicer import _CallbackServicer +from dapr.ext.grpc.aio._health_servicer import _AioHealthCheckServicer +from dapr.ext.grpc.aio._servicer import _AioCallbackServicer +from dapr.proto import appcallback_service_v1, appcallback_v1, common_v1 +from dapr.proto.runtime.v1.appcallback_pb2 import ( + TopicEventBulkRequest, + TopicEventBulkRequestEntry, + TopicEventCERequest, +) + +DEFAULT_INVOCATION_METADATA = (('key1', 'value1'), ('key2', 'value1')) + + +def fake_aio_context(invocation_metadata=DEFAULT_INVOCATION_METADATA): + """Mocks a grpc.aio ServicerContext. + + ``invocation_metadata`` is synchronous on the aio context while + ``send_initial_metadata`` is a coroutine, so the two are mocked differently. + """ + context = MagicMock() + context.invocation_metadata.return_value = invocation_metadata + context.send_initial_metadata = AsyncMock() + return context + + +class OnInvokeTests(unittest.IsolatedAsyncioTestCase): + def setUp(self): + self._servicer = _AioCallbackServicer() + self.fake_context = fake_aio_context() + + async def _on_invoke(self, method_name, method_cb): + self._servicer.register_method(method_name, method_cb) + + return await self._servicer.OnInvoke( + common_v1.InvokeRequest(method=method_name, data=GrpcAny()), + self.fake_context, + ) + + async def test_on_invoke_return_str(self): + async def method_cb(request: InvokeMethodRequest): + return 'method_str_cb' + + resp = await self._on_invoke('method_str', method_cb) + + self.assertEqual(b'method_str_cb', resp.data.value) + + async def test_on_invoke_return_bytes(self): + async def method_cb(request: InvokeMethodRequest): + return b'method_str_cb' + + resp = await self._on_invoke('method_bytes', method_cb) + + self.assertEqual(b'method_str_cb', resp.data.value) + + async def test_on_invoke_return_proto(self): + async def method_cb(request: InvokeMethodRequest): + return common_v1.StateItem(key='fake_key') + + resp = await self._on_invoke('method_proto', method_cb) + + state = common_v1.StateItem() + resp.data.Unpack(state) + + self.assertEqual('fake_key', state.key) + + async def test_on_invoke_return_invoke_method_response(self): + async def method_cb(request: InvokeMethodRequest): + return InvokeMethodResponse(data='fake_data', content_type='text/plain') + + resp = await self._on_invoke('method_resp', method_cb) + + self.assertEqual(b'fake_data', resp.data.value) + self.assertEqual('text/plain', resp.content_type) + + async def test_on_invoke_invalid_response(self): + async def method_cb(request: InvokeMethodRequest): + return 1000 + + with self.assertRaises(NotImplementedError): + await self._on_invoke('method_resp', method_cb) + + async def test_on_invoke_awaits_send_initial_metadata(self): + """Headers are sent through the aio context's coroutine, not a bare call.""" + + async def method_cb(request: InvokeMethodRequest): + resp = InvokeMethodResponse(data='fake_data', content_type='text/plain') + resp.headers = (('x-custom', 'header-value'),) + return resp + + await self._on_invoke('method_headers', method_cb) + + self.fake_context.send_initial_metadata.assert_awaited_once() + + async def test_on_invoke_receives_invocation_metadata(self): + received = [] + + async def method_cb(request: InvokeMethodRequest): + received.append(request.metadata) + return 'ok' + + await self._on_invoke('method_metadata', method_cb) + + self.assertEqual({'key1': ['value1'], 'key2': ['value1']}, received[0]) + + async def test_on_invoke_accepts_sync_handler(self): + def method_cb(request: InvokeMethodRequest): + return 'sync_result' + + resp = await self._on_invoke('method_sync', method_cb) + + self.assertEqual(b'sync_result', resp.data.value) + + async def test_non_registered_method(self): + with self.assertRaises(NotImplementedError): + await self._servicer.OnInvoke( + common_v1.InvokeRequest(method='unknown', data=GrpcAny()), + self.fake_context, + ) + + +class TopicSubscriptionTests(unittest.IsolatedAsyncioTestCase): + def setUp(self): + self._servicer = _AioCallbackServicer() + self._topic_method = AsyncMock(return_value=None) + self._servicer.register_topic('pubsub1', 'topic1', self._topic_method, {'session': 'key'}) + self.fake_context = fake_aio_context() + + def _request(self, topic='topic1', pubsub_name='pubsub1', path=''): + return appcallback_v1.TopicEventRequest( + id='event-1', + data_content_type='application/json', + data=b'{"a": 1}', + topic=topic, + pubsub_name=pubsub_name, + path=path, + ) + + def test_duplicated_topic(self): + with self.assertRaises(ValueError): + self._servicer.register_topic('pubsub1', 'topic1', self._topic_method, {}) + + async def test_list_topic_subscription(self): + resp = await self._servicer.ListTopicSubscriptions(None, self.fake_context) + + self.assertEqual('pubsub1', resp.subscriptions[0].pubsub_name) + self.assertEqual('topic1', resp.subscriptions[0].topic) + + async def test_topic_event_awaits_handler(self): + await self._servicer.OnTopicEvent(self._request(), self.fake_context) + + self._topic_method.assert_awaited_once() + + async def test_topic_event_response_status(self): + self._topic_method.return_value = TopicEventResponse('retry') + + resp = await self._servicer.OnTopicEvent(self._request(), self.fake_context) + + self.assertEqual( + appcallback_v1.TopicEventResponse.TopicEventResponseStatus.RETRY, resp.status + ) + + async def test_topic_event_accepts_sync_handler(self): + sync_handler = Mock(return_value=TopicEventResponse('drop')) + self._servicer.register_topic('pubsub2', 'topic2', sync_handler, {}) + + resp = await self._servicer.OnTopicEvent( + self._request(topic='topic2', pubsub_name='pubsub2'), self.fake_context + ) + + sync_handler.assert_called_once() + self.assertEqual( + appcallback_v1.TopicEventResponse.TopicEventResponseStatus.DROP, resp.status + ) + + async def test_non_registered_topic(self): + with self.assertRaises(NotImplementedError): + await self._servicer.OnTopicEvent(self._request(topic='unknown'), self.fake_context) + + +class BulkTopicEventTests(unittest.IsolatedAsyncioTestCase): + def setUp(self): + self._servicer = _AioCallbackServicer() + self._topic_method = AsyncMock(return_value=TopicEventResponse('success')) + self._servicer.register_topic('pubsub1', 'topic1', self._topic_method, {'session': 'key'}) + self.fake_context = fake_aio_context() + + def _request(self, entries, topic='topic1'): + return TopicEventBulkRequest( + id='bulk1', + pubsub_name='pubsub1', + topic=topic, + path='', + entries=entries, + ) + + async def test_on_bulk_topic_event(self): + entry1 = TopicEventBulkRequestEntry( + entry_id='entry1', bytes=b'hello', content_type='text/plain' + ) + entry2 = TopicEventBulkRequestEntry( + entry_id='entry2', bytes=b'{"a": 1}', content_type='application/json' + ) + + resp = await self._servicer.OnBulkTopicEvent( + self._request([entry1, entry2]), self.fake_context + ) + + self.assertEqual(2, len(resp.statuses)) + self.assertEqual('entry1', resp.statuses[0].entry_id) + self.assertEqual('entry2', resp.statuses[1].entry_id) + self.assertEqual( + appcallback_v1.TopicEventResponse.TopicEventResponseStatus.SUCCESS, + resp.statuses[0].status, + ) + self.assertEqual(2, self._topic_method.await_count) + + async def test_on_bulk_topic_event_cloud_event_entry(self): + cloud_event = TopicEventCERequest( + id='ce-1', + source='ce-source', + type='ce.type', + spec_version='1.0', + data_content_type='application/json', + data=b'{"a": 1}', + ) + entry = TopicEventBulkRequestEntry(entry_id='entry1', cloud_event=cloud_event) + + resp = await self._servicer.OnBulkTopicEvent(self._request([entry]), self.fake_context) + + self.assertEqual(1, len(resp.statuses)) + delivered = self._topic_method.await_args[0][0] + self.assertEqual('ce-1', delivered.EventID()) + + async def test_on_bulk_topic_event_handler_raises_retry(self): + self._topic_method.side_effect = ValueError('handler exploded') + entry = TopicEventBulkRequestEntry(entry_id='entry1', bytes=b'hello') + + resp = await self._servicer.OnBulkTopicEvent(self._request([entry]), self.fake_context) + + self.assertEqual( + appcallback_v1.TopicEventResponse.TopicEventResponseStatus.RETRY, + resp.statuses[0].status, + ) + + async def test_on_bulk_topic_event_alpha1(self): + entry = TopicEventBulkRequestEntry(entry_id='entry1', bytes=b'hello') + + with self.assertWarns(DeprecationWarning): + resp = await self._servicer.OnBulkTopicEventAlpha1( + self._request([entry]), self.fake_context + ) + + self.assertEqual(1, len(resp.statuses)) + + async def test_on_bulk_topic_event_non_registered(self): + entry = TopicEventBulkRequestEntry(entry_id='entry1', bytes=b'hello') + + with self.assertRaises(NotImplementedError): + await self._servicer.OnBulkTopicEvent( + self._request([entry], topic='unknown'), self.fake_context + ) + + +class BindingTests(unittest.IsolatedAsyncioTestCase): + def setUp(self): + self._servicer = _AioCallbackServicer() + self._binding_method = AsyncMock(return_value=None) + self._servicer.register_binding('binding1', self._binding_method) + self.fake_context = fake_aio_context() + + def test_duplicated_binding(self): + with self.assertRaises(ValueError): + self._servicer.register_binding('binding1', self._binding_method) + + async def test_list_bindings(self): + resp = await self._servicer.ListInputBindings(None, self.fake_context) + + self.assertEqual(['binding1'], list(resp.bindings)) + + async def test_binding_event_awaits_handler(self): + request = appcallback_v1.BindingEventRequest(name='binding1', data=b'hello') + + resp = await self._servicer.OnBindingEvent(request, self.fake_context) + + self.assertIsInstance(resp, appcallback_v1.BindingEventResponse) + self._binding_method.assert_awaited_once() + self.assertEqual(b'hello', self._binding_method.await_args[0][0].data) + + async def test_non_registered_binding(self): + request = appcallback_v1.BindingEventRequest(name='unknown', data=b'hello') + + with self.assertRaises(NotImplementedError): + await self._servicer.OnBindingEvent(request, self.fake_context) + + +class JobEventTests(unittest.IsolatedAsyncioTestCase): + def setUp(self): + self._servicer = _AioCallbackServicer() + self._handler = AsyncMock(return_value=None) + self._servicer.register_job_event('test-job', self._handler) + self.fake_context = fake_aio_context() + + def _request(self, name='test-job', payload=b'hello'): + return appcallback_v1.JobEventRequest(name=name, data=GrpcAny(value=payload)) + + def test_duplicated_job_event(self): + with self.assertRaises(ValueError): + self._servicer.register_job_event('test-job', self._handler) + + async def test_on_job_event_stable_routes_to_handler(self): + resp = await self._servicer.OnJobEvent(self._request(), self.fake_context) + + self.assertIsInstance(resp, appcallback_v1.JobEventResponse) + self._handler.assert_awaited_once() + job_event = self._handler.await_args[0][0] + self.assertEqual('test-job', job_event.name) + self.assertEqual('hello', job_event.get_data_as_string()) + + async def test_on_job_event_alpha1_routes_to_same_handler(self): + resp = await self._servicer.OnJobEventAlpha1(self._request(), self.fake_context) + + self.assertIsInstance(resp, appcallback_v1.JobEventResponse) + self._handler.assert_awaited_once() + + async def test_non_registered_job_event(self): + with self.assertRaises(NotImplementedError): + await self._servicer.OnJobEvent(self._request(name='unknown-job'), self.fake_context) + + +class TopicEventTypeDeliveryTests(unittest.IsolatedAsyncioTestCase): + """The aio servicer honours the same legacy-cloudevent opt-out as the sync one.""" + + def setUp(self): + self._servicer = _AioCallbackServicer() + self.fake_context = fake_aio_context(invocation_metadata=(('trace', 'abc'),)) + + async def _deliver(self, legacy_cloudevent): + handler = AsyncMock(return_value=None) + self._servicer.register_topic( + 'pubsub1', 'topic1', handler, {}, legacy_cloudevent=legacy_cloudevent + ) + request = appcallback_v1.TopicEventRequest( + id='event-1', + data_content_type='application/json', + data=b'{"a": 1}', + topic='topic1', + pubsub_name='pubsub1', + ) + + await self._servicer.OnTopicEvent(request, self.fake_context) + + return handler.await_args[0][0] + + async def test_legacy_default_delivers_cloudevent(self): + event = await self._deliver(legacy_cloudevent=True) + + self.assertIsInstance(event, v1.Event) + self.assertEqual('event-1', event.EventID()) + self.assertEqual('abc', event.Extensions()['_metadata_trace']) + + async def test_opt_out_delivers_subscription_message(self): + event = await self._deliver(legacy_cloudevent=False) + + self.assertIsInstance(event, SubscriptionMessage) + self.assertEqual('event-1', event.id()) + self.assertEqual({'a': 1}, event.data()) + + +class AsyncParityTests(unittest.TestCase): + """Guards the sync/aio servicer pair against drifting apart.""" + + def _rpc_names_for(self, generated, sync_cls): + """RPC names the sync servicer implements, anywhere in its MRO above the stubs.""" + generated_rpc_names = { + name for servicer in generated for name in vars(servicer) if not name.startswith('_') + } + # Union over the MRO, not vars(sync_cls): an RPC deduplicated onto a shared base + # would otherwise drop out of the guard silently. + sync_implemented = set() + for klass in sync_cls.__mro__: + if klass in generated: + break + sync_implemented |= set(vars(klass)) + return sorted(generated_rpc_names & sync_implemented) + + def test_health_servicer_pair_stays_in_parity(self): + """The health servicer pair needs the same guard as the callback pair.""" + generated = (appcallback_service_v1.AppCallbackHealthCheckServicer,) + names = self._rpc_names_for(generated, _HealthCheckServicer) + self.assertNotEqual([], names) + + for name in names: + with self.subTest(rpc=name): + aio_method = getattr(_AioHealthCheckServicer, name) + self.assertTrue(inspect.iscoroutinefunction(aio_method)) + self.assertIsNot(aio_method, getattr(_HealthCheckServicer, name)) + + def test_every_sync_rpc_has_an_awaitable_counterpart(self): + """A new RPC on the sync servicer must be mirrored as `async def` on the aio one.""" + rpc_names = self._rpc_names_for( + ( + appcallback_service_v1.AppCallbackServicer, + appcallback_service_v1.AppCallbackAlphaServicer, + ), + _CallbackServicer, + ) + self.assertNotEqual([], rpc_names) + + for name in rpc_names: + with self.subTest(rpc=name): + aio_method = getattr(_AioCallbackServicer, name) + self.assertTrue( + inspect.iscoroutinefunction(aio_method), + f'{name} must be a coroutine function on the asyncio servicer', + ) + self.assertIsNot( + aio_method, + getattr(_CallbackServicer, name), + f'{name} must be overridden by the asyncio servicer, not inherited', + ) + + def test_registration_helpers_are_shared_not_duplicated(self): + """Registration and routing must stay on the common base, not be reimplemented.""" + for name in ('register_method', 'register_topic', 'register_binding', 'register_job_event'): + with self.subTest(helper=name): + self.assertNotIn(name, vars(_AioCallbackServicer)) + self.assertIs(getattr(_AioCallbackServicer, name), getattr(_CallbackServicer, name)) + + +if __name__ == '__main__': + unittest.main()