Please add type hints

[ughstudios]

Add type hints

[leetrout]

They have types floating around (like in the client) but they are throwing them away when things surface through utility functions and such so mypy happily marches on with implict Any at every turn.

This should be fixed IMO but I am not interested in doing the work if a PR for this is not wanted.

[WisdomPill]

A PR is very well welcome. :)

The reason for many Any is because of the nature of redis,
which could use any bytes as a key or value, even images, because it treats everything as bytes.
I asked around in for a standard in the past in django-stubs for the cache and typeshed for redis without success.

Now I saw that redis-py started adding type hints which could be a good starting point to standardize types.

[WisdomPill]

if you are up for a PR here is the typing module of redis-py. Otherwise I can take it.

[some1ataplace]

Did not test this but here is a basic implementation of the type hints based on the redis-py project:

from typing import Any, Callable, Dict, List, Optional, Tuple, Union

# Basic types
String = Union[str, bytes]
RedisError = Any  # Replace with the appropriate exception class
Pipeline = Any  # Replace with the appropriate pipeline class
Lock = Any  # Replace with the appropriate lock class

# Connection pool types
ConnectionPool = Any  # Replace with the appropriate connection pool class
Connection = Any  # Replace with the appropriate connection class

# Client types
Redis = Any  # Replace with the appropriate Redis client class

# Key types
KeyType = Union[String, int, float]

# Value types
ValueType = Union[String, int, float, bool]

# TTL types
TTL = Union[int, float, None]

# Command execution types
RedisCommand = Callable[..., Any]
RedisCommandArgs = Tuple[RedisCommand, Tuple[Any, ...], Dict[str, Any]]

# Cache backend types
CacheKey = String
CacheValue = ValueType
CacheTimeout = TTL
CacheVersion = Optional[int]

# Cache client types
CacheClient = Any  # Replace with the appropriate cache client class

# Serializer types
Serializer = Any  # Replace with the appropriate serializer class

# Compressor types
Compressor = Any  # Replace with the appropriate compressor class

# Cache operations
CacheSetArgs = Tuple[CacheKey, CacheValue, CacheTimeout, CacheVersion]
CacheAddArgs = Tuple[CacheKey, CacheValue, CacheTimeout, CacheVersion]
CacheGetArgs = Tuple[CacheKey, CacheVersion]
CacheDeleteArgs = Tuple[CacheKey, CacheVersion]
CacheIncrArgs = Tuple[CacheKey, int, CacheVersion]
CacheDecrArgs = Tuple[CacheKey, int, CacheVersion]
CacheExpireArgs = Tuple[CacheKey, CacheTimeout, CacheVersion]

# Cache operations return types
CacheSetResult = bool
CacheAddResult = bool
CacheGetResult = Optional[CacheValue]
CacheDeleteResult = bool
CacheIncrResult = Optional[int]
CacheDecrResult = Optional[int]
CacheExpireResult = bool

# Cache backend methods
class CacheBackend:
    def init(
        self,
        client: CacheClient,
        serializer: Optional[Serializer] = None,
        compressor: Optional[Compressor] = None,
    ) -> None:
        self.client = client
        self.serializer = serializer
        self.compressor = compressor

    def _serialize(self, value: CacheValue) -> String:
        if self.serializer:
            return self.serializer.dumps(value)
        return value

    def _deserialize(self, value: String) -> CacheValue:
        if self.serializer:
            return self.serializer.loads(value)
        return value

    def _compress(self, value: String) -> String:
        if self.compressor:
            return self.compressor.compress(value)
        return value

    def _decompress(self, value: String) -> String:
        if self.compressor:
            return self.compressor.decompress(value)
        return value

    def set(self, *args: CacheSetArgs) -> CacheSetResult:
        key, value, timeout, version= args
        serialized_value = self._serialize(value)
        compressed_value = self._compress(serialized_value)
        result = self.client.set(key, compressed_value, timeout, version)
        return result

    def add(self, args: CacheAddArgs) -> CacheAddResult:
        key, value, timeout, version = args
        serialized_value = self._serialize(value)
        compressed_value = self._compress(serialized_value)
        result = self.client.add(key, compressed_value, timeout, version)
        return result

    def get(self,args: CacheGetArgs) -> CacheGetResult:
        key, version = args
        compressed_value = self.client.get(key, version)
        if compressed_value is None:
            return None
        decompressed_value = self._decompress(compressed_value)
        deserialized_value = self._deserialize(decompressed_value)
        return deserialized_value

    def delete(self, args: CacheDeleteArgs) -> CacheDeleteResult:
        key, version = args
        result = self.client.delete(key, version)
        return result

    def incr(self,args: CacheIncrArgs) -> CacheIncrResult:
        key, amount, version = args
        result = self.client.incr(key, amount, version)
        return result

    def decr(self, args: CacheDecrArgs) -> CacheDecrResult:
        key, amount, version = args
        result = self.client.decr(key, amount, version)
        return result

    def expire(self,args: CacheExpireArgs) -> CacheExpireResult:
        key, timeout, version = args
        result = self.client.expire(key, timeout, version)
        return result

    def clear(self) -> CacheClearResult:
        result = self.client.clear()
        return result

    def _serialize(self, value: Any) -> bytes:
        return pickle.dumps(value)

    def _deserialize(self, serialized_value: bytes) -> Any:
        return pickle.loads(serialized_value)

    def _compress(self, serialized_value: bytes) -> bytes:
        return zlib.compress(serialized_value)

    def _decompress(self, compressed_value: bytes) -> bytes:
        return zlib.decompress(compressed_value)

[WisdomPill]

hello @some1ataplace

thanks for reminding me about this, I guess we could pin redis-py version and just import types from there wherever possible? remember that there is also django-stubs which has types based on django cache that might differ and create issues. But imho we should comply with redis because django-redis is more tied with redis than django.

Would you like to contribute and add more typing from redis-py?

[LucianaAG]

is available to work on this issue?

[WisdomPill]

we started adding type hints some time ago in #696 but there is still room for improvement I guess but it will need some refactoring regarding connections