From 6675ac83c76fcc0dbd9e5f8d02b6a8ff931fadf9 Mon Sep 17 00:00:00 2001 From: nazarli-shabnam Date: Tue, 23 Jun 2026 18:27:45 +0400 Subject: [PATCH] fix(cluster): populate ClusterPipeline.command_stack via execution strategy (#3703) After the execution-strategy refactor (#3611), queued commands moved to `ClusterPipeline._execution_strategy.command_queue`, leaving the public `command_stack` attribute permanently empty on the sync cluster pipeline (and absent entirely on the async one). APM/tracing integrations such as Datadog's `ddtrace` introspect `command_stack` to build pipeline spans, so cluster pipeline traces silently lost all command information. Expose `command_stack` as a read-only property that delegates to the active execution strategy's `command_queue` on both the sync and async `ClusterPipeline`, restoring the legacy behavior without changing execution. No runtime deprecation warning is emitted (tracing tools read this on every execute); it is documented as legacy in the docstring instead. Approach proposed by @mathewlee11 and approved by the maintainers in #3703. Co-Authored-By: Claude Opus 4.8 (1M context) --- redis/asyncio/cluster.py | 22 ++++++++++++++++++++++ redis/cluster.py | 13 ++++++++++++- tests/test_asyncio/test_cluster.py | 23 +++++++++++++++++++++++ tests/test_cluster.py | 28 ++++++++++++++++++++++++++++ 4 files changed, 85 insertions(+), 1 deletion(-) diff --git a/redis/asyncio/cluster.py b/redis/asyncio/cluster.py index 7c86b85bff..95b83069cf 100644 --- a/redis/asyncio/cluster.py +++ b/redis/asyncio/cluster.py @@ -2286,6 +2286,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": @@ -2390,6 +2402,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": """ @@ -2494,6 +2512,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 be575154f7..62d95fca48 100644 --- a/redis/cluster.py +++ b/redis/cluster.py @@ -3420,7 +3420,6 @@ def __init__( **kwargs, ): """ """ - self.command_stack = [] self.nodes_manager = nodes_manager self.commands_parser = commands_parser self.refresh_table_asap = False @@ -3519,6 +3518,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