from __future__ import annotations

import datetime

import hashlib

import inspect

from dataclasses import dataclass

# Try to import the xxhash library as an optional dependency

try:

import xxhash

HAS_XXHASH = True

except ImportError:

HAS_XXHASH = False

import warnings

from enum import Enum

from typing import (

TYPE_CHECKING,

Any,

AsyncIterator,

Awaitable,

Callable,

Dict,

Iterable,

Iterator,

List,

Literal,

Mapping,

Optional,

Sequence,

Tuple,

Union,

overload,

)

from redis.asyncio.observability.recorder import (

record_streaming_lag_from_response as async_record_streaming_lag,

)

from redis.exceptions import ConnectionError, DataError, NoScriptError, RedisError

from redis.typing import (

AbsExpiryT,

ACLGetUserData,

ACLLogData,

AnyKeyT,

AsyncClientProtocol,

BitfieldOffsetT,

BlockingListPopResponse,

BlockingZSetPopResponse,

ChannelT,

CommandGetKeysAndFlagsResponse,

CommandsProtocol,

ConsumerT,

EncodableT,

ExpiryT,

FieldT,

GroupT,

HRandFieldResponse,

HScanResponse,

KeysT,

KeyT,

ListMultiPopResponse,

Number,

PatternT,

ResponseT,

ScanResponse,

ScriptTextT,

SortResponse,

StralgoResponse,

StreamIdT,

StreamRangeResponse,

SyncClientProtocol,

TimeoutSecT,

XClaimResponse,

XPendingRangeResponse,

XReadResponse,

ZMPopResponse,

ZRandMemberResponse,

ZScanResponse,

ZScoreBoundT,

ZSetRangeResponse,

)

from redis.utils import (

deprecated_function,

experimental_args,

experimental_method,

extract_expire_flags,

str_if_bytes,

)

from ..observability.attributes import PubSubDirection

from ..observability.recorder import (

record_pubsub_message,

record_streaming_lag_from_response,

)

from .helpers import at_most_one_value_set, list_or_args

if TYPE_CHECKING:

import redis.asyncio.client

import redis.asyncio.cluster

import redis.client

import redis.cluster

@dataclass

class GCRAResponse:

"""Response from the GCRA (Generic Cell Rate Algorithm) rate limiting command.

Attributes:

limited: Whether the request was rate limited (True) or allowed (False).

max_req_num: Maximum number of requests allowed (always equals max_burst + 1).

num_avail_req: Number of requests available immediately.

retry_after: Seconds until the caller should retry. Returns -1 if not limited.

full_burst_after: Seconds until the full burst allowance is restored.

"""

limited: bool

max_req_num: int

num_avail_req: int

retry_after: int

full_burst_after: int

class ACLCommands(CommandsProtocol):

"""

Redis Access Control List (ACL) commands.

see: https://redis.io/topics/acl

"""

@overload

def acl_cat(

self: SyncClientProtocol, category: str | None = None, **kwargs

) -> list[str]: ...

@overload

def acl_cat(

self: AsyncClientProtocol, category: str | None = None, **kwargs

) -> Awaitable[list[str]]: ...

def acl_cat(

self, category: str | None = None, **kwargs

) -> list[str] | Awaitable[list[str]]:

"""

Returns a list of categories or commands within a category.

If ``category`` is not supplied, returns a list of all categories.

If ``category`` is supplied, returns a list of all commands within

that category.

For more information, see https://redis.io/commands/acl-cat

"""

pieces: list[EncodableT] = [category] if category else []

return self.execute_command("ACL CAT", *pieces, **kwargs)

@overload

def acl_dryrun(

self: SyncClientProtocol, username: str, *args: EncodableT, **kwargs

) -> bytes | str: ...

@overload

def acl_dryrun(

self: AsyncClientProtocol, username: str, *args: EncodableT, **kwargs

) -> Awaitable[bytes | str]: ...

def acl_dryrun(self, username: str, *args: EncodableT, **kwargs) -> (

bytes | str

) | Awaitable[bytes | str]:

"""

Simulate the execution of a given command by a given ``username``.

For more information, see https://redis.io/commands/acl-dryrun

"""

return self.execute_command("ACL DRYRUN", username, *args, **kwargs)

@overload

def acl_deluser(self: SyncClientProtocol, *username: str, **kwargs) -> int: ...

@overload

def acl_deluser(

self: AsyncClientProtocol, *username: str, **kwargs

) -> Awaitable[int]: ...

def acl_deluser(self, *username: str, **kwargs) -> int | Awaitable[int]:

"""

Delete the ACL for the specified ``username``\\s

For more information, see https://redis.io/commands/acl-deluser

"""

return self.execute_command("ACL DELUSER", *username, **kwargs)

@overload

def acl_genpass(

self: SyncClientProtocol, bits: int | None = None, **kwargs

) -> str: ...

@overload

def acl_genpass(

self: AsyncClientProtocol, bits: int | None = None, **kwargs

) -> Awaitable[str]: ...

def acl_genpass(self, bits: int | None = None, **kwargs) -> (str) | Awaitable[str]:

"""Generate a random password value.

If ``bits`` is supplied then use this number of bits, rounded to

the next multiple of 4.

See: https://redis.io/commands/acl-genpass

"""

pieces = []

if bits is not None:

try:

b = int(bits)

if b < 0 or b > 4096:

raise ValueError

pieces.append(b)

except ValueError:

raise DataError(

"genpass optionally accepts a bits argument, between 0 and 4096."

)

return self.execute_command("ACL GENPASS", *pieces, **kwargs)

@overload

def acl_getuser(

self: SyncClientProtocol, username: str, **kwargs

) -> ACLGetUserData: ...

@overload

def acl_getuser(

self: AsyncClientProtocol, username: str, **kwargs

) -> Awaitable[ACLGetUserData]: ...

def acl_getuser(

self, username: str, **kwargs

) -> ACLGetUserData | Awaitable[ACLGetUserData]:

"""

Get the ACL details for the specified ``username``.

If ``username`` does not exist, return None

For more information, see https://redis.io/commands/acl-getuser

"""

return self.execute_command("ACL GETUSER", username, **kwargs)

@overload

def acl_help(self: SyncClientProtocol, **kwargs) -> list[str]: ...

@overload

def acl_help(self: AsyncClientProtocol, **kwargs) -> Awaitable[list[str]]: ...

def acl_help(self, **kwargs) -> list[str] | Awaitable[list[str]]:

"""The ACL HELP command returns helpful text describing

the different subcommands.

For more information, see https://redis.io/commands/acl-help

"""

return self.execute_command("ACL HELP", **kwargs)

@overload

def acl_list(self: SyncClientProtocol, **kwargs) -> list[str]: ...

@overload

def acl_list(self: AsyncClientProtocol, **kwargs) -> Awaitable[list[str]]: ...

def acl_list(self, **kwargs) -> list[str] | Awaitable[list[str]]:

"""

Return a list of all ACLs on the server

For more information, see https://redis.io/commands/acl-list

"""

return self.execute_command("ACL LIST", **kwargs)

@overload

def acl_log(

self: SyncClientProtocol, count: int | None = None, **kwargs

) -> ACLLogData: ...

@overload

def acl_log(

self: AsyncClientProtocol, count: int | None = None, **kwargs

) -> Awaitable[ACLLogData]: ...

def acl_log(

self, count: int | None = None, **kwargs

) -> ACLLogData | Awaitable[ACLLogData]:

"""

Get ACL logs as a list.

:param int count: Get logs[0:count].

:rtype: List.

For more information, see https://redis.io/commands/acl-log

"""

args = []

if count is not None:

if not isinstance(count, int):

raise DataError("ACL LOG count must be an integer")

args.append(count)

return self.execute_command("ACL LOG", *args, **kwargs)

@overload

def acl_log_reset(self: SyncClientProtocol, **kwargs) -> bool: ...

@overload

def acl_log_reset(self: AsyncClientProtocol, **kwargs) -> Awaitable[bool]: ...

def acl_log_reset(self, **kwargs) -> bool | Awaitable[bool]:

"""

Reset ACL logs.

:rtype: Boolean.

For more information, see https://redis.io/commands/acl-log

"""

args = [b"RESET"]

return self.execute_command("ACL LOG", *args, **kwargs)

@overload

def acl_load(self: SyncClientProtocol, **kwargs) -> bool: ...

@overload

def acl_load(self: AsyncClientProtocol, **kwargs) -> Awaitable[bool]: ...

def acl_load(self, **kwargs) -> bool | Awaitable[bool]:

"""

Load ACL rules from the configured ``aclfile``.

Note that the server must be configured with the ``aclfile``

directive to be able to load ACL rules from an aclfile.

For more information, see https://redis.io/commands/acl-load

"""

return self.execute_command("ACL LOAD", **kwargs)

@overload

def acl_save(self: SyncClientProtocol, **kwargs) -> bool: ...

@overload

def acl_save(self: AsyncClientProtocol, **kwargs) -> Awaitable[bool]: ...

def acl_save(self, **kwargs) -> bool | Awaitable[bool]:

"""

Save ACL rules to the configured ``aclfile``.

Note that the server must be configured with the ``aclfile``

directive to be able to save ACL rules to an aclfile.

For more information, see https://redis.io/commands/acl-save

"""

return self.execute_command("ACL SAVE", **kwargs)

@overload

def acl_setuser(

self: SyncClientProtocol,

username: str,

enabled: bool = False,

nopass: bool = False,

passwords: str | Iterable[str] | None = None,

hashed_passwords: str | Iterable[str] | None = None,

categories: Iterable[str] | None = None,

commands: Iterable[str] | None = None,

keys: Iterable[KeyT] | None = None,

channels: Iterable[ChannelT] | None = None,

selectors: Iterable[Tuple[str, KeyT]] | None = None,

reset: bool = False,

reset_keys: bool = False,

reset_channels: bool = False,

reset_passwords: bool = False,

**kwargs,

) -> bool: ...

@overload

def acl_setuser(

self: AsyncClientProtocol,

username: str,

enabled: bool = False,

nopass: bool = False,

passwords: str | Iterable[str] | None = None,

hashed_passwords: str | Iterable[str] | None = None,

categories: Iterable[str] | None = None,

commands: Iterable[str] | None = None,

keys: Iterable[KeyT] | None = None,

channels: Iterable[ChannelT] | None = None,

selectors: Iterable[Tuple[str, KeyT]] | None = None,

reset: bool = False,

reset_keys: bool = False,

reset_channels: bool = False,

reset_passwords: bool = False,

**kwargs,

) -> Awaitable[bool]: ...

def acl_setuser(

self,

username: str,

enabled: bool = False,

nopass: bool = False,

passwords: str | Iterable[str] | None = None,

hashed_passwords: str | Iterable[str] | None = None,

categories: Iterable[str] | None = None,

commands: Iterable[str] | None = None,

keys: Iterable[KeyT] | None = None,

channels: Iterable[ChannelT] | None = None,

selectors: Iterable[Tuple[str, KeyT]] | None = None,

reset: bool = False,

reset_keys: bool = False,

reset_channels: bool = False,

reset_passwords: bool = False,

**kwargs,

) -> bool | Awaitable[bool]:

"""

Create or update an ACL user.

Create or update the ACL for `username`. If the user already exists,

the existing ACL is completely overwritten and replaced with the

specified values.

For more information, see https://redis.io/commands/acl-setuser

Args:

username: The name of the user whose ACL is to be created or updated.

enabled: Indicates whether the user should be allowed to authenticate.

Defaults to `False`.

nopass: Indicates whether the user can authenticate without a password.

This cannot be `True` if `passwords` are also specified.

passwords: A list of plain text passwords to add to or remove from the user.

Each password must be prefixed with a '+' to add or a '-' to

remove. For convenience, a single prefixed string can be used

when adding or removing a single password.

hashed_passwords: A list of SHA-256 hashed passwords to add to or remove

from the user. Each hashed password must be prefixed with

a '+' to add or a '-' to remove. For convenience, a single

prefixed string can be used when adding or removing a

single password.

categories: A list of strings representing category permissions. Each string

must be prefixed with either a '+' to add the category

permission or a '-' to remove the category permission.

commands: A list of strings representing command permissions. Each string

must be prefixed with either a '+' to add the command permission

or a '-' to remove the command permission.

keys: A list of key patterns to grant the user access to. Key patterns allow

``'*'`` to support wildcard matching. For example, ``'*'`` grants

access to all keys while ``'cache:*'`` grants access to all keys that

are prefixed with ``cache:``.

`keys` should not be prefixed with a ``'~'``.

reset: Indicates whether the user should be fully reset prior to applying

the new ACL. Setting this to `True` will remove all existing

passwords, flags, and privileges from the user and then apply the

specified rules. If `False`, the user's existing passwords, flags,

and privileges will be kept and any new specified rules will be

applied on top.

reset_keys: Indicates whether the user's key permissions should be reset

prior to applying any new key permissions specified in `keys`.

If `False`, the user's existing key permissions will be kept and

any new specified key permissions will be applied on top.

reset_channels: Indicates whether the user's channel permissions should be

reset prior to applying any new channel permissions

specified in `channels`. If `False`, the user's existing

channel permissions will be kept and any new specified

channel permissions will be applied on top.

reset_passwords: Indicates whether to remove all existing passwords and the

`nopass` flag from the user prior to applying any new

passwords specified in `passwords` or `hashed_passwords`.

If `False`, the user's existing passwords and `nopass`

status will be kept and any new specified passwords or

hashed passwords will be applied on top.

"""

encoder = self.get_encoder()

pieces: List[EncodableT] = [username]

if reset:

pieces.append(b"reset")

if reset_keys:

pieces.append(b"resetkeys")

if reset_channels:

pieces.append(b"resetchannels")

if reset_passwords:

pieces.append(b"resetpass")

if enabled:

pieces.append(b"on")

else:

pieces.append(b"off")

if (passwords or hashed_passwords) and nopass:

raise DataError(

"Cannot set 'nopass' and supply 'passwords' or 'hashed_passwords'"

)

if passwords:

# as most users will have only one password, allow remove_passwords

# to be specified as a simple string or a list

passwords = list_or_args(passwords, [])

for i, password in enumerate(passwords):

password = encoder.encode(password)

if password.startswith(b"+"):

pieces.append(b">%s" % password[1:])

elif password.startswith(b"-"):

pieces.append(b"<%s" % password[1:])

else:

raise DataError(

f"Password {i} must be prefixed with a "

f'"+" to add or a "-" to remove'

)

if hashed_passwords:

# as most users will have only one password, allow remove_passwords

# to be specified as a simple string or a list

hashed_passwords = list_or_args(hashed_passwords, [])

for i, hashed_password in enumerate(hashed_passwords):

hashed_password = encoder.encode(hashed_password)

if hashed_password.startswith(b"+"):

pieces.append(b"#%s" % hashed_password[1:])

elif hashed_password.startswith(b"-"):

pieces.append(b"!%s" % hashed_password[1:])

else:

raise DataError(

f"Hashed password {i} must be prefixed with a "

f'"+" to add or a "-" to remove'

)

if nopass:

pieces.append(b"nopass")

if categories:

for category in categories:

category = encoder.encode(category)

# categories can be prefixed with one of (+@, +, -@, -)

if category.startswith(b"+@"):

pieces.append(category)

elif category.startswith(b"+"):

pieces.append(b"+@%s" % category[1:])

elif category.startswith(b"-@"):

pieces.append(category)

elif category.startswith(b"-"):

pieces.append(b"-@%s" % category[1:])

else:

raise DataError(

f'Category "{encoder.decode(category, force=True)}" '

'must be prefixed with "+" or "-"'

)

if commands:

for cmd in commands:

cmd = encoder.encode(cmd)

if not cmd.startswith(b"+") and not cmd.startswith(b"-"):

raise DataError(

f'Command "{encoder.decode(cmd, force=True)}" '

'must be prefixed with "+" or "-"'

)

pieces.append(cmd)

if keys:

for key in keys:

key = encoder.encode(key)

if not key.startswith(b"%") and not key.startswith(b"~"):

key = b"~%s" % key

pieces.append(key)

if channels:

for channel in channels:

channel = encoder.encode(channel)

pieces.append(b"&%s" % channel)

if selectors:

for cmd, key in selectors:

cmd = encoder.encode(cmd)

if not cmd.startswith(b"+") and not cmd.startswith(b"-"):

raise DataError(

f'Command "{encoder.decode(cmd, force=True)}" '

'must be prefixed with "+" or "-"'

)

key = encoder.encode(key)

if not key.startswith(b"%") and not key.startswith(b"~"):

key = b"~%s" % key

pieces.append(b"(%s %s)" % (cmd, key))

return self.execute_command("ACL SETUSER", *pieces, **kwargs)

@overload

def acl_users(self: SyncClientProtocol, **kwargs) -> list[str]: ...

@overload

def acl_users(self: AsyncClientProtocol, **kwargs) -> Awaitable[list[str]]: ...

def acl_users(self, **kwargs) -> list[str] | Awaitable[list[str]]:

"""Returns a list of all registered users on the server.

For more information, see https://redis.io/commands/acl-users

"""

return self.execute_command("ACL USERS", **kwargs)

@overload

def acl_whoami(self: SyncClientProtocol, **kwargs) -> str: ...

@overload

def acl_whoami(self: AsyncClientProtocol, **kwargs) -> Awaitable[str]: ...

def acl_whoami(self, **kwargs) -> str | Awaitable[str]:

"""Get the username for the current connection

For more information, see https://redis.io/commands/acl-whoami

"""

return self.execute_command("ACL WHOAMI", **kwargs)

AsyncACLCommands = ACLCommands

class HotkeysMetricsTypes(Enum):

CPU = "CPU"

NET = "NET"

class ManagementCommands(CommandsProtocol):

"""

Redis management commands

"""

@overload

def auth(

self: SyncClientProtocol,

password: str,

username: str | None = None,

**kwargs,

) -> bool: ...

@overload

def auth(

self: AsyncClientProtocol,

password: str,

username: str | None = None,

**kwargs,

) -> Awaitable[bool]: ...

def auth(

self, password: str, username: str | None = None, **kwargs

) -> bool | Awaitable[bool]:

"""

Authenticates the user. If you do not pass username, Redis will try to

authenticate for the "default" user. If you do pass username, it will

authenticate for the given user.

For more information, see https://redis.io/commands/auth

"""

pieces = []

if username is not None:

pieces.append(username)

pieces.append(password)

return self.execute_command("AUTH", *pieces, **kwargs)

@overload

def bgrewriteaof(self: SyncClientProtocol, **kwargs) -> bool | bytes | str: ...

@overload

def bgrewriteaof(

self: AsyncClientProtocol, **kwargs

) -> Awaitable[bool | bytes | str]: ...

def bgrewriteaof(self, **kwargs) -> (bool | bytes | str) | Awaitable[

bool | bytes | str

]:

"""Tell the Redis server to rewrite the AOF file from data in memory.

For more information, see https://redis.io/commands/bgrewriteaof

"""

return self.execute_command("BGREWRITEAOF", **kwargs)

@overload

def bgsave(

self: SyncClientProtocol, schedule: bool = True, **kwargs

) -> bool | bytes | str: ...

@overload

def bgsave(

self: AsyncClientProtocol, schedule: bool = True, **kwargs

) -> Awaitable[bool | bytes | str]: ...

def bgsave(self, schedule: bool = True, **kwargs) -> (

bool | bytes | str

) | Awaitable[bool | bytes | str]:

"""

Tell the Redis server to save its data to disk. Unlike save(),

this method is asynchronous and returns immediately.

For more information, see https://redis.io/commands/bgsave

"""

pieces = []

if schedule:

pieces.append("SCHEDULE")

return self.execute_command("BGSAVE", *pieces, **kwargs)

@overload

def role(self: SyncClientProtocol) -> list[Any]: ...

@overload

def role(self: AsyncClientProtocol) -> Awaitable[list[Any]]: ...

def role(self) -> list[Any] | Awaitable[list[Any]]:

"""

Provide information on the role of a Redis instance in

the context of replication, by returning if the instance

is currently a master, slave, or sentinel.

For more information, see https://redis.io/commands/role

"""

return self.execute_command("ROLE")

@overload

def client_kill(self: SyncClientProtocol, address: str, **kwargs) -> bool | int: ...

@overload

def client_kill(

self: AsyncClientProtocol, address: str, **kwargs

) -> Awaitable[bool | int]: ...

def client_kill(self, address: str, **kwargs) -> (bool | int) | Awaitable[

bool | int

]:

"""Disconnects the client at ``address`` (ip:port)

For more information, see https://redis.io/commands/client-kill

"""

return self.execute_command("CLIENT KILL", address, **kwargs)

@overload

def client_kill_filter(

self: SyncClientProtocol,

_id: str | None = None,

_type: str | None = None,

addr: str | None = None,

skipme: bool | None = None,

laddr: bool | None = None,

user: str | None = None,

maxage: int | None = None,

**kwargs,

) -> int: ...

@overload

def client_kill_filter(

self: AsyncClientProtocol,

_id: str | None = None,

_type: str | None = None,

addr: str | None = None,

skipme: bool | None = None,

laddr: bool | None = None,

user: str | None = None,

maxage: int | None = None,

**kwargs,

) -> Awaitable[int]: ...

def client_kill_filter(

self,

_id: str | None = None,

_type: str | None = None,

addr: str | None = None,

skipme: bool | None = None,

laddr: bool | None = None,

user: str | None = None,

maxage: int | None = None,

**kwargs,

) -> int | Awaitable[int]:

"""

Disconnects client(s) using a variety of filter options

:param _id: Kills a client by its unique ID field

:param _type: Kills a client by type where type is one of 'normal',

'master', 'slave' or 'pubsub'

:param addr: Kills a client by its 'address:port'

:param skipme: If True, then the client calling the command

will not get killed even if it is identified by one of the filter

options. If skipme is not provided, the server defaults to skipme=True

:param laddr: Kills a client by its 'local (bind) address:port'

:param user: Kills a client for a specific user name

:param maxage: Kills clients that are older than the specified age in seconds

"""

args = []

if _type is not None:

client_types = ("normal", "master", "slave", "pubsub")

if str(_type).lower() not in client_types:

raise DataError(f"CLIENT KILL type must be one of {client_types!r}")

args.extend((b"TYPE", _type))

if skipme is not None:

if not isinstance(skipme, bool):

raise DataError("CLIENT KILL skipme must be a bool")

if skipme:

args.extend((b"SKIPME", b"YES"))

else:

args.extend((b"SKIPME", b"NO"))

if _id is not None:

args.extend((b"ID", _id))

if addr is not None:

args.extend((b"ADDR", addr))

if laddr is not None:

args.extend((b"LADDR", laddr))

if user is not None:

args.extend((b"USER", user))

if maxage is not None:

args.extend((b"MAXAGE", maxage))

if not args:

raise DataError(

"CLIENT KILL <filter> <value> ... ... <filter> "

"<value> must specify at least one filter"

)

return self.execute_command("CLIENT KILL", *args, **kwargs)

@overload

def client_info(self: SyncClientProtocol, **kwargs) -> dict[str, str | int]: ...

@overload

def client_info(

self: AsyncClientProtocol, **kwargs

) -> Awaitable[dict[str, str | int]]: ...

def client_info(

self, **kwargs

) -> dict[str, str | int] | Awaitable[dict[str, str | int]]:

"""

Returns information and statistics about the current

client connection.

For more information, see https://redis.io/commands/client-info

"""

return self.execute_command("CLIENT INFO", **kwargs)

@overload

def client_list(

self: SyncClientProtocol,

_type: str | None = None,

client_id: List[EncodableT] = [],

**kwargs,

) -> list[dict[str, str]]: ...

@overload

def client_list(

self: AsyncClientProtocol,

_type: str | None = None,

client_id: List[EncodableT] = [],

**kwargs,

) -> Awaitable[list[dict[str, str]]]: ...

def client_list(

self, _type: str | None = None, client_id: List[EncodableT] = [], **kwargs

) -> list[dict[str, str]] | Awaitable[list[dict[str, str]]]:

"""

Returns a list of currently connected clients.

If type of client specified, only that type will be returned.

:param _type: optional. one of the client types (normal, master,

replica, pubsub)

:param client_id: optional. a list of client ids

For more information, see https://redis.io/commands/client-list

"""

args = []

if _type is not None:

client_types = ("normal", "master", "replica", "pubsub")

if str(_type).lower() not in client_types:

raise DataError(f"CLIENT LIST _type must be one of {client_types!r}")

args.append(b"TYPE")

args.append(_type)

if not isinstance(client_id, list):

raise DataError("client_id must be a list")

if client_id:

args.append(b"ID")

args += client_id

return self.execute_command("CLIENT LIST", *args, **kwargs)

@overload

def client_getname(self: SyncClientProtocol, **kwargs) -> str | None: ...

@overload

def client_getname(

self: AsyncClientProtocol, **kwargs

) -> Awaitable[str | None]: ...

def client_getname(self, **kwargs) -> (str | None) | Awaitable[str | None]:

"""

Returns the current connection name

For more information, see https://redis.io/commands/client-getname

"""

return self.execute_command("CLIENT GETNAME", **kwargs)

@overload

def client_getredir(self: SyncClientProtocol, **kwargs) -> int: ...

@overload

def client_getredir(self: AsyncClientProtocol, **kwargs) -> Awaitable[int]: ...

def client_getredir(self, **kwargs) -> int | Awaitable[int]:

"""

Returns the ID (an integer) of the client to whom we are

redirecting tracking notifications.

see: https://redis.io/commands/client-getredir

"""

return self.execute_command("CLIENT GETREDIR", **kwargs)

@overload

def client_reply(

self: SyncClientProtocol,

reply: Literal["ON", "OFF", "SKIP"],

**kwargs,

) -> bytes | str: ...

@overload

def client_reply(

self: AsyncClientProtocol,

reply: Literal["ON", "OFF", "SKIP"],

**kwargs,

) -> Awaitable[bytes | str]: ...

def client_reply(self, reply: Literal["ON", "OFF", "SKIP"], **kwargs) -> (

bytes | str

) | Awaitable[bytes | str]:

"""

Enable and disable redis server replies.

``reply`` Must be ON OFF or SKIP,

ON - The default most with server replies to commands

OFF - Disable server responses to commands

SKIP - Skip the response of the immediately following command.

Note: When setting OFF or SKIP replies, you will need a client object

with a timeout specified in seconds, and will need to catch the

TimeoutError.

The test_client_reply unit test illustrates this, and

conftest.py has a client with a timeout.

See https://redis.io/commands/client-reply

"""

replies = ["ON", "OFF", "SKIP"]

if reply not in replies:

raise DataError(f"CLIENT REPLY must be one of {replies!r}")

return self.execute_command("CLIENT REPLY", reply, **kwargs)

@overload

def client_id(self: SyncClientProtocol, **kwargs) -> int: ...

@overload

def client_id(self: AsyncClientProtocol, **kwargs) -> Awaitable[int]: ...

def client_id(self, **kwargs) -> int | Awaitable[int]:

"""

Returns the current connection id

For more information, see https://redis.io/commands/client-id

"""

return self.execute_command("CLIENT ID", **kwargs)

@overload

def client_tracking_on(

self: SyncClientProtocol,

clientid: int | None = None,

prefix: Sequence[KeyT] = [],

bcast: bool = False,

optin: bool = False,

optout: bool = False,

noloop: bool = False,

) -> bytes | str: ...

@overload

def client_tracking_on(

self: AsyncClientProtocol,

clientid: int | None = None,

prefix: Sequence[KeyT] = [],

bcast: bool = False,

optin: bool = False,

optout: bool = False,

noloop: bool = False,

) -> Awaitable[bytes | str]: ...

def client_tracking_on(

self,

clientid: int | None = None,

prefix: Sequence[KeyT] = [],

bcast: bool = False,

optin: bool = False,

optout: bool = False,

noloop: bool = False,

) -> (bytes | str) | Awaitable[bytes | str]:

"""

Turn on the tracking mode.

For more information, about the options look at client_tracking func.

See https://redis.io/commands/client-tracking

"""

return self.client_tracking(

True, clientid, prefix, bcast, optin, optout, noloop

)

@overload

def client_tracking_off(

self: SyncClientProtocol,

clientid: int | None = None,

prefix: Sequence[KeyT] = [],

bcast: bool = False,

optin: bool = False,

optout: bool = False,

noloop: bool = False,

) -> bytes | str: ...