Improved type annotations

[mahenzon]

Pull Request check-list

Reviewed and checked all of these items:

• tests and lints pass with this change - checked in my fork: Improved type annotations mahenzon/redis-py#3
• CI tests pass with this change - I enabled it first in my forked repo and waited for the GitHub action to finish - checked in my fork: Improved type annotations mahenzon/redis-py#3
• Is the new or changed code fully tested? - example provided below
• Is a documentation update included (if this change modifies existing APIs, or introduces new ones)? - no API changes
• Is there an example added to the examples folder (if applicable)? - Do I need to add this checks.py file example to the repo?
• Was the change added to CHANGES file? - Work in progress.

Description of change

Hello, there're some issues with type annotations: #2399 #3169

Example:

from typing import reveal_type

from redis import Redis

redis = Redis(
    host="localhost",
    port=6379,
    decode_responses=True,
)


res = redis.get("some-key")
reveal_type(res)
# note: Revealed type is "Union[typing.Awaitable[Any], Any]"

We can see that sync redis client may return awaitable result, but it will never do.

This is made for compatibility with async redis, but it introduces some challenges when checking code with static type checkers like mypy.

Also it's Any instead of str or bytes because we can't predict, if decode_responses is True or False - it'll be addressed later.

I'd like to make it work this way:

from typing import reveal_type

from redis import Redis
from redis.asyncio import Redis as AsyncRedis

redis = Redis(
    host="localhost",
    port=6379,
    decode_responses=True,
)

async_redis = AsyncRedis(
    host="localhost",
    port=6379,
    decode_responses=True,
)


res = redis.get("some-key")
reveal_type(res)
# note: Revealed type is "Union[builtins.str, None]"

coro = async_redis.get("some-key")
reveal_type(coro)
# note: Revealed type is "typing.Awaitable[Union[builtins.str, None]]"

I started reworking annotations, so type annotations for sync / async redis work as expected - sync redis doesn't return Awaitable, async redis returns Awaitable.

Important

The goal is not to make the whole redis-py source code mypy-comliant and fully annotated.
The goal is to make usage of redis-py in python projects compatible with mypy - make return type annotations more precise. So devs don't need to add casts and asserts to their code.

Example code. where I checked new typing

File: checks.py:

import random
import string
from typing import Optional, List, reveal_type

from redis import Redis
from redis.asyncio import Redis as AsyncRedis

redis = Redis(
    host="localhost",
    port=6379,
    decode_responses=True,
)

async_redis = AsyncRedis(
    host="localhost",
    port=6379,
    decode_responses=True,
)


def get_string() -> Optional[str]:
    res = redis.get("some-key")
    reveal_type(res)  # checks.py:23: note: Revealed type is "Union[builtins.str, None]"
    return res


async def async_get_string() -> Optional[str]:
    coro = async_redis.get("some-key")
    reveal_type(coro)  # checks.py:29: note: Revealed type is "typing.Awaitable[Union[builtins.str, None]]"
    res = await coro
    reveal_type(res)  # checks.py:31: note: Revealed type is "Union[builtins.str, None]"
    return res


def get_values() -> List[str]:
    vals = redis.hvals(name="name")
    reveal_type(vals)  # checks.py:37: note: Revealed type is "builtins.list[builtins.str]"
    res = [v.lower() for v in vals]
    reveal_type(res)  # checks.py:39: note: Revealed type is "builtins.list[builtins.str]"
    return res


async def async_get_values() -> List[str]:
    coro = async_redis.hvals(name="name")
    reveal_type(coro)  # checks.py:45: note: Revealed type is "typing.Awaitable[builtins.list[builtins.str]]"
    vals = await coro
    reveal_type(vals)  # checks.py:47: note: Revealed type is "builtins.list[builtins.str]"
    res = [v.lower() for v in vals]
    reveal_type(res)  # checks.py:49: note: Revealed type is "builtins.list[builtins.str]"
    return res


def checks() -> None:
    hget_res = redis.hget("hash-name", "key")
    reveal_type(hget_res)  # checks.py:55: note: Revealed type is "Union[builtins.str, None]"
    brpoplpush_res = redis.brpoplpush("src", "dst")
    reveal_type(brpoplpush_res)  # checks.py:57: note: Revealed type is "Union[builtins.str, None]"
    lindex_res = redis.lindex("name", 0)
    reveal_type(lindex_res)  # checks.py:59: note: Revealed type is "Union[builtins.str, None]"
    append_res = redis.append("key", "value")
    reveal_type(append_res)  # checks.py:61: note: Revealed type is "builtins.int"


async def async_checks() -> None:
    hget_res = await async_redis.hget("hash-name", "key")
    reveal_type(hget_res)  # checks.py:67: note: Revealed type is "Union[builtins.str, None]"
    brpoplpush_res = await async_redis.brpoplpush("src", "dst")
    reveal_type(brpoplpush_res)  # checks.py:69: note: Revealed type is "Union[builtins.str, None]"
    lindex_res = await async_redis.lindex("name", 0)
    reveal_type(lindex_res)  # checks.py:71: note: Revealed type is "Union[builtins.str, None]"
    append_res = await async_redis.append("key", "value")
    reveal_type(append_res)  # checks.py:73: note: Revealed type is "builtins.int"


def main() -> None:
    bitop_res = redis.bitop("NOT", "result", "key1")
    print(bitop_res)  # 0
    redis.set("foo", "val")
    bitop_res = redis.bitop("NOT", "res1", "foo")
    print(bitop_res)  # 3
    # decode response: - there's an issue decoding such result - works only with decode_responses=False
    # print(redis.get("res1"))
    res = redis.copy("foo", "bar")
    print("res:", res)  # res: True
    res = redis.copy("foo", "".join(random.choices(string.ascii_letters, k=20)))
    print("res:", res)  # res: True

    list_name = "".join(random.choices(string.ascii_letters, k=10))
    res_rpush = redis.rpush(list_name, "qwerty", "foobar")
    print("res rpush:", res_rpush)  # res rpush: 2
    print("list values", redis.lrange(list_name, 0, -1))  # list values ['qwerty', 'foobar']
    res_lset = redis.lset(list_name, 0, "some-val")
    print("res lset:", res_lset)  # res lset: True
    print("list values", redis.lrange(list_name, 0, -1))  # list values ['some-val', 'foobar']

    res_hsetnx = redis.hsetnx("hash-name", "key", "value")
    print("res hsetnx:", res_hsetnx)  # res hsetnx: True

    set_name = "some-set"
    some_set_value = "".join(random.choices(string.ascii_letters, k=10))
    another_set_value = "".join(random.choices(string.ascii_letters, k=10))
    print("is member", redis.sismember(set_name, some_set_value))  # is member False
    redis.sadd(set_name, some_set_value)
    print("is member", redis.sismember(set_name, some_set_value))  # is member True
    print("are members", redis.smismember(set_name, [some_set_value, another_set_value]))  # are members [True, False]


if __name__ == '__main__':
    main()

I checked it with mypy with strict=True enabled: mypy --strict checks.py, added comments to the code example above.

If these changes are ok for redis-py, I'll continue working on the update.

---

There's one major issue: these annotations are ok when decode_responses=True, but if not passed, or decode_responses=False, then some methods will return bytes instead of str - I'm working on a solution for this.

---

Also I fixed parsing for some commands which should return bools: 23ff994

[petyaslavova]

Hi @mahenzon, thank you for your contribution. We will go over it soon.

[trajano]

Would this solve #3574?

[mahenzon]

Would this solve #3574?

Hi @trajano, thank you for mentioning this issue, I'll check it too.

[Copilot]

Pull Request Overview

This PR improves type annotations across the codebase to ensure that both synchronous and asynchronous Redis clients report more precise return types that are compatible with static type checkers. Key changes include the introduction of new type variables in redis/typing.py, refined method return types in redis/commands/core.py (including generic-based command classes), and updates to command parser mappings in redis/_parsers/helpers.py.

Reviewed Changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 2 comments.

File	Description
redis/typing.py	Added new type definitions and ResponseType* type variables
redis/commands/helpers.py	Updated function signature for list_or_args to support Optional args
redis/commands/core.py	Refactored return types for various command methods using generics
redis/_parsers/helpers.py	Adjusted SMISMEMBER parser entries

[Copilot]

The return type for smismember is annotated as ResponseTypeBoolean but the underlying parser returns a list of booleans. Consider defining and using a dedicated type (e.g., ResponseTypeBooleanList) to accurately reflect the returned value.

[Copilot]

[nitpick] Duplicate entries for the SMISMEMBER key are added in this dictionary. Consolidating these duplicates could improve maintainability and clarity.

[connebs]

Would love to see this merged!

[methane]

Suggested change

	StringListType = List[str]
	StringListType = list[str]

[trajano]

Though it won't be a problem for me since I'm on the latest Python, but using the small list version will cause issues with those on 3.8 which means something like this cannot be used with say Kodi

[methane]

Suggested change

	bound=Union[Awaitable[BooleanType], BooleanType],
	bound=Union[Awaitable[bool], bool],

[trajano]

If we're not after 3.8 compatibility, then I think it should be

bound=bool  | Awaitable[bool]

ruff has a rule that would do that swap along with removing Optional in place of | None

[Harry-Lees]

If we're not after 3.8 compatibility, then I think it should be

According to the README, the current version of this project (6.2.0) seems to support Python 3.9+ meaning the list[str] syntax above is usable, but the bool | Awaitable[bool] syntax would not be, as this was introduced in Python 3.10 (source)

[methane]

Suggested change

	OptionalStringListType = Optional[List[str]]
	OptionalStringListType = Optional[list[str]]

[methane]

Suggested change

	bound=Union[Awaitable[IntegerType], IntegerType],
	bound=Union[Awaitable[int], int],

[methane]

get() returns bytes, not str.

>>> import redis
>>> client = redis.Redis()
>>> client.set("foo", b"abcdef")
True
>>> client.get("foo")
b'abcdef'
>>> client.set("bar", "abcdef")
True
>>> client.get("bar")
b'abcdef'

[mahenzon]

Yeah, that's a major issue with decode_responses=True or decode_responses=False

If True, then str is returned. Otherwise bytes.

I'm going to try to fix this with overrides and Literals, not sure yet how to implement this. Other way is just to implement types, like in some other libs. Still working on a solution here

[trajano]

It may need to be asserted by the user e.g.

if not isinstance(res, str):
   msg = f"Expected str type got {type(res)}"
   raise TypeError(msg)

But if decodeResponses is False then return bytes
But if decodeResponses is True then return any
that should be doable by @overload right?

[Harry-Lees]

Other way is just to implement types, like in some other libs. Still working on a solution here

I assume you mean having .pyi files for typing in some of the files? My opinion carries no weight in this project, but I think I'd be a +1 on this approach. Having stub files for these commands seems like a nice balance between several options.

It seems quite difficult to accurately represent the type annotations in a clean way due to the way the command classes are structured. I think the approach taken in this PR is interesting, and works for this scenario, but "feels" quite hacky (I realise this is not necessarily the most constructive criticism), I had to double-take a couple of times to figure out what each generic represents.

Alternative Approaches

Higher-Kinded Generics (Would be nice)

If Python supported higher-kinded TypeVars, it could be semi-cleanly implemented with something like

from collections.abc import Awaitable
from typing import Annotated, Generic, TypeVar

U = TypeVar("U")
A = Annotated[U, ...]
T = TypeVar("T", A, Awaitable)

class Base(Generic[T]):
    def ping(self) -> T[bool]:
        return True

class AsyncClient(Base[Awaitable]):
    pass

class Client(Base[A]):
    pass

Client().ping()  # Should be `bool`
AsyncClient().ping()  # Should be `Awaitable[bool]`

Where A here represents an opaque type designed to be "invisible" to the end-user.

Type Overriding

It is also possible to override the types of the superclass functions in a subclass like

class Superclass:
    def get(self, name: KeyT) -> Any | Awaitable[Any]: ...

class Subclass(Superclass):
    if TYPE_CHECKING:
        def get(self, name: KeyT) -> Awaitable[str | byets]: ...

But this will become very verbose for this project having to re-type all of the functions inside the source code.

Stubs

Which brings me back to the idea of stubs, the stub files provided by typeshed appeared to accurately represent most of the cases for redis-py including overloads for decode_responses which shows how it is possible to represent this behaviour in the stub files.

IMO, this is a nice balance between having a "straightforward" implementation while also not polluting the code with hundreds of lines of typing information.

[trajano]

I think it's best to just leave it as any

You cannot have

redis.Redis(decode_responses: Literal[True]) redis.Redis(decode_responses: Literal[False]) having two different signatures on construction.

I am thinking the use of str | bytes will cause more problems for the users of the library. Ideally there would be a

make_redis(decode_responses: Literal[True]) -> ResponseDecodingRedis
make_redis(decode_responses: Literal[False]-> Redis

[mahenzon]

Well, yes, we can't benefit from overloading __init__ directly since it doesn't return anything. But there're ways around. I see it this way:

protocols

from typing import Protocol


class RedisInterface(Protocol):

    def get(self, key: str) -> str | bytes | None: ...


class RedisDecoded(Protocol):
    def get(self, key: str) -> str | None: ...


class RedisEncoded(Protocol):
    def get(self, key: str) -> bytes | None: ...

implementation

from redis.protocols import RedisInterface

# fake implementation just for example
class RedisImplementation(RedisInterface):

    def __init__(self, decoded: bool = False):
        self.decoded = decoded

    def get(self, key: str) -> str | bytes | None:
        if key == "get-none":
            return None
        if self.decoded:
            return "foo"
        return b"bar"

overloads (for example, stored in redis.typed)

@overload
def Redis() -> RedisEncoded: ...


@overload
def Redis(decoded: Literal[False] = ...) -> RedisEncoded: ...


@overload
def Redis(decoded: Literal[True] = ...) -> RedisDecoded: ...


def Redis(decoded: bool = False) -> RedisEncoded | RedisDecoded | RedisImplementation:
    return RedisImplementation(decoded=decoded)

usage

from redis.typed import Redis

redis = Redis()
redis_decoded = Redis(decoded=True)
redis_encoded = Redis(decoded=False)

reveal_type(redis.get("get-none"))
# MyPy: Revealed type is "Union[builtins.bytes, None]"
# Runtime type is 'NoneType'
reveal_type(redis.get("f"))
# MyPy: Revealed type is "Union[builtins.bytes, None]"
# Runtime type is 'bytes'

reveal_type(redis_encoded.get("get-none"))
# MyPy: Revealed type is "Union[builtins.bytes, None]"
# Runtime type is 'NoneType'
reveal_type(redis_encoded.get("f"))
# MyPy: Revealed type is "Union[builtins.bytes, None]"
# Runtime type is 'bytes'

reveal_type(redis_decoded.get("get-none"))
# MyPy: Revealed type is "Union[builtins.str, None]"
# Runtime type is 'NoneType'
reveal_type(redis_decoded.get("f"))
# MyPy: Revealed type is "Union[builtins.str, None]"
# Runtime type is 'str'

[trajano]

So basically have a separate .typed package and leave the originals untyped. I guess considering the current state of the code that would be a good way of deferring the debt.