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/_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/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/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, 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()