Cluster Batch (from version 2.0.0)

Home / python / cluster_batch Note - Starting version 2.0.0, the Batch class will be available and replace the Transaction class.

glide.async_commands.batch.ClusterBatch

Bases: BaseBatch

Batch implementation for cluster GlideClusterClient. Batches allow the execution of a group of commands in a single step.

Batch Response

An array of command responses is returned by the client exec command, in the order they were given. Each element in the array represents a command given to the ClusterBatch. The response for each command depends on the executed Valkey command. Specific response types are documented alongside each method.

Parameters:

Name	Type	Description	Default
is_atomic	bool	Determines whether the batch is atomic or non-atomic. If True, the batch will be executed as an atomic transaction. If False, the batch will be executed as a non-atomic pipeline.	required

See Valkey Transactions (Atomic Batches) and Valkey Pipelines (Non-Atomic Batches) for details.

Examples:

Atomic Batch - Transaction in a Cluster:

>>> transaction = ClusterBatch(is_atomic=True)  # Atomic (Transaction)
>>> transaction.set("key", "value")
>>> transaction.get("key")
>>> result = await client.exec(transaction, false)
>>> print(result)
[OK, b"value"]

Non-Atomic Batch - Pipeline in a Cluster:

>>> pipeline = ClusterBatch(is_atomic=False)  # Non-Atomic (Pipeline)
>>> pipeline.set("key1", "value1")
>>> pipeline.set("key2", "value2")
>>> pipeline.get("key1")
>>> pipeline.get("key2")
>>> result = await client.exec(pipeline, false)
>>> print(result)
[OK, OK, b"value1", b"value2"]

Source code in glide/async_commands/batch.py

class ClusterBatch(BaseBatch):
    """
    Batch implementation for cluster GlideClusterClient. Batches allow the execution of a group
    of commands in a single step.

    Batch Response:
        An ``array`` of command responses is returned by the client ``exec`` command,
        in the order they were given. Each element in the array represents a command given to the ClusterBatch.
        The response for each command depends on the executed Valkey command.
        Specific response types are documented alongside each method.

    Args:
        is_atomic (bool): Determines whether the batch is atomic or non-atomic. If ``True``,
            the batch will be executed as an atomic transaction. If ``False``,
            the batch will be executed as a non-atomic pipeline.

    See [Valkey Transactions (Atomic Batches)](https://valkey.io/topics/transactions/) and [Valkey Pipelines (Non-Atomic Batches)](https://valkey.io/topics/pipelining/) for details.

    Examples:
        ### Atomic Batch - Transaction in a Cluster:
        >>> transaction = ClusterBatch(is_atomic=True)  # Atomic (Transaction)
        >>> transaction.set("key", "value")
        >>> transaction.get("key")
        >>> result = await client.exec(transaction, false)
        >>> print(result)
        [OK, b"value"]

        ### Non-Atomic Batch - Pipeline in a Cluster:
        >>> pipeline = ClusterBatch(is_atomic=False)  # Non-Atomic (Pipeline)
        >>> pipeline.set("key1", "value1")
        >>> pipeline.set("key2", "value2")
        >>> pipeline.get("key1")
        >>> pipeline.get("key2")
        >>> result = await client.exec(pipeline, false)
        >>> print(result)
        [OK, OK, b"value1", b"value2"]

    """

    def copy(
        self,
        source: TEncodable,
        destination: TEncodable,
        replace: Optional[bool] = None,
    ) -> "ClusterBatch":
        """
        Copies the value stored at the `source` to the `destination` key. When `replace` is True,
        removes the `destination` key first if it already exists, otherwise performs no action.

        See [valkey.io](https://valkey.io/commands/copy) for more details.

        Args:
            source (TEncodable): The key to the source value.
            destination (TEncodable): The key where the value should be copied to.
            replace (Optional[bool]): If the destination key should be removed before copying the value to it.

        Command response:
            bool: True if the source was copied.

            Otherwise, return False.

        Since: Valkey version 6.2.0.
        """
        args = [source, destination]
        if replace is not None:
            args.append("REPLACE")

        return self.append_command(RequestType.Copy, args)

    def publish(
        self, message: str, channel: str, sharded: bool = False
    ) -> "ClusterBatch":
        """
        Publish a message on pubsub channel.
        This command aggregates PUBLISH and SPUBLISH commands functionalities.
        The mode is selected using the 'sharded' parameter

        See [PUBLISH](https://valkey.io/commands/publish) and [SPUBLISH](https://valkey.io/commands/spublish)
        for more details.

        Args:
            message (str): Message to publish
            channel (str): Channel to publish the message on.
            sharded (bool): Use sharded pubsub mode. Available since Valkey version 7.0.

        Returns:
            int: Number of subscriptions in that shard that received the message.
        """
        return self.append_command(
            RequestType.SPublish if sharded else RequestType.Publish, [channel, message]
        )

    def pubsub_shardchannels(
        self, pattern: Optional[TEncodable] = None
    ) -> "ClusterBatch":
        """
        Lists the currently active shard channels.

        See [valkey.io](https://valkey.io/commands/pubsub-shardchannels) for details.

        Args:
            pattern (Optional[TEncodable]): A glob-style pattern to match active shard channels.
                If not provided, all active shard channels are returned.

        Command response:
            List[bytes]: A list of currently active shard channels matching the given pattern.

            If no pattern is specified, all active shard channels are returned.
        """
        command_args = [pattern] if pattern is not None else []
        return self.append_command(RequestType.PubSubShardChannels, command_args)

    def pubsub_shardnumsub(
        self, channels: Optional[List[TEncodable]] = None
    ) -> "ClusterBatch":
        """
        Returns the number of subscribers (exclusive of clients subscribed to patterns) for the specified shard channels.

        Note:
            It is valid to call this command without channels. In this case, it will just return an empty map.

        See [valkey.io](https://valkey.io/commands/pubsub-shardnumsub) for details.

        Args:
            channels (Optional[List[str]]): The list of shard channels to query for the number of subscribers.
                If not provided, returns an empty map.

        Command response:
            Mapping[bytes, int]: A map where keys are the shard channel names and values are the number of subscribers.
        """
        return self.append_command(
            RequestType.PubSubShardNumSub, channels if channels else []
        )

copy(source, destination, replace=None)

Copies the value stored at the source to the destination key. When replace is True, removes the destination key first if it already exists, otherwise performs no action.

See valkey.io for more details.

Parameters:

Name	Type	Description	Default
source	TEncodable	The key to the source value.	required
destination	TEncodable	The key where the value should be copied to.	required
replace	Optional[bool]	If the destination key should be removed before copying the value to it.	None

Command response

bool: True if the source was copied.

Otherwise, return False.

Since: Valkey version 6.2.0.

Source code in glide/async_commands/batch.py

def copy(
    self,
    source: TEncodable,
    destination: TEncodable,
    replace: Optional[bool] = None,
) -> "ClusterBatch":
    """
    Copies the value stored at the `source` to the `destination` key. When `replace` is True,
    removes the `destination` key first if it already exists, otherwise performs no action.

    See [valkey.io](https://valkey.io/commands/copy) for more details.

    Args:
        source (TEncodable): The key to the source value.
        destination (TEncodable): The key where the value should be copied to.
        replace (Optional[bool]): If the destination key should be removed before copying the value to it.

    Command response:
        bool: True if the source was copied.

        Otherwise, return False.

    Since: Valkey version 6.2.0.
    """
    args = [source, destination]
    if replace is not None:
        args.append("REPLACE")

    return self.append_command(RequestType.Copy, args)

publish(message, channel, sharded=False)

Publish a message on pubsub channel. This command aggregates PUBLISH and SPUBLISH commands functionalities. The mode is selected using the 'sharded' parameter

See PUBLISH and SPUBLISH for more details.

Parameters:

Name	Type	Description	Default
message	str	Message to publish	required
channel	str	Channel to publish the message on.	required
sharded	bool	Use sharded pubsub mode. Available since Valkey version 7.0.	False

Returns:

Name	Type	Description
int	ClusterBatch	Number of subscriptions in that shard that received the message.

Source code in glide/async_commands/batch.py

def publish(
    self, message: str, channel: str, sharded: bool = False
) -> "ClusterBatch":
    """
    Publish a message on pubsub channel.
    This command aggregates PUBLISH and SPUBLISH commands functionalities.
    The mode is selected using the 'sharded' parameter

    See [PUBLISH](https://valkey.io/commands/publish) and [SPUBLISH](https://valkey.io/commands/spublish)
    for more details.

    Args:
        message (str): Message to publish
        channel (str): Channel to publish the message on.
        sharded (bool): Use sharded pubsub mode. Available since Valkey version 7.0.

    Returns:
        int: Number of subscriptions in that shard that received the message.
    """
    return self.append_command(
        RequestType.SPublish if sharded else RequestType.Publish, [channel, message]
    )

pubsub_shardchannels(pattern=None)

Lists the currently active shard channels.

See valkey.io for details.

Parameters:

Name	Type	Description	Default
pattern	Optional[TEncodable]	A glob-style pattern to match active shard channels. If not provided, all active shard channels are returned.	None

Command response

List[bytes]: A list of currently active shard channels matching the given pattern.

If no pattern is specified, all active shard channels are returned.

Source code in glide/async_commands/batch.py

def pubsub_shardchannels(
    self, pattern: Optional[TEncodable] = None
) -> "ClusterBatch":
    """
    Lists the currently active shard channels.

    See [valkey.io](https://valkey.io/commands/pubsub-shardchannels) for details.

    Args:
        pattern (Optional[TEncodable]): A glob-style pattern to match active shard channels.
            If not provided, all active shard channels are returned.

    Command response:
        List[bytes]: A list of currently active shard channels matching the given pattern.

        If no pattern is specified, all active shard channels are returned.
    """
    command_args = [pattern] if pattern is not None else []
    return self.append_command(RequestType.PubSubShardChannels, command_args)

pubsub_shardnumsub(channels=None)

Returns the number of subscribers (exclusive of clients subscribed to patterns) for the specified shard channels.

Note

It is valid to call this command without channels. In this case, it will just return an empty map.

See valkey.io for details.

Parameters:

Name	Type	Description	Default
channels	Optional[List[str]]	The list of shard channels to query for the number of subscribers. If not provided, returns an empty map.	None

Command response

Mapping[bytes, int]: A map where keys are the shard channel names and values are the number of subscribers.

Source code in glide/async_commands/batch.py

def pubsub_shardnumsub(
    self, channels: Optional[List[TEncodable]] = None
) -> "ClusterBatch":
    """
    Returns the number of subscribers (exclusive of clients subscribed to patterns) for the specified shard channels.

    Note:
        It is valid to call this command without channels. In this case, it will just return an empty map.

    See [valkey.io](https://valkey.io/commands/pubsub-shardnumsub) for details.

    Args:
        channels (Optional[List[str]]): The list of shard channels to query for the number of subscribers.
            If not provided, returns an empty map.

    Command response:
        Mapping[bytes, int]: A map where keys are the shard channel names and values are the number of subscribers.
    """
    return self.append_command(
        RequestType.PubSubShardNumSub, channels if channels else []
    )