[Backport release-1.18] Pass per-operation StateItem metadata in async execute_state_transaction

dapr/aio/clients/grpc/client.py

@@ -1026,6 +1026,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

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:

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))

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))