Improved type annotations

[mahenzon]

Pull Request check-list

Reviewed and checked all of these items:

• [X] tests and lints pass with this change - checked in my fork: https://github.com/mahenzon/redis-py/pull/3
• [X] 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: https://github.com/mahenzon/redis-py/pull/3
• [X] 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: https://github.com/redis/redis-py/pull/3619/commits/23ff99465ef9a9279f40aff9a89c46696bd197b5

[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.

[connebs]

Would love to see this merged!

[Graeme22]

This is desperately needed! Anything I can do to help this along?

[mahenzon]

@Graeme22 hi, there's no detailed plan for this issue. you can take a look at what's been done with basic commands and work on the next commands class. You'll need to check what redis returns to create proper annotations