Method execute_command

redis/asyncio/cluster.py:952–1096

Execute a raw command on the appropriate cluster node or target_nodes. It will retry the command as specified by the retries property of the :attr:`retry` & then raise an exception. :param args: | Raw command args :param kwargs: - t

(self, *args: EncodableT, **kwargs: Any)

        )

    async def execute_command(self, *args: EncodableT, **kwargs: Any) -> Any:
        """
        Execute a raw command on the appropriate cluster node or target_nodes.

        It will retry the command as specified by the retries property of
        the :attr:`retry` & then raise an exception.

        :param args:
            | Raw command args
        :param kwargs:

            - target_nodes: :attr:`NODE_FLAGS` or :class:`~.ClusterNode`
              or List[:class:`~.ClusterNode`] or Dict[Any, :class:`~.ClusterNode`]
            - Rest of the kwargs are passed to the Redis connection

        :raises RedisClusterException: if target_nodes is not provided & the command
            can't be mapped to a slot
        """
        command = args[0]
        target_nodes = []
        target_nodes_specified = False
        retry_attempts = self.retry.get_retries()

        passed_targets = kwargs.pop("target_nodes", None)
        if passed_targets and not self._is_node_flag(passed_targets):
            target_nodes = self._parse_target_nodes(passed_targets)
            target_nodes_specified = True
            retry_attempts = 0

        command_policies = await self._policy_resolver.resolve(args[0].lower())

        if not command_policies and not target_nodes_specified:
            command_flag = self.command_flags.get(command)
            if not command_flag:
                # Fallback to default policy
                if not self.get_default_node():
                    slot = None
                else:
                    slot = await self._determine_slot(*args)
                if slot is None:
                    command_policies = CommandPolicies()
                else:
                    command_policies = CommandPolicies(
                        request_policy=RequestPolicy.DEFAULT_KEYED,
                        response_policy=ResponsePolicy.DEFAULT_KEYED,
                    )
            else:
                if command_flag in self._command_flags_mapping:
                    command_policies = CommandPolicies(
                        request_policy=self._command_flags_mapping[command_flag]
                    )
                else:
                    command_policies = CommandPolicies()
        elif not command_policies and target_nodes_specified:
            command_policies = CommandPolicies()

        # Add one for the first execution
        execute_attempts = 1 + retry_attempts