From 2d49ef1348f317f175e02ca23ebf2bcaf85ca460 Mon Sep 17 00:00:00 2001 From: LilDojd Date: Wed, 8 Jul 2026 16:40:23 +0400 Subject: [PATCH] Pass per-operation StateItem metadata in async execute_state_transaction The async client omitted metadata= when building StateItem, unlike the sync client (added in #791), so per-operation metadata e.g. outbox.projection and CloudEvent attribute overrides for the transactional outbox was silently dropped. Mirrors the sync client Signed-off-by: LilDojd Change-Id: I568579c8d1c1d3599bf1fd03dcd1587b6a6a6964 --- dapr/aio/clients/grpc/client.py | 1 + tests/clients/fake_dapr_server.py | 5 +++++ tests/clients/test_dapr_grpc_client.py | 10 +++++++++- tests/clients/test_dapr_grpc_client_async.py | 10 +++++++++- 4 files changed, 24 insertions(+), 2 deletions(-) diff --git a/dapr/aio/clients/grpc/client.py b/dapr/aio/clients/grpc/client.py index 3d089b816..47beeff0a 100644 --- a/dapr/aio/clients/grpc/client.py +++ b/dapr/aio/clients/grpc/client.py @@ -1028,6 +1028,7 @@ async def execute_state_transaction( key=o.key, value=to_bytes(o.data) if o.data is not None else to_bytes(''), etag=common_v1.Etag(value=o.etag) if o.etag is not None else None, + metadata=o.metadata, ), ) for o in operations diff --git a/tests/clients/fake_dapr_server.py b/tests/clients/fake_dapr_server.py index 4b54f0379..f15826fed 100644 --- a/tests/clients/fake_dapr_server.py +++ b/tests/clients/fake_dapr_server.py @@ -23,6 +23,7 @@ def __init__(self, grpc_port: int = 50001, http_port: int = 8080): self._http_server = FakeHttpServer(self.http_port) # Needed for the healthcheck endpoint api_service_v1.add_DaprServicer_to_server(self, self._grpc_server) self.store = {} + self.transaction_operation_metadata: Dict[str, Dict[str, str]] = {} self.shutdown_received = False self.locks_to_owner = {} # (store_name, resource_id) -> lock_owner self.metadata: Dict[str, str] = {} @@ -284,7 +285,11 @@ def ExecuteStateTransaction(self, request, context): headers = () trailers = () + self.transaction_operation_metadata = {} for operation in request.operations: + self.transaction_operation_metadata[operation.request.key] = dict( + operation.request.metadata + ) if operation.operationType == 'delete': del self.store[operation.request.key] else: diff --git a/tests/clients/test_dapr_grpc_client.py b/tests/clients/test_dapr_grpc_client.py index c90b6fd50..f2638f972 100644 --- a/tests/clients/test_dapr_grpc_client.py +++ b/tests/clients/test_dapr_grpc_client.py @@ -562,12 +562,20 @@ def test_transaction_then_get_states(self): dapr.execute_state_transaction( store_name='statestore', operations=[ - TransactionalStateOperation(key=key, data=value, etag='foo'), + TransactionalStateOperation( + key=key, data=value, etag='foo', metadata={'outbox.projection': 'true'} + ), TransactionalStateOperation(key=another_key, data=another_value), ], transactional_metadata={'metakey': 'metavalue'}, ) + self.assertEqual( + self._fake_dapr_server.transaction_operation_metadata[key], + {'outbox.projection': 'true'}, + ) + self.assertEqual(self._fake_dapr_server.transaction_operation_metadata[another_key], {}) + resp = dapr.get_bulk_state(store_name='statestore', keys=[key, another_key]) self.assertEqual(resp.items[0].key, key) self.assertEqual(resp.items[0].data, to_bytes(value)) diff --git a/tests/clients/test_dapr_grpc_client_async.py b/tests/clients/test_dapr_grpc_client_async.py index ccec1084a..8ef183a89 100644 --- a/tests/clients/test_dapr_grpc_client_async.py +++ b/tests/clients/test_dapr_grpc_client_async.py @@ -577,12 +577,20 @@ async def test_transaction_then_get_states(self): await dapr.execute_state_transaction( store_name='statestore', operations=[ - TransactionalStateOperation(key=key, data=value, etag='foo'), + TransactionalStateOperation( + key=key, data=value, etag='foo', metadata={'outbox.projection': 'true'} + ), TransactionalStateOperation(key=another_key, data=another_value), ], transactional_metadata={'metakey': 'metavalue'}, ) + self.assertEqual( + self._fake_dapr_server.transaction_operation_metadata[key], + {'outbox.projection': 'true'}, + ) + self.assertEqual(self._fake_dapr_server.transaction_operation_metadata[another_key], {}) + resp = await dapr.get_bulk_state(store_name='statestore', keys=[key, another_key]) self.assertEqual(resp.items[0].key, key) self.assertEqual(resp.items[0].data, to_bytes(value))