You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
{{ message }}
Repository navigation
Commit e2e4ccc
Browse filesBrowse the repository at this point in the historyBrowse files
feat: add asyncio support to the gRPC app extension (dapr.ext.grpc.aio)
dapr.ext.grpc.App runs on grpc.server() with a thread pool, so handlers must be
synchronous. Applications built on asyncio have had to vendor their own callback
server to use `async def` handlers.
Add dapr.ext.grpc.aio, backed by grpc.aio.server(). It exports the same names as
dapr.ext.grpc; swapping the import is the only change an app needs, beyond
`async def` handlers and awaiting run()/stop().
_AioCallbackServicer is a sibling of _CallbackServicer over the base extracted in
the previous commit - neither subclasses the other, so there is no sync/async
mixing in one MRO and no misleading isinstance relationship. It supplies only the
gRPC entry points, which await the handler result and the grpc.aio context
coroutines.
The asyncio servicer covers the full current surface: service invocation, pub/sub
including bulk events, input bindings, job events on both the stable and alpha
services, and health checks. AppCallbackAlphaServicer is registered on the
server, and send_initial_metadata is awaited, since it is a coroutine on the aio
context rather than a plain method.
Handlers may be plain functions - results are awaited only when awaitable - so
register_health_check(lambda: None) keeps working. The server is built on the
first run()/start() call rather than in __init__, because grpc.aio.server() binds
to whichever event loop is current when it is called; add_external_service()
therefore queues its registration and replays it at creation.
Adds unit tests including an end-to-end suite over a real grpc.aio server and
parity guards asserting every sync RPC is mirrored as a coroutine, plus the
invoke-simple-async and pubsub-simple-async examples.
The async client (dapr.aio.clients.DaprClient) already exists; this closes the
remaining server-side gap.
Closes#695
Signed-off-by: Sai Kishore Punagani <63619246+saikishore-p@users.noreply.github.com>
├── test_health_servicer.py # Async and sync health check callbacks
29
+
└── test_server.py # End-to-end over a real grpc.aio server
20
30
```
21
31
22
32
Installed via the `grpc` extra on core dapr: `pip install "dapr[grpc]"`.
@@ -42,6 +52,8 @@ from dapr.ext.grpc import (
42
52
43
53
Note: `InvokeMethodRequest`, `InvokeMethodResponse`, `BindingRequest`, `TopicEventResponse`, `Job`, `JobEvent`, and failure policies are actually defined in the core SDK (`dapr/clients/grpc/`) and re-exported here.
44
54
55
+
`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()`.
56
+
45
57
## App class (`app.py`)
46
58
47
59
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
88
100
-`TopicEventResponse('success'|'retry'|'drop')` → explicit status
89
101
-`None` → defaults to SUCCESS
90
102
103
+
## Asyncio app (`aio/`)
104
+
105
+
`dapr.ext.grpc.aio.App` is the asyncio counterpart, backed by `grpc.aio.server()`. Same decorators, same registration semantics, same wire behavior.
Differences from the synchronous `App`, all of them forced by the async runtime:
121
+
122
+
-**`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.
123
+
-**`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()`.
124
+
-**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.
125
+
-**`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.)
126
+
-**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.
127
+
-**Decorators return the handler**, so the decorated name stays bound. The sync decorators return `None`.
128
+
-**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.
129
+
130
+
### Sharing with the sync implementation
131
+
132
+
`_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.
133
+
134
+
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.
135
+
136
+
`_HealthCheckServicerBase` splits the health servicer the same way: registration in the base, the gRPC entry point in each sibling.
137
+
91
138
## Internal routing (`_servicer.py`)
92
139
93
-
`_CallbackServicer` implements `AppCallbackServicer` + `AppCallbackAlphaServicer` gRPC service interfaces. It maintains internal registries:
140
+
`_CallbackServicerBase` implements `AppCallbackServicer` + `AppCallbackAlphaServicer` gRPC service interfaces; `_CallbackServicer` adds the synchronous entry points. It maintains internal registries:
94
141
95
142
-`_invoke_method_map` — method name → handler
96
143
-`_topic_map` — topic key → handler
@@ -124,6 +171,13 @@ app.register_health_check(lambda: None) # Not a decorator — direct registrati
124
171
uv run python -m unittest discover -v ./tests/ext/grpc
125
172
```
126
173
174
+
`unittest discover` covers the whole tree including `aio/` — `IsolatedAsyncioTestCase` and
175
+
`subTest` are both native unittest. pytest runs the same tests:
176
+
177
+
```bash
178
+
uv run pytest ./tests/ext/grpc
179
+
```
180
+
127
181
Test patterns:
128
182
-`test_app.py` — decorator registration, health check registration
129
183
-`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:
132
186
133
187
## Key details
134
188
135
-
-**Synchronous only**: Uses `grpc.server()` with `ThreadPoolExecutor(10)`. No async handler support.
189
+
-**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.
-**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=...)`.)
191
+
-**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=...)`.)
138
192
-**Duplicate registration**: Registering the same method/topic/binding name twice raises `ValueError`
139
193
-**Missing handlers**: Calling an unregistered method/topic/binding raises `NotImplementedError` (gRPC UNIMPLEMENTED)
0 commit comments