Add Async Support (#1899)

Co-authored-by: Chayim I. Kirshen <c@kirshen.com>
Co-authored-by: dvora-h <dvora.heller@redis.com>

1 files changed, 45 insertions, 0 deletions

diff --git a/redis/typing.py b/redis/typing.py
new file mode 100644
index 0000000..d96e4e3
--- /dev/null
+++ b/redis/typing.py
@@ -0,0 +1,45 @@
+# from __future__ import annotations
+
+from datetime import datetime, timedelta
+from typing import TYPE_CHECKING, Iterable, TypeVar, Union
+
+from redis.compat import Protocol
+
+if TYPE_CHECKING:
+    from redis.asyncio.connection import ConnectionPool as AsyncConnectionPool
+    from redis.connection import ConnectionPool
+
+
+EncodedT = Union[bytes, memoryview]
+DecodedT = Union[str, int, float]
+EncodableT = Union[EncodedT, DecodedT]
+AbsExpiryT = Union[int, datetime]
+ExpiryT = Union[float, timedelta]
+ZScoreBoundT = Union[float, str]  # str allows for the [ or ( prefix
+BitfieldOffsetT = Union[int, str]  # str allows for #x syntax
+_StringLikeT = Union[bytes, str, memoryview]
+KeyT = _StringLikeT  # Main redis key space
+PatternT = _StringLikeT  # Patterns matched against keys, fields etc
+FieldT = EncodableT  # Fields within hash tables, streams and geo commands
+KeysT = Union[KeyT, Iterable[KeyT]]
+ChannelT = _StringLikeT
+GroupT = _StringLikeT  # Consumer group
+ConsumerT = _StringLikeT  # Consumer name
+StreamIdT = Union[int, _StringLikeT]
+ScriptTextT = _StringLikeT
+TimeoutSecT = Union[int, float, _StringLikeT]
+# Mapping is not covariant in the key type, which prevents
+# Mapping[_StringLikeT, X from accepting arguments of type Dict[str, X]. Using
+# a TypeVar instead of a Union allows mappings with any of the permitted types
+# to be passed. Care is needed if there is more than one such mapping in a
+# type signature because they will all be required to be the same key type.
+AnyKeyT = TypeVar("AnyKeyT", bytes, str, memoryview)
+AnyFieldT = TypeVar("AnyFieldT", bytes, str, memoryview)
+AnyChannelT = TypeVar("AnyChannelT", bytes, str, memoryview)
+
+
+class CommandsProtocol(Protocol):
+    connection_pool: Union["AsyncConnectionPool", "ConnectionPool"]
+
+    def execute_command(self, *args, **options):
+        ...