diff --git a/redis/asyncio/cluster.py b/redis/asyncio/cluster.py index 9a80a988df..30954b222a 100644 --- a/redis/asyncio/cluster.py +++ b/redis/asyncio/cluster.py @@ -2400,6 +2400,18 @@ def __bool__(self) -> bool: def __len__(self) -> int: return len(self._execution_strategy) + @property + def command_stack(self) -> List["PipelineCommand"]: + """ + Deprecated. Returns the commands currently queued on the pipeline. + + Retained for backwards compatibility with code that introspects the + queued commands directly (for example APM/tracing integrations such as + Datadog's ``ddtrace``). The commands are owned by the underlying + execution strategy; prefer ``len(pipeline)`` to check the queue size. + """ + return self._execution_strategy.command_queue + def execute_command( self, *args: Union[KeyT, EncodableT], **kwargs: Any ) -> "ClusterPipeline": @@ -2504,6 +2516,12 @@ def __repr__(self) -> str: class ExecutionStrategy(ABC): + @property + @abstractmethod + def command_queue(self) -> List["PipelineCommand"]: + """The commands currently queued on the pipeline.""" + pass + @abstractmethod async def initialize(self) -> "ClusterPipeline": """ @@ -2608,6 +2626,10 @@ def __init__(self, pipe: ClusterPipeline) -> None: self._pipe: ClusterPipeline = pipe self._command_queue: List["PipelineCommand"] = [] + @property + def command_queue(self) -> List["PipelineCommand"]: + return self._command_queue + async def initialize(self) -> "ClusterPipeline": if self._pipe.cluster_client._initialize: await self._pipe.cluster_client.initialize() diff --git a/redis/cluster.py b/redis/cluster.py index e75cf238d7..0ea54605bd 100644 --- a/redis/cluster.py +++ b/redis/cluster.py @@ -3419,7 +3419,6 @@ def __init__( **kwargs, ): """ """ - self.command_stack = [] self.nodes_manager = nodes_manager self.commands_parser = commands_parser self.refresh_table_asap = False @@ -3518,6 +3517,18 @@ def __bool__(self): "Pipeline instances should always evaluate to True on Python 3+" return True + @property + def command_stack(self): + """ + Deprecated. Returns the commands currently queued on the pipeline. + + Retained for backwards compatibility with code that introspects the + queued commands directly (for example APM/tracing integrations such as + Datadog's ``ddtrace``). The commands are owned by the underlying + execution strategy; prefer ``len(pipeline)`` to check the queue size. + """ + return self._execution_strategy.command_queue + def execute_command(self, *args, **kwargs): """ Wrapper function for pipeline_execute_command diff --git a/tests/test_asyncio/test_cluster.py b/tests/test_asyncio/test_cluster.py index afaa1f7b7f..bc0b98bdf4 100644 --- a/tests/test_asyncio/test_cluster.py +++ b/tests/test_asyncio/test_cluster.py @@ -3809,6 +3809,29 @@ def custom_callback(response): finally: await r.aclose() + async def test_command_stack_reflects_queued_commands(self) -> None: + """ + ClusterPipeline.command_stack should expose the commands queued on the + pipeline. Regression test for #3703: after the execution-strategy + refactor (#3611) the async pipeline had no command_stack at all, which + silently broke APM/tracing integrations (e.g. Datadog ddtrace) that + introspect it. + """ + r = await get_mocked_redis_client(host=default_host, port=default_port) + try: + pipeline = r.pipeline() + assert pipeline.command_stack == [] + + pipeline.set("{foo}1", "1").set("{foo}2", "2") + + assert len(pipeline.command_stack) == 2 + # command_stack delegates to the execution strategy's command queue + assert pipeline.command_stack is pipeline._execution_strategy.command_queue + assert pipeline.command_stack[0].args == ("SET", "{foo}1", "1") + assert pipeline.command_stack[1].args == ("SET", "{foo}2", "2") + finally: + await r.aclose() + async def test_blocked_arguments(self, r: RedisCluster) -> None: """Test handling for blocked pipeline arguments.""" diff --git a/tests/test_cluster.py b/tests/test_cluster.py index e09af591cc..eb390f08b2 100644 --- a/tests/test_cluster.py +++ b/tests/test_cluster.py @@ -3845,6 +3845,34 @@ def test_redis_cluster_pipeline(self, r): pipe.get("foo") assert pipe.execute() == [True, b"bar"] + def test_command_stack_reflects_queued_commands(self): + """ + ClusterPipeline.command_stack should expose the commands queued on the + pipeline. Regression test for #3703: after the execution-strategy + refactor (#3611) command_stack was always empty, which silently broke + APM/tracing integrations (e.g. Datadog ddtrace) that introspect it. + """ + r = get_mocked_redis_client(host=default_host, port=default_port) + with r.pipeline() as pipe: + assert pipe.command_stack == [] + + pipe.set("{foo}1", "1").set("{foo}2", "2") + + assert len(pipe.command_stack) == 2 + # command_stack delegates to the execution strategy's command queue + assert pipe.command_stack is pipe._execution_strategy.command_queue + assert pipe.command_stack[0].args == ("SET", "{foo}1", "1") + assert pipe.command_stack[1].args == ("SET", "{foo}2", "2") + + @pytest.mark.parametrize("transaction", [False, True]) + def test_command_stack_delegates_to_execution_strategy(self, transaction): + """command_stack should delegate to the active execution strategy's + command queue for both the pipeline and transaction strategies.""" + r = get_mocked_redis_client(host=default_host, port=default_port) + pipe = r.pipeline(transaction=transaction) + assert pipe.command_stack is pipe._execution_strategy.command_queue + assert pipe.command_stack == [] + def test_mget_disabled(self, r): """ Test that mget is disabled for ClusterPipeline