from typing import Any, overload, Coroutine, Awaitable


class RedisError(Exception): ...


class Redis:
    """High-level Redis client with all Redis commands."""

    encoding: str | None

    @overload
    def __init__(self, pool_or_conn: Any) -> None: ...

    @overload
    def __init__(
        self,
        pool_or_conn: Any = ...,
        *,
        host: str,
        port: int,
        db: int = ...,
        password: str | None = ...,
        encoding: str | None = ...,
        ssl: Any = ...,
    ) -> None: ...

    # ==================== Connection Commands ====================
    def close(self) -> None: ...
    async def wait_closed(self) -> None: ...
    async def auth(self, password: str) -> bool: ...
    async def select(self, db: int) -> bool: ...
    async def quit(self) -> bool: ...
    async def swapdb(self, db1: int, db2: int) -> bool: ...

    # ==================== String Commands ====================
    async def append(self, key: str | bytes, value: str | bytes) -> int: ...
    async def bitcount(self, key: str | bytes, start: int | None = ..., end: int | None = ...) -> int: ...
    async def bitop_and(self, destkey: str | bytes, key: str | bytes, *keys: str | bytes) -> int: ...
    async def bitop_or(self, destkey: str | bytes, key: str | bytes, *keys: str | bytes) -> int: ...
    async def bitop_xor(self, destkey: str | bytes, key: str | bytes, *keys: str | bytes) -> int: ...
    async def bitop_not(self, destkey: str | bytes, key: str | bytes) -> int: ...
    async def bitpos(self, key: str | bytes, bit: int, start: int | None = ..., end: int | None = ...) -> int: ...
    async def decr(self, key: str | bytes) -> int: ...
    async def decrby(self, key: str | bytes, decrement: int) -> int: ...
    async def get(self, key: str | bytes, *, encoding: str | None = ...) -> bytes | str | None: ...
    async def getbit(self, key: str | bytes, offset: int) -> int: ...
    async def getrange(self, key: str | bytes, start: int, end: int, *, encoding: str | None = ...) -> bytes | str: ...
    async def getset(self, key: str | bytes, value: str | bytes, *, encoding: str | None = ...) -> bytes | str | None: ...
    async def incr(self, key: str | bytes) -> int: ...
    async def incrby(self, key: str | bytes, increment: int) -> int: ...
    async def incrbyfloat(self, key: str | bytes, increment: float) -> float: ...
    async def mget(self, key: str | bytes, *keys: str | bytes, encoding: str | None = ...) -> list[Any]: ...
    async def mset(self, *args: Any) -> bool: ...
    async def msetnx(self, key: str | bytes, value: str | bytes, *pairs: Any) -> bool: ...
    async def psetex(self, key: str | bytes, milliseconds: int, value: str | bytes) -> bool: ...
    async def set(self, key: str | bytes, value: str | bytes, *, expire: int = ..., pexpire: int = ..., exist: str | None = ...) -> bool: ...
    async def setbit(self, key: str | bytes, offset: int, value: int) -> int: ...
    async def setex(self, key: str | bytes, seconds: int, value: str | bytes) -> bool: ...
    async def setnx(self, key: str | bytes, value: str | bytes) -> bool: ...
    async def setrange(self, key: str | bytes, offset: int, value: str | bytes) -> int: ...
    async def strlen(self, key: str | bytes) -> int: ...

    # ==================== List Commands ====================
    async def blpop(self, key: str | bytes, *keys: str | bytes, timeout: int = ..., encoding: str | None = ...) -> list[Any] | None: ...
    async def brpop(self, key: str | bytes, *keys: str | bytes, timeout: int = ..., encoding: str | None = ...) -> list[Any] | None: ...
    async def brpoplpush(self, sourcekey: str | bytes, destkey: str | bytes, timeout: int = ..., encoding: str | None = ...) -> bytes | str | None: ...
    async def lindex(self, key: str | bytes, index: int, *, encoding: str | None = ...) -> bytes | str | None: ...
    async def linsert(self, key: str | bytes, pivot: str | bytes, value: str | bytes, before: bool = ...) -> int: ...
    async def llen(self, key: str | bytes) -> int: ...
    async def lpop(self, key: str | bytes, *, encoding: str | None = ...) -> bytes | str | None: ...
    async def lpush(self, key: str | bytes, value: str | bytes, *values: str | bytes) -> int: ...
    async def lpushx(self, key: str | bytes, value: str | bytes) -> int: ...
    async def lrange(self, key: str | bytes, start: int, stop: int, *, encoding: str | None = ...) -> list[Any]: ...
    async def lrem(self, key: str | bytes, count: int, value: str | bytes) -> int: ...
    async def lset(self, key: str | bytes, index: int, value: str | bytes) -> bool: ...
    async def ltrim(self, key: str | bytes, start: int, stop: int) -> bool: ...
    async def rpop(self, key: str | bytes, *, encoding: str | None = ...) -> bytes | str | None: ...
    async def rpoplpush(self, sourcekey: str | bytes, destkey: str | bytes, *, encoding: str | None = ...) -> bytes | str | None: ...
    async def rpush(self, key: str | bytes, value: str | bytes, *values: str | bytes) -> int: ...
    async def rpushx(self, key: str | bytes, value: str | bytes) -> int: ...

    # ==================== Set Commands ====================
    async def sadd(self, key: str | bytes, member: str | bytes, *members: str | bytes) -> int: ...
    async def scard(self, key: str | bytes) -> int: ...
    async def sdiff(self, key: str | bytes, *keys: str | bytes, encoding: str | None = ...) -> list[Any]: ...
    async def sdiffstore(self, destkey: str | bytes, key: str | bytes, *keys: str | bytes) -> int: ...
    async def sinter(self, key: str | bytes, *keys: str | bytes, encoding: str | None = ...) -> list[Any]: ...
    async def sinterstore(self, destkey: str | bytes, key: str | bytes, *keys: str | bytes) -> int: ...
    async def sismember(self, key: str | bytes, member: str | bytes) -> bool: ...
    async def smembers(self, key: str | bytes, *, encoding: str | None = ...) -> list[Any]: ...
    async def smove(self, sourcekey: str | bytes, destkey: str | bytes, member: str | bytes) -> bool: ...
    async def spop(self, key: str | bytes, count: int | None = ..., *, encoding: str | None = ...) -> bytes | str | list[Any] | None: ...
    async def srandmember(self, key: str | bytes, count: int | None = ..., *, encoding: str | None = ...) -> bytes | str | list[Any] | None: ...
    async def srem(self, key: str | bytes, member: str | bytes, *members: str | bytes) -> int: ...
    async def sscan(self, key: str | bytes, cursor: int = ..., match: str | None = ..., count: int | None = ...) -> tuple[int, list[Any]]: ...

    # ==================== Hash Commands ====================
    async def hdel(self, key: str | bytes, field: str | bytes, *fields: str | bytes) -> int: ...
    async def hexists(self, key: str | bytes, field: str | bytes) -> bool: ...
    async def hget(self, key: str | bytes, field: str | bytes, *, encoding: str | None = ...) -> bytes | str | None: ...
    async def hgetall(self, key: str | bytes, *, encoding: str | None = ...) -> dict[Any, Any]: ...
    async def hincrby(self, key: str | bytes, field: str | bytes, increment: int) -> int: ...
    async def hincrbyfloat(self, key: str | bytes, field: str | bytes, increment: float) -> float: ...
    async def hkeys(self, key: str | bytes, *, encoding: str | None = ...) -> list[Any]: ...
    async def hlen(self, key: str | bytes) -> int: ...
    async def hmget(self, key: str | bytes, field: str | bytes, *fields: str | bytes, encoding: str | None = ...) -> list[Any]: ...
    async def hmset(self, key: str | bytes, field: str | bytes, value: str | bytes, *fv_pairs: Any) -> bool: ...
    async def hmset_dict(self, key: str | bytes, mapping: dict[Any, Any]) -> bool: ...
    async def hset(self, key: str | bytes, field: str | bytes, value: str | bytes) -> int: ...
    async def hsetnx(self, key: str | bytes, field: str | bytes, value: str | bytes) -> bool: ...
    async def hstrlen(self, key: str | bytes, field: str | bytes) -> int: ...
    async def hvals(self, key: str | bytes, *, encoding: str | None = ...) -> list[Any]: ...
    async def hscan(self, key: str | bytes, cursor: int = ..., match: str | None = ..., count: int | None = ...) -> tuple[int, list[Any]]: ...

    # ==================== Sorted Set Commands ====================
    async def zadd(self, key: str | bytes, score: float, member: str | bytes, *score_members: Any) -> int: ...
    async def zcard(self, key: str | bytes) -> int: ...
    async def zcount(self, key: str | bytes, min: float | str, max: float | str) -> int: ...
    async def zincrby(self, key: str | bytes, increment: float, member: str | bytes) -> float: ...
    async def zinterstore(self, destkey: str | bytes, numkeys: int, key: str | bytes, *keys: str | bytes, weights: list[int] | None = ..., aggregate: str | None = ...) -> int: ...
    async def zlexcount(self, key: str | bytes, min: str, max: str) -> int: ...
    async def zpopmax(self, key: str | bytes, count: int | None = ..., encoding: str | None = ...) -> list[Any]: ...
    async def zpopmin(self, key: str | bytes, count: int | None = ..., encoding: str | None = ...) -> list[Any]: ...
    async def zrange(self, key: str | bytes, start: int, stop: int, *, withscores: bool = ..., encoding: str | None = ...) -> list[Any]: ...
    async def zrangebylex(self, key: str | bytes, min: str, max: str, *, offset: int | None = ..., count: int | None = ..., encoding: str | None = ...) -> list[Any]: ...
    async def zrangebyscore(self, key: str | bytes, min: float | str, max: float | str, *, offset: int | None = ..., count: int | None = ..., withscores: bool = ..., encoding: str | None = ...) -> list[Any]: ...
    async def zrank(self, key: str | bytes, member: str | bytes) -> int | None: ...
    async def zrem(self, key: str | bytes, member: str | bytes, *members: str | bytes) -> int: ...
    async def zremrangebylex(self, key: str | bytes, min: str, max: str) -> int: ...
    async def zremrangebyrank(self, key: str | bytes, start: int, stop: int) -> int: ...
    async def zremrangebyscore(self, key: str | bytes, min: float | str, max: float | str) -> int: ...
    async def zrevrange(self, key: str | bytes, start: int, stop: int, *, withscores: bool = ..., encoding: str | None = ...) -> list[Any]: ...
    async def zrevrangebylex(self, key: str | bytes, max: str, min: str, *, offset: int | None = ..., count: int | None = ..., encoding: str | None = ...) -> list[Any]: ...
    async def zrevrangebyscore(self, key: str | bytes, max: float | str, min: float | str, *, offset: int | None = ..., count: int | None = ..., withscores: bool = ..., encoding: str | None = ...) -> list[Any]: ...
    async def zrevrank(self, key: str | bytes, member: str | bytes) -> int | None: ...
    async def zscan(self, key: str | bytes, cursor: int = ..., match: str | None = ..., count: int | None = ...) -> tuple[int, list[Any]]: ...
    async def zscore(self, key: str | bytes, member: str | bytes) -> float | None: ...
    async def zunionstore(self, destkey: str | bytes, numkeys: int, key: str | bytes, *keys: str | bytes, weights: list[int] | None = ..., aggregate: str | None = ...) -> int: ...
    async def bzpopmax(self, key: str | bytes, *keys: str | bytes, timeout: int = ..., encoding: str | None = ...) -> list[Any] | None: ...
    async def bzpopmin(self, key: str | bytes, *keys: str | bytes, timeout: int = ..., encoding: str | None = ...) -> list[Any] | None: ...

    # ==================== Generic/Key Commands ====================
    async def delete(self, key: str | bytes, *keys: str | bytes) -> int: ...
    async def dump(self, key: str | bytes) -> bytes | None: ...
    async def exists(self, key: str | bytes, *keys: str | bytes) -> int: ...
    async def expire(self, key: str | bytes, timeout: int) -> bool: ...
    async def expireat(self, key: str | bytes, timestamp: int) -> bool: ...
    async def keys(self, pattern: str | bytes, *, encoding: str | None = ...) -> list[Any]: ...
    async def migrate(self, host: str, port: int, key: str | bytes, dest_db: int, timeout: int, *, copy: bool = ..., replace: bool = ...) -> bool: ...
    async def move(self, key: str | bytes, db: int) -> bool: ...
    async def object_encoding(self, key: str | bytes) -> bytes | str: ...
    async def object_idletime(self, key: str | bytes) -> int | None: ...
    async def object_refcount(self, key: str | bytes) -> int | None: ...
    async def persist(self, key: str | bytes) -> bool: ...
    async def pexpire(self, key: str | bytes, timeout: int) -> bool: ...
    async def pexpireat(self, key: str | bytes, timestamp: int) -> bool: ...
    async def pttl(self, key: str | bytes) -> int: ...
    async def randomkey(self, *, encoding: str | None = ...) -> bytes | str | None: ...
    async def rename(self, key: str | bytes, newkey: str | bytes) -> bool: ...
    async def renamenx(self, key: str | bytes, newkey: str | bytes) -> bool: ...
    async def restore(self, key: str | bytes, ttl: int, value: bytes) -> bool: ...
    async def scan(self, cursor: int = ..., match: str | None = ..., count: int | None = ...) -> tuple[int, list[Any]]: ...
    async def sort(self, key: str | bytes, *get_patterns: str | bytes, by: str | None = ..., offset: int | None = ..., count: int | None = ..., asc: bool | None = ..., alpha: bool = ..., store: str | bytes | None = ...) -> list[Any]: ...
    async def touch(self, key: str | bytes, *keys: str | bytes) -> int: ...
    async def ttl(self, key: str | bytes) -> int: ...
    async def type(self, key: str | bytes) -> bytes: ...
    async def unlink(self, key: str | bytes, *keys: str | bytes) -> int: ...
    async def wait(self, numslaves: int, timeout: int) -> int: ...

    # ==================== Pub/Sub Commands ====================
    async def publish(self, channel: str | bytes, message: str | bytes) -> int: ...
    async def publish_json(self, channel: str | bytes, obj: Any) -> int: ...
    async def pubsub_channels(self, pattern: str | None = ..., *, encoding: str | None = ...) -> list[Any]: ...
    async def pubsub_numpat(self) -> int: ...
    async def pubsub_numsub(self, *channels: str | bytes) -> dict[Any, int]: ...
    async def subscribe(self, channel: str | bytes, *channels: str | bytes) -> list[Any]: ...
    async def unsubscribe(self, channel: str | bytes | None = ..., *channels: str | bytes) -> list[Any]: ...
    async def psubscribe(self, pattern: str | bytes, *patterns: str | bytes) -> list[Any]: ...
    async def punsubscribe(self, pattern: str | bytes | None = ..., *patterns: str | bytes) -> list[Any]: ...

    # ==================== Transaction Commands ====================
    async def multi_exec(self, *commands: Any) -> list[Any]: ...
    async def watch(self, key: str | bytes, *keys: str | bytes) -> bool: ...
    async def unwatch(self) -> bool: ...

    # ==================== Scripting Commands ====================
    async def eval(self, script: str | bytes, keys: list[str | bytes] | None = ..., args: list[Any] | None = ...) -> Any: ...
    async def evalsha(self, digest: str | bytes, keys: list[str | bytes] | None = ..., args: list[Any] | None = ...) -> Any: ...
    async def script_exists(self, digest: str | bytes, *digests: str | bytes) -> list[bool]: ...
    async def script_flush(self) -> bool: ...
    async def script_kill(self) -> bool: ...
    async def script_load(self, script: str | bytes) -> bytes: ...

    # ==================== Server Commands ====================
    async def bgrewriteaof(self) -> bytes: ...
    async def bgsave(self) -> bytes: ...
    async def client_getname(self, *, encoding: str | None = ...) -> bytes | str | None: ...
    async def client_kill(self, *args: Any) -> int: ...
    async def client_list(self, *, encoding: str | None = ...) -> bytes | str: ...
    async def client_pause(self, timeout: int) -> bool: ...
    async def client_reply(self, *args: Any) -> bool: ...
    async def client_setname(self, name: str | bytes) -> bool: ...
    async def command(self) -> list[Any]: ...
    async def command_count(self) -> int: ...
    async def command_getkeys(self, command: str | bytes, *args: Any, encoding: str = ...) -> list[Any]: ...
    async def command_info(self, command: str | bytes, *commands: str | bytes) -> list[Any]: ...
    async def config_get(self, parameter: str | bytes = ..., *, encoding: str | None = ...) -> dict[Any, Any]: ...
    async def config_resetstat(self) -> bool: ...
    async def config_rewrite(self) -> bool: ...
    async def config_set(self, parameter: str | bytes, value: str | bytes) -> bool: ...
    async def dbsize(self) -> int: ...
    async def debug_object(self, key: str | bytes) -> bytes: ...
    async def debug_sleep(self, timeout: float) -> None: ...
    async def flushall(self, async_mode: bool = ...) -> bool: ...
    async def flushdb(self, async_mode: bool = ...) -> bool: ...
    async def info(self, section: str | None = ..., *, encoding: str | None = ...) -> bytes | str | dict[str, Any]: ...
    async def lastsave(self) -> int: ...
    async def monitor(self) -> Any: ...
    async def ping(self, message: str | bytes | None = ..., *, encoding: str | None = ...) -> bytes | str: ...
    async def save(self) -> bool: ...
    async def shutdown(self, save: bool = ...) -> None: ...
    async def slaveof(self, host: str, port: int) -> bool: ...
    async def slowlog_get(self, count: int | None = ...) -> list[Any]: ...
    async def slowlog_len(self) -> int: ...
    async def slowlog_reset(self) -> bool: ...
    async def sync(self) -> Any: ...
    async def time(self) -> tuple[int, int]: ...
    async def role(self) -> list[Any]: ...

    # ==================== Geo Commands ====================
    async def geoadd(self, key: str | bytes, longitude: float, latitude: float, member: str | bytes, *latlng_member: Any) -> int: ...
    async def geodist(self, key: str | bytes, member1: str | bytes, member2: str | bytes, unit: str | None = ...) -> float | None: ...
    async def geohash(self, key: str | bytes, member: str | bytes, *members: str | bytes, encoding: str | None = ...) -> list[Any]: ...
    async def geopos(self, key: str | bytes, member: str | bytes, *members: str | bytes) -> list[Any]: ...
    async def georadius(self, key: str | bytes, longitude: float, latitude: float, radius: float, unit: str, *, count: int | None = ..., sort: str | None = ..., encoding: str | None = ...) -> list[Any]: ...
    async def georadiusbymember(self, key: str | bytes, member: str | bytes, radius: float, unit: str, *, count: int | None = ..., sort: str | None = ..., encoding: str | None = ...) -> list[Any]: ...

    # ==================== HyperLogLog Commands ====================
    async def pfadd(self, key: str | bytes, element: str | bytes, *elements: str | bytes) -> int: ...
    async def pfcount(self, key: str | bytes, *keys: str | bytes) -> int: ...
    async def pfmerge(self, destkey: str | bytes, sourcekey: str | bytes, *sourcekeys: str | bytes) -> bool: ...

    # ==================== Stream Commands ====================
    async def xack(self, key: str | bytes, group: str | bytes, id: str | bytes, *ids: str | bytes) -> int: ...
    async def xadd(self, key: str | bytes, id: str | bytes, *field_value_pairs: Any) -> bytes: ...
    async def xclaim(self, key: str | bytes, group: str | bytes, consumer: str | bytes, min_idle_time: int, id: str | bytes, *ids: str | bytes) -> list[Any]: ...
    async def xdel(self, key: str | bytes, id: str | bytes, *ids: str | bytes) -> int: ...
    async def xgroup_create(self, key: str | bytes, group: str | bytes, id: str | bytes = ...) -> bool: ...
    async def xgroup_delconsumer(self, key: str | bytes, group: str | bytes, consumer: str | bytes) -> int: ...
    async def xgroup_destroy(self, key: str | bytes, group: str | bytes) -> int: ...
    async def xgroup_setid(self, key: str | bytes, group: str | bytes, id: str | bytes) -> bool: ...
    async def xlen(self, key: str | bytes) -> int: ...
    async def xpending(self, key: str | bytes, group: str | bytes) -> list[Any]: ...
    async def xrange(self, key: str | bytes, start: str | bytes, end: str | bytes, count: int | None = ...) -> list[Any]: ...
    async def xread(self, *keys_ids: Any, count: int | None = ..., timeout: int | None = ...) -> list[Any]: ...
    async def xread_group(self, group: str | bytes, consumer: str | bytes, *keys_ids: Any, count: int | None = ..., timeout: int | None = ...) -> list[Any]: ...
    async def xrevrange(self, key: str | bytes, end: str | bytes, start: str | bytes, count: int | None = ...) -> list[Any]: ...
    async def xtrim(self, key: str | bytes, maxlen: int, approximate: bool = ...) -> int: ...

    # ==================== Connection Pool Properties ====================
    @property
    def db(self) -> int: ...
    @property
    def closed(self) -> bool: ...
    @property
    def address(self) -> tuple[str, int] | str: ...
    @property
    def connection(self) -> Any: ...
    @property
    def in_transaction(self) -> bool: ...


async def create_redis(
    address: str | tuple[str, int],
    *,
    db: int = ...,
    password: str | None = ...,
    encoding: str | None = ...,
    ssl: Any = ...,
    minsize: int = ...,
    maxsize: int = ...,
    parser: Any = ...,
    create_connection_timeout: float | None = ...,
) -> Redis: ...


async def create_redis_pool(
    address: str | tuple[str, int],
    *,
    db: int = ...,
    password: str | None = ...,
    encoding: str | None = ...,
    ssl: Any = ...,
    minsize: int = ...,
    maxsize: int = ...,
    parser: Any = ...,
    create_connection_timeout: float | None = ...,
) -> Redis: ...


# Errors and Exceptions
class ProtocolError(RedisError): ...
class ReplyError(RedisError): ...
class ConnectionClosedError(RedisError): ...
class ConnectionForcedCloseError(RedisError): ...
class MasterNotFoundError(RedisError): ...
class MultiExecError(RedisError): ...
class PipelineError(RedisError): ...
class ReadOnlyError(RedisError): ...
class MaxClientsError(RedisError): ...
class AuthError(RedisError): ...
class ChannelClosedError(RedisError): ...
class WatchVariableError(RedisError): ...
class PoolClosedError(RedisError): ...
class SlaveNotFoundError(RedisError): ...
class MasterReplyError(RedisError): ...
class SlaveReplyError(RedisError): ...


__all__: tuple[str, ...]