Repository: Andrew-Chen-Wang/django-async-redis (language: python)

Issue:
aioredis v2
I've been assisting (pretty minimally) in the migration from aioredis 1.3.1 to aioredis 2.0. We're planning on a major release out of beta soon (by the looks of it September), so gear up as we begin the migration process here as well, dropping aioredis 1.3.1 support completely.


Discussion / hints:
We just published aioredis 2.0.0 Will try to get a migration going soon.
Holding off for now since:
- Django hasn't merged my pr for async cache yet (I also haven't updated it in awhile and we still can't figure out the naming)
- 2.0.0 is actually a little unstable when it comes to closing connections. Figuring out the solution now.

Fix patch:
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
new file mode 100644
index 0000000..d3207ed
--- /dev/null
+++ b/.github/workflows/ci.yml
@@ -0,0 +1,57 @@
+name: CI
+
+on:
+  push:
+    branches: [ main ]
+  pull_request:
+    branches: [ main ]
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ['3.7', '3.8', '3.9', '3.10', '3.11']
+        django-version: ['3.2', '4.1']
+        exclude:
+          - python-version: '3.10'
+            django-version: '3.2'
+          - python-version: '3.11'
+            django-version: '3.2'
+          - python-version: '3.7'
+            django-version: '4.1'
+    services:
+      redis:
+        image: redis:6
+        ports:
+          - 6379:6379
+
+    steps:
+    - uses: actions/checkout@v2
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v2
+      with:
+        python-version: ${{ matrix.python-version }}
+    - name: Get pip cache dir
+      id: pip-cache
+      run: |
+        echo "::set-output name=dir::$(pip cache dir)"
+    - name: Cache
+      uses: actions/cache@v2
+      with:
+        path: ${{ steps.pip-cache.outputs.dir }}
+        key:
+          ${{ matrix.python-version }}-v1-${{ hashFiles('**/setup.py') }}
+        restore-keys: |
+          ${{ matrix.python-version }}-v1-
+    - name: Install dependencies
+      run: |
+        sudo apt-get install -y zlib1g-dev
+        python -m pip install --upgrade pip wheel
+        python -m pip install tox tox-gh-actions coveralls
+    - name: Tox Test
+      run: tox
+      env:
+        PYTHON_VER: ${{ matrix.python-version }}
+        DJANGO: ${{ matrix.django-version }}
diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml
new file mode 100644
index 0000000..d54dbd7
--- /dev/null
+++ b/.github/workflows/publish.yml
@@ -0,0 +1,31 @@
+# This workflows will upload a Python Package using Twine when a release is created
+# For more information see: https://help.github.com/en/actions/language-and-framework-guides/using-python-with-github-actions#publishing-to-package-registries
+
+name: Upload Python Package
+
+on:
+  release:
+    types: [created]
+
+jobs:
+  deploy:
+
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v2
+    - name: Set up Python
+      uses: actions/setup-python@v2
+      with:
+        python-version: '3.x'
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install setuptools wheel twine
+    - name: Build and publish
+      env:
+        TWINE_USERNAME: __token__
+        TWINE_PASSWORD: ${{ secrets.PYPI_PASSWORD }}
+      run: |
+        python setup.py sdist bdist_wheel
+        twine upload dist/*
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 33f87a7..c69473c 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -10,12 +10,12 @@ repos:
       - id: check-yaml
 
   - repo: https://github.com/psf/black
-    rev: 19.10b0
+    rev: 22.12.0
     hooks:
       - id: black
 
   - repo: https://github.com/timothycrosley/isort
-    rev: 4.3.21
+    rev: 5.11.4
     hooks:
       - id: isort
-        args: ['--settings-path=setup.cfg', '-rc']
+        args: ['--settings-path=setup.cfg']
diff --git a/.travis.yml b/.travis.yml
deleted file mode 100644
index 7b9e569..0000000
--- a/.travis.yml
+++ /dev/null
@@ -1,70 +0,0 @@
-os: linux
-dist: xenial
-
-language: python
-cache: pip
-services:
-  - redis
-
-matrix:
-  include:
-    - env: TOXENV=lint
-    - python: 3.6
-      env: TOXENV=py36-django30-aioredislatest
-    - python: 3.6
-      env: TOXENV=py36-django30-aioredismaster
-    - python: 3.7
-      env: TOXENV=py37-django30-aioredislatest
-    - python: 3.7
-      env: TOXENV=py37-django30-aioredismaster
-    - python: 3.8
-      env: TOXENV=py38-django30-aioredislatest
-    - python: 3.8
-      env: TOXENV=py38-django30-aioredismaster
-    - python: 3.6
-      env: TOXENV=py36-django31-aioredislatest
-    - python: 3.6
-      env: TOXENV=py36-django31-aioredismaster
-    - python: 3.7
-      env: TOXENV=py37-django31-aioredislatest
-    - python: 3.7
-      env: TOXENV=py37-django31-aioredismaster
-    - python: 3.8
-      env: TOXENV=py38-django31-aioredislatest
-    - python: 3.8
-      env: TOXENV=py38-django31-aioredismaster
-    - python: 3.6
-      env: TOXENV=py36-djangomaster-aioredislatest
-    - python: 3.6
-      env: TOXENV=py36-djangomaster-aioredismaster
-    - python: 3.7
-      env: TOXENV=py37-djangomaster-aioredislatest
-    - python: 3.7
-      env: TOXENV=py37-djangomaster-aioredismaster
-    - python: 3.8
-      env: TOXENV=py38-djangomaster-aioredislatest
-    - python: 3.8
-      env: TOXENV=py38-djangomaster-aioredismaster
-  allow_failures:
-    - env: TOXENV=lint
-      if: branch=master
-    - env: TOXENV=py36-djangomaster-aioredislatest
-    - env: TOXENV=py36-djangomaster-aioredismaster
-    - env: TOXENV=py37-djangomaster-aioredislatest
-    - env: TOXENV=py37-djangomaster-aioredismaster
-    - env: TOXENV=py38-djangomaster-aioredislatest
-    - env: TOXENV=py38-djangomaster-aioredismaster
-
-install: pip install tox
-script: tox
-
-deploy:
-  provider: pypi
-  distributions: sdist bdist_wheel
-  user: __token__
-  password:
-    secure: E71Tbg2TNUKZMvCxw+FbEizJL/kwBd8PJ9Xg0Abnm1QBsaeOY8G0mq8HArwHeYlH/2485qRNVeXNfzY8fCQbG29cXhzUm7Bv+gCW3rqeQGKMPmRwEupQq4XB2xCzc8xlbhGk0OctT10jCu3ZZ5lyfKbEVYF06oSDjC3M5uS701+M41XjnHu2vufLbs9Dcc5W0yDRopLsgDxWsDuyaEf2guT0/NrbJ5r68TD8s+r681xJcoh4LqCuZw1vLlbvK0xoI0RxsTIKEyAwIX3CMZwuYKuk6i/bTsT5qPLvDQQqvTNYNMTEKR6yiRs9ZQEWCWpyfCM74J3YXxYYvejWk6Tc0kalzmqVkFCUubnJltdxWl6ZrJ7pRnauXddMPETTGAqvbrY+snP1pHDrXr74i0wiPXPcYuFgKXWzk7vUZ5TT8brvwKCUITP1kp59zvEFEtMvZdyC1me9d3rIzZGlItt/3XY6PeTYKyghlzHc+t0gy4UG1ZKyKvvku5XSQIT62a89ZQtPjb+9B8O1w9KvYLX17VR5O1vDOVAzSncFtZIXP3Qhm2pYM4vPGGSOXmPhutyRwalHGWWLX5Xw4pJMmUd+5xyDnaCNfAxxIb3JHXgGNfV21e7wRXhxRAy+esoxGuwQ2PExNMO5qwO/o3SX7+r6Qq33F7opSzI1aL1pbSBrXwM=
-  on:
-    tags: true
-    repo: Andrew-Chen-Wang/django-async-redis
-    python: 3.8
diff --git a/django_async_redis/__init__.py b/django_async_redis/__init__.py
index f72f1c9..617970b 100644
--- a/django_async_redis/__init__.py
+++ b/django_async_redis/__init__.py
@@ -2,4 +2,24 @@
 
 __author__ = """Andrew Chen Wang"""
 __email__ = "acwangpython@gmail.com"
-__version__ = "0.1.0"
+
+VERSION = (0, 2, 0)
+__version__ = ".".join(map(str, VERSION))
+
+
+def get_redis_connection(alias="default", write=True):
+    """
+    Helper used for obtaining a raw redis client.
+    """
+
+    from django.core.cache import caches
+
+    cache = caches[alias]
+
+    if not hasattr(cache, "client"):
+        raise NotImplementedError("This backend does not support this feature")
+
+    if not hasattr(cache.client, "get_client"):
+        raise NotImplementedError("This backend does not support this feature")
+
+    return cache.client.get_client(write)
diff --git a/django_async_redis/cache.py b/django_async_redis/cache.py
index 684c470..0a2ef6f 100644
--- a/django_async_redis/cache.py
+++ b/django_async_redis/cache.py
@@ -1,9 +1,8 @@
 import functools
 import logging
 from collections import OrderedDict
-from typing import Any, AsyncGenerator, Optional, Union
+from typing import Any, AsyncIterator, Callable, Optional, Union
 
-from asgiref.sync import async_to_sync
 from django.conf import settings
 from django.core.cache.backends.base import BaseCache
 from django.utils.module_loading import import_string
@@ -19,7 +18,9 @@
 CONNECTION_INTERRUPTED = object()
 
 
-def omit_exception(method=None, return_value=None):
+def omit_exception(
+    method: Optional[Callable] = None, return_value: Optional[Any] = None
+):
     """
     Simple decorator that intercepts connection
     errors and ignores these if settings specify this.
@@ -35,7 +36,7 @@ def _decorator(self, *args, **kwargs):
         except ConnectionInterrupted as e:
             if self._ignore_exceptions:
                 if self._log_ignored_exceptions:
-                    self.logger.error(str(e))
+                    self.logger.exception("Exception ignored")
 
                 return return_value
             raise e.__cause__
@@ -70,10 +71,6 @@ def __init__(self, server, params):
             else None
         )
 
-    def close(self, **kwargs):
-        # TODO Remove this once Django's close_caches implement cache.close_async
-        async_to_sync(self.close_async)()
-
     @property
     def client(self) -> Union[DefaultClient]:
         """
@@ -84,81 +81,72 @@ def client(self) -> Union[DefaultClient]:
         return self._client
 
     @omit_exception
-    async def set_async(self, *args, **kwargs):
-        """
-        :param key: the cache key.
-        :param value: the cache's value.
-        :param timeout: the timeout in seconds. Will be sent as pexpire.
-        :param version: cache version.
-        :param client: the client to use.
-        :param nx: sets the nx flag. Not set if xx is True.
-        :param xx: sets the xx flag. Takes precedence over nx if True.
-        """
+    async def aset(self, *args, **kwargs):
         return await self.client.set(*args, **kwargs)
 
     @omit_exception
-    async def incr_version_async(self, *args, **kwargs):
+    async def aincr_version(self, *args, **kwargs):
         return await self.client.incr_version(*args, **kwargs)
 
     @omit_exception
-    async def add_async(self, *args, **kwargs):
+    async def aadd(self, *args, **kwargs):
         return await self.client.add(*args, **kwargs)
 
-    async def get_async(self, key, default=None, version=None, client=None):
-        value = await self._get_async(key, default, version, client)
+    async def aget(self, key, default=None, version=None, client=None):
+        value = await self._aget(key, default, version, client)
         if value is CONNECTION_INTERRUPTED:
             value = default
         return value
 
     @omit_exception(return_value=CONNECTION_INTERRUPTED)
-    async def _get_async(self, key, default, version, client):
+    async def _aget(self, key, default, version, client):
         return await self.client.get(
             key, default=default, version=version, client=client
         )
 
     @omit_exception
-    async def delete_async(self, *args, **kwargs):
+    async def adelete(self, *args, **kwargs):
         return await self.client.delete(*args, **kwargs)
 
     @omit_exception
-    async def delete_pattern_async(self, *args, **kwargs):
+    async def adelete_pattern(self, *args, **kwargs):
         kwargs["itersize"] = kwargs.get("itersize", DJANGO_ASYNC_REDIS_SCAN_ITERSIZE)
         return await self.client.delete_pattern(*args, **kwargs)
 
     @omit_exception
-    async def delete_many_async(self, *args, **kwargs):
+    async def adelete_many(self, *args, **kwargs):
         return await self.client.delete_many(*args, **kwargs)
 
     @omit_exception
-    async def clear_async(self) -> None:
+    async def aclear(self) -> None:
         await self.client.clear()
 
     @omit_exception(return_value={})
-    async def get_many_async(self, *args, **kwargs) -> OrderedDict:
+    async def aget_many(self, *args, **kwargs) -> OrderedDict:
         return await self.client.get_many(*args, **kwargs)
 
     @omit_exception
-    async def set_many_async(self, *args, **kwargs):
+    async def aset_many(self, *args, **kwargs):
         return await self.client.set_many(*args, **kwargs)
 
     @omit_exception
-    async def incr_async(self, *args, **kwargs):
+    async def aincr(self, *args, **kwargs):
         return await self.client.incr(*args, **kwargs)
 
     @omit_exception
-    async def decr_async(self, *args, **kwargs):
+    async def adecr(self, *args, **kwargs):
         return await self.client.decr(*args, **kwargs)
 
     @omit_exception
-    async def has_key_async(self, *args, **kwargs):
+    async def ahas_key(self, *args, **kwargs):
         return await self.client.has_key(*args, **kwargs)
 
     @omit_exception
-    async def keys_async(self, *args, **kwargs):
+    async def akeys(self, *args, **kwargs):
         return await self.client.keys(*args, **kwargs)
 
     @omit_exception
-    async def iter_keys_async(self, *args, **kwargs) -> AsyncGenerator[str, Any]:
+    async def aiter_keys(self, *args, **kwargs) -> AsyncIterator[str]:
         """
         Returns a coroutine that the dev will
         await since an async generator is returned.
@@ -166,21 +154,21 @@ async def iter_keys_async(self, *args, **kwargs) -> AsyncGenerator[str, Any]:
         return self.client.iter_keys(*args, **kwargs)
 
     @omit_exception
-    async def ttl_async(self, *args, **kwargs) -> Optional[int]:
+    async def attl(self, *args, **kwargs) -> Optional[int]:
         return await self.client.ttl(*args, **kwargs)
 
     @omit_exception
-    async def persist_async(self, *args, **kwargs):
+    async def apersist(self, *args, **kwargs):
         return await self.client.persist(*args, **kwargs)
 
     @omit_exception
-    async def expire_async(self, *args, **kwargs):
+    async def aexpire(self, *args, **kwargs):
         return await self.client.expire(*args, **kwargs)
 
     @omit_exception
-    async def close_async(self, **kwargs) -> None:
+    async def aclose(self, **kwargs) -> None:
         await self.client.close(**kwargs)
 
     @omit_exception
-    async def touch_async(self, *args, **kwargs):
+    async def atouch(self, *args, **kwargs):
         return await self.client.touch(*args, **kwargs)
diff --git a/django_async_redis/client/default.py b/django_async_redis/client/default.py
index 260f880..bba2428 100644
--- a/django_async_redis/client/default.py
+++ b/django_async_redis/client/default.py
@@ -2,39 +2,32 @@
 import re
 import socket
 from collections import OrderedDict
-from typing import List, Optional, Tuple, Union
+from datetime import datetime
+from typing import Any, AsyncIterator, Dict, List, Optional, Union
 
-from aioredis import Redis
-from aioredis.abc import AbcPool
-from aioredis.errors import ConnectionClosedError, ReplyError
 from django.conf import settings
-from django.core.cache.backends.base import DEFAULT_TIMEOUT, get_key_func
+from django.core.cache.backends.base import DEFAULT_TIMEOUT, BaseCache, get_key_func
 from django.core.exceptions import ImproperlyConfigured
 from django.utils.module_loading import import_string
+from redis.asyncio import Redis
+from redis.exceptions import ConnectionError, ResponseError, TimeoutError
 
 from .. import pool
-from ..compressors.base import BaseCompressor
 from ..exceptions import CompressorError, ConnectionInterrupted
-from ..serializers.base import BaseSerializer
 from ..util import CacheKey
 
 
-_main_exceptions = (
-    ConnectionClosedError,
-    ConnectionResetError,
-    ReplyError,
-    socket.timeout,
-)
+_main_exceptions = (TimeoutError, ResponseError, ConnectionError, socket.timeout)
 
 special_re = re.compile("([*?[])")
 
 
-def glob_escape(s):
+def glob_escape(s: str) -> str:
     return special_re.sub(r"[\1]", s)
 
 
 class DefaultClient:
-    def __init__(self, server: Union[list, tuple, set], params: dict, backend):
+    def __init__(self, server, params: Dict[str, Any], backend: BaseCache) -> None:
         self._backend = backend
         self._server = server
         self._params = params
@@ -64,31 +57,24 @@ def __init__(self, server: Union[list, tuple, set], params: dict, backend):
         )
         compressor_cls = import_string(compressor_path)
 
-        self._serializer: BaseSerializer = serializer_cls(options=self._options)
-        self._compressor: BaseCompressor = compressor_cls(options=self._options)
+        self._serializer = serializer_cls(options=self._options)
+        self._compressor = compressor_cls(options=self._options)
 
         self.connection_factory = pool.get_connection_factory(options=self._options)
 
-    def __contains__(self, key):
-        return self.has_key(key)
-
-    async def close_connections(self) -> None:
-        """
-        Clears pool connections. Mostly used for testing with pytest
-        https://aioredis.readthedocs.io/en/v1.3.0/api_reference.html#aioredis.ConnectionsPool.clear
+    def get_next_client_index(
+        self, write: bool = True, tried: Optional[List[int]] = None
+    ) -> int:
         """
-        client = await self.get_client(write=False)
-        assert isinstance(
-            client.connection, AbcPool
-        ), "Must subclass aioredis.abc.AbcPool"
-        await client.connection.clear()
+        Return a next index for read client. This function implements a default
+        behavior for get a next read client for a replication setup.
 
-    def get_next_client_index(self, write=True, tried=()) -> int:
-        """
-        Return the next index for read client. This function implements a default
-        behavior for getting the next read client for a replication setup.
-        Overwrite this function if you want a specific behavior.
+        Overwrite this function if you want a specific
+        behavior.
         """
+        if tried is None:
+            tried = list()
+
         if tried and len(tried) < len(self._server):
             not_tried = [i for i in range(0, len(self._server)) if i not in tried]
             return random.choice(not_tried)
@@ -99,15 +85,19 @@ def get_next_client_index(self, write=True, tried=()) -> int:
         return random.randint(1, len(self._server) - 1)
 
     async def get_client(
-        self, write=True, tried=(), show_index=False
-    ) -> Union[Redis, Tuple[Redis, int]]:
+        self,
+        write: bool = True,
+        tried: Optional[List[int]] = None,
+        show_index: bool = False,
+    ):
         """
-        Method used to obtain a raw redis client.
+        Method used for obtain a raw redis client.
+
         This function is used by almost all cache backend
-        operations to obtain a native redis client/connection
+        operations for obtain a native redis client/connection
         instance.
         """
-        index = self.get_next_client_index(write=write, tried=tried or [])
+        index = self.get_next_client_index(write=write, tried=tried)
 
         if self._clients[index] is None:
             self._clients[index] = await self.connect(index)
@@ -117,7 +107,7 @@ async def get_client(
         else:
             return self._clients[index]
 
-    async def connect(self, index=0) -> Redis:
+    async def connect(self, index: int = 0) -> Redis:
         """
         Given a connection index, returns a new raw redis client/connection
         instance. Index is used for replication setups and indicates that
@@ -125,21 +115,27 @@ async def connect(self, index=0) -> Redis:
         """
         return await self.connection_factory.connect(self._server[index])
 
+    async def disconnect(self, index=0, client=None):
+        """delegates the connection factory to disconnect the client"""
+        if not client:
+            client = self._clients[index]
+        return await self.connection_factory.disconnect(client) if client else None
+
     async def set(
         self,
-        key,
-        value,
-        timeout: Optional[int] = DEFAULT_TIMEOUT,
-        version=None,
-        client: Redis = None,
+        key: Any,
+        value: Any,
+        timeout: Optional[float] = DEFAULT_TIMEOUT,
+        version: Optional[int] = None,
+        client: Optional[Redis] = None,
         nx: bool = False,
         xx: bool = False,
-    ):
+    ) -> bool:
         """
         Persist a value to the cache, and set an optional expiration time.
 
         Also supports optional nx parameter. If set to True - will use redis
-        setnx instead of set. Note: xx overrides nx in aioredis.
+        setnx instead of set.
         """
         nkey = self.make_key(key, version=version)
         nvalue = self.encode(value)
@@ -148,54 +144,53 @@ async def set(
             timeout = self._backend.default_timeout
 
         original_client = client
-        tried = []
+        tried: List[int] = []
         while True:
             try:
                 if client is None:
                     client, index = await self.get_client(
                         write=True, tried=tried, show_index=True
                     )
-                with (await client) as c:
-                    if timeout is not None:
-                        # Convert to milliseconds
-                        timeout = int(timeout * 1000)
-
-                        if timeout <= 0:
-                            if nx:
-                                # Using negative timeouts when nx is True should
-                                # not expire (in our case delete) the value if it exists.
-                                # Obviously expire not existent value is noop.
-                                return not await self.has_key(
-                                    key, version=version, client=client
-                                )
-                            else:
-                                # redis doesn't support negative timeouts in ex flags
-                                # so it seems that it's better to just delete the key
-                                # than to set it and than expire in a pipeline
-                                return await self.delete(
-                                    key, client=client, version=version
-                                )
-
-                    # aioredis expects xx/nx flag to be strings. Convert bool to strings
-                    flag = None
-                    if xx:
-                        flag = client.SET_IF_EXIST
-                    elif nx:
-                        flag = client.SET_IF_NOT_EXIST
-                    return bool(await c.set(nkey, nvalue, pexpire=timeout, exist=flag))
+
+                if timeout is not None:
+                    # Convert to milliseconds
+                    timeout = int(timeout * 1000)
+
+                    if timeout <= 0:
+                        if nx:
+                            # Using negative timeouts when nx is True should
+                            # not expire (in our case delete) the value if it exists.
+                            # Obviously expire not existent value is noop.
+                            return not await self.has_key(
+                                key, version=version, client=client
+                            )
+                        else:
+                            # redis doesn't support negative timeouts in ex flags
+                            # so it seems that it's better to just delete the key
+                            # than to set it and than expire in a pipeline
+                            return bool(
+                                await self.delete(key, client=client, version=version)
+                            )
+
+                return bool(await client.set(nkey, nvalue, nx=nx, px=timeout, xx=xx))
             except _main_exceptions as e:
                 if (
                     not original_client
                     and not self._replica_read_only
                     and len(tried) < len(self._server)
                 ):
-                    # noinspection PyUnboundLocalVariable
                     tried.append(index)
                     client = None
                     continue
                 raise ConnectionInterrupted(connection=client) from e
 
-    async def incr_version(self, key, delta=1, version=None, client: Redis = None):
+    async def incr_version(
+        self,
+        key: Any,
+        delta: int = 1,
+        version: Optional[int] = None,
+        client: Optional[Redis] = None,
+    ) -> int:
         """
         Adds delta to the cache version for the supplied key. Returns the
         new version.
@@ -210,11 +205,10 @@ async def incr_version(self, key, delta=1, version=None, client: Redis = None):
         old_key = self.make_key(key, version)
         value = await self.get(old_key, version=version, client=client)
 
-        with await client as c:
-            try:
-                ttl = await c.ttl(old_key)
-            except _main_exceptions as e:
-                raise ConnectionInterrupted(connection=client) from e
+        try:
+            ttl = await self.ttl(old_key, version=version, client=client)
+        except _main_exceptions as e:
+            raise ConnectionInterrupted(connection=client) from e
 
         if value is None:
             raise ValueError("Key '%s' not found" % key)
@@ -229,19 +223,32 @@ async def incr_version(self, key, delta=1, version=None, client: Redis = None):
         return version + delta
 
     async def add(
-        self, key, value, timeout=DEFAULT_TIMEOUT, version=None, client: Redis = None
-    ):
+        self,
+        key: Any,
+        value: Any,
+        timeout: Any = DEFAULT_TIMEOUT,
+        version: Optional[Any] = None,
+        client: Optional[Redis] = None,
+    ) -> bool:
         """
         Add a value to the cache, failing if the key already exists.
+
         Returns ``True`` if the object was added, ``False`` if not.
         """
         return await self.set(
             key, value, timeout, version=version, client=client, nx=True
         )
 
-    async def get(self, key, default=None, version=None, client: Redis = None):
+    async def get(
+        self,
+        key: Any,
+        default=None,
+        version: Optional[int] = None,
+        client: Optional[Redis] = None,
+    ) -> Any:
         """
         Retrieve a value from the cache.
+
         Returns decoded value if key is found, the default if not.
         """
         if client is None:
@@ -249,54 +256,136 @@ async def get(self, key, default=None, version=None, client: Redis = None):
 
         key = self.make_key(key, version=version)
 
-        with await client as c:
-            try:
-                value = await c.get(key)
-            except _main_exceptions as e:
-                raise ConnectionInterrupted(connection=client) from e
+        try:
+            value = await client.get(key)
+        except _main_exceptions as e:
+            raise ConnectionInterrupted(connection=client) from e
 
         if value is None:
             return default
+
         return self.decode(value)
 
-    async def persist(self, key, version=None, client: Redis = None):
+    async def persist(
+        self, key: Any, version: Optional[int] = None, client: Optional[Redis] = None
+    ) -> bool:
+        if client is None:
+            client = await self.get_client(write=True)
+
+        key = self.make_key(key, version=version)
+
+        return await client.persist(key)
+
+    async def expire(
+        self,
+        key: Any,
+        timeout,
+        version: Optional[int] = None,
+        client: Optional[Redis] = None,
+    ) -> bool:
+        if client is None:
+            client = await self.get_client(write=True)
+
+        key = self.make_key(key, version=version)
+
+        return await client.expire(key, timeout)
+
+    async def pexpire(self, key, timeout, version=None, client=None) -> bool:
+        if client is None:
+            client = await self.get_client(write=True)
+
+        key = self.make_key(key, version=version)
+
+        # Temporary casting until https://github.com/redis/redis-py/issues/1664
+        # is fixed.
+        return bool(await client.pexpire(key, timeout))
+
+    async def pexpire_at(
+        self,
+        key: Any,
+        when: Union[datetime, int],
+        version: Optional[int] = None,
+        client: Optional[Redis] = None,
+    ) -> bool:
+        """
+        Set an expire flag on a ``key`` to ``when``, which can be represented
+        as an integer indicating unix time or a Python datetime object.
+        """
         if client is None:
             client = await self.get_client(write=True)
 
         key = self.make_key(key, version=version)
 
-        with await client as c:
-            if await c.exists(key):
-                await c.persist(key)
+        return bool(await client.pexpireat(key, when))
 
-    async def expire(self, key, timeout, version=None, client: Redis = None):
+    async def expire_at(
+        self,
+        key: Any,
+        when: Union[datetime, int],
+        version: Optional[int] = None,
+        client: Optional[Redis] = None,
+    ) -> bool:
+        """
+        Set an expire flag on a ``key`` to ``when``, which can be represented
+        as an integer indicating unix time or a Python datetime object.
+        """
         if client is None:
             client = await self.get_client(write=True)
 
         key = self.make_key(key, version=version)
 
-        with await client as c:
-            if await c.exists(key):
-                await c.expire(key, timeout)
+        return await client.expireat(key, when)
 
-    async def delete(self, key, version=None, prefix=None, client: Redis = None):
+    async def lock(
+        self,
+        key,
+        version: Optional[int] = None,
+        timeout=None,
+        sleep=0.1,
+        blocking_timeout=None,
+        client: Optional[Redis] = None,
+        thread_local=True,
+    ):
+        if client is None:
+            client = await self.get_client(write=True)
+
+        key = self.make_key(key, version=version)
+        return client.lock(
+            key,
+            timeout=timeout,
+            sleep=sleep,
+            blocking_timeout=blocking_timeout,
+            thread_local=thread_local,
+        )
+
+    async def delete(
+        self,
+        key: Any,
+        version: Optional[int] = None,
+        prefix: Optional[str] = None,
+        client: Optional[Redis] = None,
+    ) -> int:
         """
         Remove a key from the cache.
         """
         if client is None:
             client = await self.get_client(write=True)
 
-        with await client as c:
-            try:
-                return await c.delete(
-                    self.make_key(key, version=version, prefix=prefix)
-                )
-            except _main_exceptions as e:
-                raise ConnectionInterrupted(connection=client) from e
+        try:
+            return await client.delete(
+                self.make_key(key, version=version, prefix=prefix)
+            )
+        except _main_exceptions as e:
+            raise ConnectionInterrupted(connection=client) from e
 
     async def delete_pattern(
-        self, pattern, version=None, prefix=None, client: Redis = None, itersize=None
-    ):
+        self,
+        pattern: str,
+        version: Optional[int] = None,
+        prefix: Optional[str] = None,
+        client: Optional[Redis] = None,
+        itersize: Optional[int] = None,
+    ) -> int:
         """
         Remove all keys matching pattern.
         """
@@ -306,38 +395,40 @@ async def delete_pattern(
 
         pattern = self.make_pattern(pattern, version=version, prefix=prefix)
 
-        kwargs = {"match": pattern}
-        if itersize:
-            kwargs["count"] = itersize
+        try:
+            count = 0
+            pipeline = client.pipeline()
 
-        with await client as c:
-            try:
-                count = 0
-                async for key in c.iscan(**kwargs):
-                    await c.delete(key)
-                    count += 1
-                return count
-            except _main_exceptions as e:
-                raise ConnectionInterrupted(connection=client) from e
+            async for key in client.scan_iter(match=pattern, count=itersize):
+                pipeline.delete(key)
+                count += 1
+            await pipeline.execute()
 
-    async def delete_many(self, keys, version=None, client: Redis = None):
+            return count
+        except _main_exceptions as e:
+            raise ConnectionInterrupted(connection=client) from e
+
+    async def delete_many(
+        self, keys, version: Optional[int] = None, client: Optional[Redis] = None
+    ):
         """
         Remove multiple keys at once.
         """
-        keys = [self.make_key(k, version=version) for k in keys]
-        if not keys:
-            return
 
         if client is None:
             client = await self.get_client(write=True)
 
-        with await client as c:
-            try:
-                return await c.delete(*keys)
-            except _main_exceptions as e:
-                raise ConnectionInterrupted(connection=client) from e
+        keys = [self.make_key(k, version=version) for k in keys]
 
-    async def clear(self, client: Redis = None):
+        if not keys:
+            return
+
+        try:
+            return await client.delete(*keys)
+        except _main_exceptions as e:
+            raise ConnectionInterrupted(connection=client) from e
+
+    async def clear(self, client: Optional[Redis] = None) -> None:
         """
         Flush all cache keys.
         """
@@ -345,13 +436,12 @@ async def clear(self, client: Redis = None):
         if client is None:
             client = await self.get_client(write=True)
 
-        with await client as c:
-            try:
-                await c.flushdb()
-            except _main_exceptions as e:
-                raise ConnectionInterrupted(connection=client) from e
+        try:
+            await client.flushdb()
+        except _main_exceptions as e:
+            raise ConnectionInterrupted(connection=client) from e
 
-    def decode(self, value):
+    def decode(self, value: Union[bytes, int]) -> Any:
         """
         Decode the given value.
         """
@@ -366,7 +456,7 @@ def decode(self, value):
             value = self._serializer.loads(value)
         return value
 
-    def encode(self, value):
+    def encode(self, value: Any) -> Union[bytes, Any]:
         """
         Encode the given value.
         """
@@ -378,124 +468,127 @@ def encode(self, value):
 
         return value
 
-    async def get_many(self, keys, version=None, client: Redis = None):
+    async def get_many(
+        self, keys, version: Optional[int] = None, client: Optional[Redis] = None
+    ) -> OrderedDict:
         """
         Retrieve many keys.
         """
-        if not keys:
-            return {}
 
         if client is None:
             client = await self.get_client(write=False)
 
+        if not keys:
+            return OrderedDict()
+
         recovered_data = OrderedDict()
+
         map_keys = OrderedDict((self.make_key(k, version=version), k) for k in keys)
 
-        with await client as c:
-            try:
-                results = await c.mget(*map_keys)
-            except _main_exceptions as e:
-                raise ConnectionInterrupted(connection=client) from e
+        try:
+            results = await client.mget(*map_keys)
+        except _main_exceptions as e:
+            raise ConnectionInterrupted(connection=client) from e
 
-            for key, value in zip(map_keys, results):
-                if value is None:
-                    continue
-                recovered_data[map_keys[key]] = self.decode(value)
-            return recovered_data
+        for key, value in zip(map_keys, results):
+            if value is None:
+                continue
+            recovered_data[map_keys[key]] = self.decode(value)
+        return recovered_data
 
     async def set_many(
-        self, data, timeout=DEFAULT_TIMEOUT, version=None, client: Redis = None
-    ):
+        self,
+        data: Dict[Any, Any],
+        timeout: Optional[float] = DEFAULT_TIMEOUT,
+        version: Optional[int] = None,
+        client: Optional[Redis] = None,
+    ) -> None:
         """
         Set a bunch of values in the cache at once from a dict of key/value
         pairs. This is much more efficient than calling set() multiple times.
+
         If timeout is given, that timeout will be used for the key; otherwise
         the default cache timeout will be used.
         """
-        if timeout is DEFAULT_TIMEOUT:
-            timeout = self._backend.default_timeout
-
         if client is None:
             client = await self.get_client(write=True)
 
-        with await client as c:
-            try:
-                # Use pipeline since mset doesn't have timeout
-                pipeline = c.pipeline()
-                for key, value in data.items():
-                    nkey = self.make_key(key, version=version)
-                    nvalue = self.encode(value)
-                    if timeout is not None:
-                        timeout = int(timeout * 1000)
-                        if timeout <= 0:
-                            await self.delete(key, client=client, version=version)
-                            continue
-                    pipeline.set(nkey, nvalue, pexpire=timeout)
-
-                await pipeline.execute()
-            except _main_exceptions as e:
-                raise ConnectionInterrupted(connection=client) from e
+        try:
+            pipeline = client.pipeline()
+            for key, value in data.items():
+                await self.set(key, value, timeout, version=version, client=pipeline)
+            await pipeline.execute()
+        except _main_exceptions as e:
+            raise ConnectionInterrupted(connection=client) from e
 
     async def _incr(
-        self, key, delta=1, version=None, client: Redis = None, ignore_key_check=False
-    ):
+        self,
+        key: Any,
+        delta: int = 1,
+        version: Optional[int] = None,
+        client: Optional[Redis] = None,
+        ignore_key_check: bool = False,
+    ) -> int:
         if client is None:
             client = await self.get_client(write=True)
 
         key = self.make_key(key, version=version)
 
-        with await client as c:
+        try:
             try:
-                try:
-                    # if key expired after exists check, then we get
-                    # key with wrong value and ttl -1.
-                    # use lua script for atomicity
-                    if not ignore_key_check:
-                        lua = """
-                        local exists = redis.call('EXISTS', KEYS[1])
-                        if (exists == 1) then
-                            return redis.call('INCRBY', KEYS[1], ARGV[1])
-                        else return false end
-                        """
-                    else:
-                        lua = """
+                # if key expired after exists check, then we get
+                # key with wrong value and ttl -1.
+                # use lua script for atomicity
+                if not ignore_key_check:
+                    lua = """
+                    local exists = redis.call('EXISTS', KEYS[1])
+                    if (exists == 1) then
                         return redis.call('INCRBY', KEYS[1], ARGV[1])
-                        """
-                    value = await c.eval(lua, [key], [delta])
-                    if value is None:
-                        raise ValueError("Key '%s' not found" % key)
-                except ReplyError:
-                    # if cached value or total value is greater than 64 bit signed
-                    # integer.
-                    # elif int is encoded. so redis sees the data as string.
-                    # In this situation redis will throw ResponseError
-
-                    # try to keep TTL of key
-
-                    timeout = await c.ttl(key)
-                    # returns -2 if the key does not exist
-                    # means, that key have expired
-                    if timeout == -2:
-                        await client.quit()
-                        raise ValueError("Key '%s' not found" % key)
-                    value = await self.get(key, version=version, client=client) + delta
-                    await self.set(
-                        key, value, version=version, timeout=timeout, client=client
-                    )
-            except _main_exceptions as e:
-                raise ConnectionInterrupted(connection=client) from e
+                    else return false end
+                    """
+                else:
+                    lua = """
+                    return redis.call('INCRBY', KEYS[1], ARGV[1])
+                    """
+                value = await client.eval(lua, 1, key, delta)
+                if value is None:
+                    raise ValueError("Key '%s' not found" % key)
+            except ResponseError:
+                # if cached value or total value is greater than 64 bit signed
+                # integer.
+                # elif int is encoded. so redis sees the data as string.
+                # In this situations redis will throw ResponseError
+
+                # try to keep TTL of key
+                timeout = await self.ttl(key, version=version, client=client)
+
+                # returns -2 if the key does not exist
+                # means, that key have expired
+                if timeout == -2:
+                    raise ValueError("Key '%s' not found" % key)
+                value = await self.get(key, version=version, client=client) + delta
+                await self.set(
+                    key, value, version=version, timeout=timeout, client=client
+                )
+        except _main_exceptions as e:
+            raise ConnectionInterrupted(connection=client) from e
 
         return value
 
-    def incr(
-        self, key, delta=1, version=None, client: Redis = None, ignore_key_check=False
-    ):
+    async def incr(
+        self,
+        key: Any,
+        delta: int = 1,
+        version: Optional[int] = None,
+        client: Optional[Redis] = None,
+        ignore_key_check: bool = False,
+    ) -> int:
         """
         Add delta to value in the cache. If the key does not exist, raise a
         ValueError exception. if ignore_key_check=True then the key will be
         created and set to the delta value by default.
         """
-        return self._incr(
+        return await self._incr(
             key=key,
             delta=delta,
             version=version,
@@ -503,14 +596,22 @@ def incr(
             ignore_key_check=ignore_key_check,
         )
 
-    def decr(self, key, delta=1, version=None, client: Redis = None):
+    async def decr(
+        self,
+        key: Any,
+        delta: int = 1,
+        version: Optional[int] = None,
+        client: Optional[Redis] = None,
+    ) -> int:
         """
-        Decrease delta to value in the cache. If the key does not exist, raise a
+        Decreace delta to value in the cache. If the key does not exist, raise a
         ValueError exception.
         """
-        return self._incr(key=key, delta=-delta, version=version, client=client)
+        return await self._incr(key=key, delta=-delta, version=version, client=client)
 
-    async def ttl(self, key, version=None, client: Redis = None):
+    async def ttl(
+        self, key: Any, version: Optional[int] = None, client: Optional[Redis] = None
+    ) -> Optional[int]:
         """
         Executes TTL redis command and return the "time-to-live" of specified key.
         If key is a non volatile key, it returns None.
@@ -519,10 +620,10 @@ async def ttl(self, key, version=None, client: Redis = None):
             client = await self.get_client(write=False)
 
         key = self.make_key(key, version=version)
-        with await client as c:
-            if not await c.exists(key):
-                return 0
-            t = await c.ttl(key)
+        if not await client.exists(key):
+            return 0
+
+        t = await client.ttl(key)
 
         if t >= 0:
             return t
@@ -534,23 +635,53 @@ async def ttl(self, key, version=None, client: Redis = None):
             # Should never reach here
             return None
 
-    async def has_key(self, key, version=None, client: Redis = None):
+    async def pttl(self, key, version=None, client=None):
+        """
+        Executes PTTL redis command and return the "time-to-live" of specified key.
+        If key is a non volatile key, it returns None.
+        """
+        if client is None:
+            client = await self.get_client(write=False)
+
+        key = self.make_key(key, version=version)
+        if not await client.exists(key):
+            return 0
+
+        t = await client.pttl(key)
+
+        if t >= 0:
+            return t
+        elif t == -1:
+            return None
+        elif t == -2:
+            return 0
+        else:
+            # Should never reach here
+            return None
+
+    async def has_key(
+        self, key: Any, version: Optional[int] = None, client: Optional[Redis] = None
+    ) -> bool:
         """
         Test if key exists.
         """
 
         if client is None:
             client = await self.get_client(write=False)
-        with await client as c:
-            key = self.make_key(key, version=version)
-            try:
-                return await c.exists(key) == 1
-            except _main_exceptions as e:
-                raise ConnectionInterrupted(connection=client) from e
+
+        key = self.make_key(key, version=version)
+        try:
+            return await client.exists(key) == 1
+        except _main_exceptions as e:
+            raise ConnectionInterrupted(connection=client) from e
 
     async def iter_keys(
-        self, search, itersize=None, client: Redis = None, version=None
-    ):
+        self,
+        search: str,
+        itersize: Optional[int] = None,
+        client: Optional[Redis] = None,
+        version: Optional[int] = None,
+    ) -> AsyncIterator[str]:
         """
         Same as keys, but uses redis >= 2.8 cursors
         for make memory efficient keys iteration.
@@ -560,12 +691,12 @@ async def iter_keys(
             client = await self.get_client(write=False)
 
         pattern = self.make_pattern(search, version=version)
+        async for item in client.scan_iter(match=pattern, count=itersize):
+            yield self.reverse_key(item.decode())
 
-        with await client as c:
-            async for item in c.iscan(match=pattern, count=itersize):
-                yield self.reverse_key(item.decode())
-
-    async def keys(self, search, version=None, client: Redis = None):
+    async def keys(
+        self, search: str, version: Optional[int] = None, client: Optional[Redis] = None
+    ) -> List[Any]:
         """
         Execute KEYS command and return matched results.
         Warning: this can return huge number of results, in
@@ -576,14 +707,15 @@ async def keys(self, search, version=None, client: Redis = None):
         if client is None:
             client = await self.get_client(write=False)
 
-        with await client as c:
-            pattern = self.make_pattern(search, version=version)
-            try:
-                return [self.reverse_key(k.decode()) for k in await c.keys(pattern)]
-            except _main_exceptions as e:
-                raise ConnectionInterrupted(connection=client) from e
+        pattern = self.make_pattern(search, version=version)
+        try:
+            return [self.reverse_key(k.decode()) for k in await client.keys(pattern)]
+        except _main_exceptions as e:
+            raise ConnectionInterrupted(connection=client) from e
 
-    def make_key(self, key, version=None, prefix=None) -> CacheKey:
+    def make_key(
+        self, key: Any, version: Optional[Any] = None, prefix: Optional[str] = None
+    ) -> CacheKey:
         if isinstance(key, CacheKey):
             return key
 
@@ -595,7 +727,9 @@ def make_key(self, key, version=None, prefix=None) -> CacheKey:
 
         return CacheKey(self._backend.key_func(key, prefix, version))
 
-    def make_pattern(self, pattern, version=None, prefix=None) -> CacheKey:
+    def make_pattern(
+        self, pattern: str, version: Optional[int] = None, prefix: Optional[str] = None
+    ) -> CacheKey:
         if isinstance(pattern, CacheKey):
             return pattern
 
@@ -605,20 +739,32 @@ def make_pattern(self, pattern, version=None, prefix=None) -> CacheKey:
 
         if version is None:
             version = self._backend.version
-        version = glob_escape(str(version))
+        version_str = glob_escape(str(version))
 
-        return CacheKey(self._backend.key_func(pattern, prefix, version))
+        return CacheKey(self._backend.key_func(pattern, prefix, version_str))
 
     async def close(self, **kwargs):
-        if getattr(settings, "DJANGO_ASYNC_REDIS_CLOSE_CONNECTION", False):
-            for i, c in enumerate(self._clients):
-                c.close()
-                await c.wait_closed()
-                self._clients[i] = None
+        close_flag = self._options.get(
+            "CLOSE_CONNECTION",
+            getattr(settings, "DJANGO_ASYNC_REDIS_CLOSE_CONNECTION", False),
+        )
+        if close_flag:
+            await self.do_close_clients()
+
+    async def do_close_clients(self):
+        """default implementation: Override in custom client"""
+        num_clients = len(self._clients)
+        for idx in range(num_clients):
+            await self.disconnect(index=idx)
+        self._clients = [None] * num_clients
 
     async def touch(
-        self, key, timeout=DEFAULT_TIMEOUT, version=None, client: Redis = None
-    ):
+        self,
+        key: Any,
+        timeout: Optional[float] = DEFAULT_TIMEOUT,
+        version: Optional[int] = None,
+        client: Optional[Redis] = None,
+    ) -> bool:
         """
         Sets a new expiration for a key.
         """
@@ -629,11 +775,10 @@ async def touch(
         if client is None:
             client = await self.get_client(write=True)
 
-        with await client as c:
-            key = self.make_key(key, version=version)
-            if timeout is None:
-                return bool(await c.persist(key))
-            else:
-                # Convert to milliseconds
-                timeout = int(timeout * 1000)
-                return bool(await c.pexpire(key, timeout))
+        key = self.make_key(key, version=version)
+        if timeout is None:
+            return bool(await client.persist(key))
+        else:
+            # Convert to milliseconds
+            timeout = int(timeout * 1000)
+            return bool(await client.pexpire(key, timeout))
diff --git a/django_async_redis/compressors/base.py b/django_async_redis/compressors/base.py
index 9d2e7fa..d476ec7 100644
--- a/django_async_redis/compressors/base.py
+++ b/django_async_redis/compressors/base.py
@@ -2,8 +2,8 @@ class BaseCompressor:
     def __init__(self, options):
         self._options = options
 
-    def compress(self, value):
+    def compress(self, value: bytes) -> bytes:
         raise NotImplementedError
 
-    def decompress(self, value):
+    def decompress(self, value: bytes) -> bytes:
         raise NotImplementedError
diff --git a/django_async_redis/compressors/identity.py b/django_async_redis/compressors/identity.py
index a4ec4e3..4946886 100644
--- a/django_async_redis/compressors/identity.py
+++ b/django_async_redis/compressors/identity.py
@@ -2,8 +2,8 @@
 
 
 class IdentityCompressor(BaseCompressor):
-    def compress(self, value):
+    def compress(self, value: bytes) -> bytes:
         return value
 
-    def decompress(self, value):
+    def decompress(self, value: bytes) -> bytes:
         return value
diff --git a/django_async_redis/compressors/lz4.py b/django_async_redis/compressors/lz4.py
new file mode 100644
index 0000000..ffcd479
--- /dev/null
+++ b/django_async_redis/compressors/lz4.py
@@ -0,0 +1,20 @@
+from lz4.frame import compress as _compress
+from lz4.frame import decompress as _decompress
+
+from ..exceptions import CompressorError
+from .base import BaseCompressor
+
+
+class Lz4Compressor(BaseCompressor):
+    min_length = 15
+
+    def compress(self, value: bytes) -> bytes:
+        if len(value) > self.min_length:
+            return _compress(value)
+        return value
+
+    def decompress(self, value: bytes) -> bytes:
+        try:
+            return _decompress(value)
+        except Exception as e:
+            raise CompressorError(e)
diff --git a/django_async_redis/compressors/lzma.py b/django_async_redis/compressors/lzma.py
new file mode 100644
index 0000000..b8c17bc
--- /dev/null
+++ b/django_async_redis/compressors/lzma.py
@@ -0,0 +1,20 @@
+import lzma
+
+from ..exceptions import CompressorError
+from .base import BaseCompressor
+
+
+class LzmaCompressor(BaseCompressor):
+    min_length = 100
+    preset = 4
+
+    def compress(self, value: bytes) -> bytes:
+        if len(value) > self.min_length:
+            return lzma.compress(value, preset=self.preset)
+        return value
+
+    def decompress(self, value: bytes) -> bytes:
+        try:
+            return lzma.decompress(value)
+        except lzma.LZMAError as e:
+            raise CompressorError(e)
diff --git a/django_async_redis/compressors/zlib.py b/django_async_redis/compressors/zlib.py
new file mode 100644
index 0000000..014fc77
--- /dev/null
+++ b/django_async_redis/compressors/zlib.py
@@ -0,0 +1,20 @@
+import zlib
+
+from ..exceptions import CompressorError
+from .base import BaseCompressor
+
+
+class ZlibCompressor(BaseCompressor):
+    min_length = 15
+    preset = 6
+
+    def compress(self, value: bytes) -> bytes:
+        if len(value) > self.min_length:
+            return zlib.compress(value, self.preset)
+        return value
+
+    def decompress(self, value: bytes) -> bytes:
+        try:
+            return zlib.decompress(value)
+        except zlib.error as e:
+            raise CompressorError(e)
diff --git a/django_async_redis/compressors/zstd.py b/django_async_redis/compressors/zstd.py
new file mode 100644
index 0000000..ddf49d0
--- /dev/null
+++ b/django_async_redis/compressors/zstd.py
@@ -0,0 +1,19 @@
+import pyzstd
+
+from ..exceptions import CompressorError
+from .base import BaseCompressor
+
+
+class ZStdCompressor(BaseCompressor):
+    min_length = 15
+
+    def compress(self, value: bytes) -> bytes:
+        if len(value) > self.min_length:
+            return pyzstd.compress(value)
+        return value
+
+    def decompress(self, value: bytes) -> bytes:
+        try:
+            return pyzstd.decompress(value)
+        except pyzstd.ZstdError as e:
+            raise CompressorError(e)
diff --git a/django_async_redis/exceptions.py b/django_async_redis/exceptions.py
index 2aa7390..be96145 100644
--- a/django_async_redis/exceptions.py
+++ b/django_async_redis/exceptions.py
@@ -2,10 +2,10 @@ class ConnectionInterrupted(Exception):
     def __init__(self, connection, parent=None):
         self.connection = connection
 
-    def __str__(self):
+    def __str__(self) -> str:
         error_type = type(self.__cause__).__name__
         error_msg = str(self.__cause__)
-        return "Redis {}: {}".format(error_type, error_msg)
+        return f"Redis {error_type}: {error_msg}"
 
 
 class CompressorError(Exception):
diff --git a/django_async_redis/hash_ring.py b/django_async_redis/hash_ring.py
new file mode 100644
index 0000000..be456f7
--- /dev/null
+++ b/django_async_redis/hash_ring.py
@@ -0,0 +1,58 @@
+import bisect
+import hashlib
+from typing import Dict, Iterable, Iterator, List, Optional, Tuple
+
+
+class HashRing:
+    nodes: List[str] = []
+
+    def __init__(self, nodes: Iterable[str] = (), replicas: int = 128) -> None:
+        self.replicas: int = replicas
+        self.ring: Dict[str, str] = {}
+        self.sorted_keys: List[str] = []
+
+        for node in nodes:
+            self.add_node(node)
+
+    def add_node(self, node: str) -> None:
+        self.nodes.append(node)
+
+        for x in range(self.replicas):
+            _key = f"{node}:{x}"
+            _hash = hashlib.sha256(_key.encode()).hexdigest()
+
+            self.ring[_hash] = node
+            self.sorted_keys.append(_hash)
+
+        self.sorted_keys.sort()
+
+    def remove_node(self, node: str) -> None:
+        self.nodes.remove(node)
+        for x in range(self.replicas):
+            _hash = hashlib.sha256(f"{node}:{x}".encode()).hexdigest()
+            del self.ring[_hash]
+            self.sorted_keys.remove(_hash)
+
+    def get_node(self, key: str) -> Optional[str]:
+        n, i = self.get_node_pos(key)
+        return n
+
+    def get_node_pos(self, key: str) -> Tuple[Optional[str], Optional[int]]:
+        if len(self.ring) == 0:
+            return None, None
+
+        _hash = hashlib.sha256(key.encode()).hexdigest()
+        idx = bisect.bisect(self.sorted_keys, _hash)
+        idx = min(idx - 1, (self.replicas * len(self.nodes)) - 1)
+        return self.ring[self.sorted_keys[idx]], idx
+
+    def iter_nodes(self, key: str) -> Iterator[Tuple[Optional[str], Optional[str]]]:
+        if len(self.ring) == 0:
+            yield None, None
+
+        node, pos = self.get_node_pos(key)
+        for k in self.sorted_keys[pos:]:
+            yield k, self.ring[k]
+
+    def __call__(self, key: str) -> Optional[str]:
+        return self.get_node(key)
diff --git a/django_async_redis/pool.py b/django_async_redis/pool.py
index b75a6a6..63f714c 100644
--- a/django_async_redis/pool.py
+++ b/django_async_redis/pool.py
@@ -1,10 +1,16 @@
-from typing import Any, Dict
+from typing import TYPE_CHECKING, Any, Dict
+from urllib.parse import parse_qs, urlparse
 
-from aioredis import Redis
-from aioredis.parser import Reader
-from aioredis.pool import ConnectionsPool
 from django.conf import settings
+from django.core.exceptions import ImproperlyConfigured
 from django.utils.module_loading import import_string
+from redis.asyncio import Redis
+from redis.asyncio.connection import DefaultParser, to_bool
+from redis.asyncio.sentinel import Sentinel
+
+
+if TYPE_CHECKING:
+    from redis.asyncio import ConnectionPool
 
 
 class ConnectionFactory:
@@ -15,115 +21,161 @@ class ConnectionFactory:
     # ConnectionFactory is instantiated, as Django creates new cache client
     # (DefaultClient) instance for every request.
 
-    _pools = {}
+    _pools: Dict[str, "ConnectionPool"] = {}
 
     def __init__(self, options):
-        # Pool
         pool_cls_path = options.get(
-            "CONNECTION_POOL_CLS", "aioredis.pool.ConnectionsPool"
-        )
-        # Must subclass AbcPool
-        self.pool_cls: ConnectionsPool = import_string(pool_cls_path)
-
-        pool_func_path = options.get("CONNECTION_POOL_FUNC", "aioredis.create_pool")
-        self.pool_func = import_string(pool_func_path)
-        self.pool_func_kwargs: Dict[str, Any] = options.get(
-            "CONNECTION_POOL_KWARGS", {}
+            "CONNECTION_POOL_CLASS", "redis.asyncio.connection.ConnectionPool"
         )
+        self.pool_cls = import_string(pool_cls_path)
+        self.pool_cls_kwargs = options.get("CONNECTION_POOL_KWARGS", {})
 
-        # Client
-        redis_client_func_path = options.get(
-            "REDIS_CLIENT_FUNC", "aioredis.create_redis"
+        redis_client_cls_path = options.get(
+            "REDIS_CLIENT_CLASS", "redis.asyncio.client.Redis"
         )
-        self.redis_client_func = import_string(redis_client_func_path)
-        self.redis_client_func_kwargs = options.get("REDIS_CLIENT_KWARGS", {})
-        # Must subclass AbcConnection
-        redis_client_cls_path = options.get("REDIS_CLIENT_CLASS", "aioredis.Redis")
         self.redis_client_cls = import_string(redis_client_cls_path)
+        self.redis_client_cls_kwargs = options.get("REDIS_CLIENT_KWARGS", {})
 
         self.options = options
 
-    def make_connection_params(self, address) -> Dict[str, Any]:
+    def make_connection_params(self, url: str) -> Dict[str, Any]:
         """
         Given a main connection parameters, build a complete
         dict of connection parameters.
         """
 
         kwargs = {
-            "address": address,
-            "parser": self.get_parser_cls(),
+            "url": url,
+            "parser_class": self.get_parser_cls(),
         }
 
         password = self.options.get("PASSWORD", None)
         if password:
             kwargs["password"] = password
 
-        create_connection_timeout = self.options.get("CREATE_CONNECTION_TIMEOUT", None)
-        if create_connection_timeout:
+        socket_timeout = self.options.get("SOCKET_TIMEOUT", None)
+        if socket_timeout:
             assert isinstance(
-                create_connection_timeout, (int, float)
-            ), "Create connection timeout should be float or integer"
-            assert (
-                create_connection_timeout > 0
-            ), "Create connection timeout should be greater than 0"
-            kwargs["create_connection_timeout"] = create_connection_timeout
+                socket_timeout, (int, float)
+            ), "Socket timeout should be float or integer"
+            kwargs["socket_timeout"] = socket_timeout
+
+        socket_connect_timeout = self.options.get("SOCKET_CONNECT_TIMEOUT", None)
+        if socket_connect_timeout:
+            assert isinstance(
+                socket_connect_timeout, (int, float)
+            ), "Socket connect timeout should be float or integer"
+            kwargs["socket_connect_timeout"] = socket_connect_timeout
 
         return kwargs
 
-    async def connect(self, address: str) -> Redis:
+    async def connect(self, url: str) -> Redis:
         """
         Given a basic connection parameters,
         return a new connection.
         """
-        params: Dict[str, Any] = self.make_connection_params(address)
+        params: Dict[str, Any] = self.make_connection_params(url)
         return await self.get_connection(params)
 
+    async def disconnect(self, connection: Redis):
+        """
+        Given a not null client connection it disconnect from the Redis server.
+        The default implementation uses a pool to hold connections.
+        """
+        await connection.connection_pool.disconnect()
+
     async def get_connection(self, params: Dict[str, Any]) -> Redis:
         """
-        Given pre-formatted params, return a
+        Given a now preformatted params, return a
         new connection.
-
         The default implementation uses a cached pools
         for create new connection.
         """
-        # Creates new connection if needed. If pool is closed, womp womp
-        pool: ConnectionsPool = await self.get_or_create_connection_pool(params)
-        # If no free connection, waits for another conn.
-        # Raises RunTime error if we run out of connections by maxsize
-        return self.redis_client_cls(pool)
+        pool = self.get_or_create_connection_pool(params)
+        return await self.redis_client_cls(
+            connection_pool=pool, **self.redis_client_cls_kwargs
+        )
 
     def get_parser_cls(self):
         cls = self.options.get("PARSER_CLASS", None)
         if cls is None:
-            return Reader
+            return DefaultParser
         return import_string(cls)
 
-    async def get_or_create_connection_pool(
-        self, params: Dict[str, Any]
-    ) -> ConnectionsPool:
+    def get_or_create_connection_pool(self, params: Dict[str, Any]):
         """
-        Given connection parameters and return a new
+        Given a connection parameters and return a new
         or cached connection pool for them.
-
         Reimplement this method if you want distinct
         connection pool instance caching behavior.
         """
-        key = params["address"]
+        key = params["url"]
         if key not in self._pools:
-            self._pools[key] = await self.get_connection_pool(params)
+            self._pools[key] = self.get_connection_pool(params)
         return self._pools[key]
 
-    async def get_connection_pool(self, params: Dict[str, Any]) -> ConnectionsPool:
+    def get_connection_pool(self, params: Dict[str, Any]):
         """
-        Given connection parameters, return a new
+        Given a connection parameters, return a new
         connection pool for them.
-
         Overwrite this method if you want a custom
         behavior on creating connection pool.
         """
         cp_params = dict(params)
-        cp_params.update(self.pool_func_kwargs)
-        pool: ConnectionsPool = await self.pool_func(**cp_params)
+        cp_params.update(self.pool_cls_kwargs)
+        pool: "ConnectionPool" = self.pool_cls.from_url(**cp_params)
+
+        if pool.connection_kwargs.get("password", None) is None:
+            pool.connection_kwargs["password"] = params.get("password", None)
+            pool.reset()
+
+        return pool
+
+
+class SentinelConnectionFactory(ConnectionFactory):
+    def __init__(self, options):
+        # allow overriding the default SentinelConnectionPool class
+        options.setdefault(
+            "CONNECTION_POOL_CLASS", "redis.asyncio.sentinel.SentinelConnectionPool"
+        )
+        super().__init__(options)
+
+        sentinels = options.get("SENTINELS")
+        if not sentinels:
+            raise ImproperlyConfigured(
+                "SENTINELS must be provided as a list of (host, port)."
+            )
+
+        # provide the connection pool kwargs to the sentinel in case it
+        # needs to use the socket options for the sentinels themselves
+        connection_kwargs = self.make_connection_params(None)
+        connection_kwargs.pop("url")
+        connection_kwargs.update(self.pool_cls_kwargs)
+        self._sentinel = Sentinel(
+            sentinels,
+            sentinel_kwargs=options.get("SENTINEL_KWARGS"),
+            **connection_kwargs,
+        )
+
+    def get_connection_pool(self, params: Dict[str, Any]):
+        """
+        Given a connection parameters, return a new sentinel connection pool
+        for them.
+        """
+        url = urlparse(params["url"])
+
+        # explicitly set service_name and sentinel_manager for the
+        # SentinelConnectionPool constructor since will be called by from_url
+        cp_params = dict(params)
+        cp_params.update(service_name=url.hostname, sentinel_manager=self._sentinel)
+        pool = super().get_connection_pool(cp_params)
+
+        # convert "is_master" to a boolean if set on the URL, otherwise if not
+        # provided it defaults to True.
+        is_master = parse_qs(url.query).get("is_master")
+        if is_master:
+            pool.is_master = to_bool(is_master[0])
+
         return pool
 
 
diff --git a/django_async_redis/serializers/__init__.py b/django_async_redis/serializers/__init__.py
index 73b89bd..e69de29 100644
--- a/django_async_redis/serializers/__init__.py
+++ b/django_async_redis/serializers/__init__.py
@@ -1,4 +0,0 @@
-from .json import JSONSerializer
-
-
-__all__ = ["JSONSerializer"]
diff --git a/django_async_redis/serializers/base.py b/django_async_redis/serializers/base.py
index ada86ad..a4e3ce2 100644
--- a/django_async_redis/serializers/base.py
+++ b/django_async_redis/serializers/base.py
@@ -1,9 +1,12 @@
+from typing import Any
+
+
 class BaseSerializer:
     def __init__(self, options):
-        ...
+        pass
 
-    def dumps(self, value):
+    def dumps(self, value: Any) -> bytes:
         raise NotImplementedError
 
-    def loads(self, value):
+    def loads(self, value: bytes) -> Any:
         raise NotImplementedError
diff --git a/django_async_redis/serializers/json.py b/django_async_redis/serializers/json.py
index 4c7fede..06e6254 100644
--- a/django_async_redis/serializers/json.py
+++ b/django_async_redis/serializers/json.py
@@ -9,8 +9,8 @@
 class JSONSerializer(BaseSerializer):
     encoder_class = DjangoJSONEncoder
 
-    def dumps(self, value) -> bytes:
+    def dumps(self, value: Any) -> bytes:
         return json.dumps(value, cls=self.encoder_class).encode()
 
-    def loads(self, value) -> Any:
+    def loads(self, value: bytes) -> Any:
         return json.loads(value.decode())
diff --git a/django_async_redis/serializers/msgpack.py b/django_async_redis/serializers/msgpack.py
new file mode 100644
index 0000000..d41591f
--- /dev/null
+++ b/django_async_redis/serializers/msgpack.py
@@ -0,0 +1,13 @@
+from typing import Any
+
+import msgpack
+
+from .base import BaseSerializer
+
+
+class MSGPackSerializer(BaseSerializer):
+    def dumps(self, value: Any) -> bytes:
+        return msgpack.dumps(value)
+
+    def loads(self, value: bytes) -> Any:
+        return msgpack.loads(value, raw=False)
diff --git a/django_async_redis/serializers/pickle.py b/django_async_redis/serializers/pickle.py
index a2ff2f9..c304360 100644
--- a/django_async_redis/serializers/pickle.py
+++ b/django_async_redis/serializers/pickle.py
@@ -7,19 +7,26 @@
 
 
 class PickleSerializer(BaseSerializer):
-    def __init__(self, options):
-        self._pickle_version = -1
+    def __init__(self, options) -> None:
+        self._pickle_version = pickle.DEFAULT_PROTOCOL
         self.setup_pickle_version(options)
 
-    def setup_pickle_version(self, options):
+        super().__init__(options=options)
+
+    def setup_pickle_version(self, options) -> None:
         if "PICKLE_VERSION" in options:
             try:
                 self._pickle_version = int(options["PICKLE_VERSION"])
+                if self._pickle_version > pickle.HIGHEST_PROTOCOL:
+                    raise ImproperlyConfigured(
+                        f"PICKLE_VERSION can't be higher than pickle.HIGHEST_PROTOCOL:"
+                        f" {pickle.HIGHEST_PROTOCOL}"
+                    )
             except (ValueError, TypeError):
                 raise ImproperlyConfigured("PICKLE_VERSION value must be an integer")
 
-    def dumps(self, value) -> bytes:
+    def dumps(self, value: Any) -> bytes:
         return pickle.dumps(value, self._pickle_version)
 
-    def loads(self, value) -> Any:
+    def loads(self, value: bytes) -> Any:
         return pickle.loads(value)
diff --git a/django_async_redis/util.py b/django_async_redis/util.py
index edae90e..4cf54e9 100644
--- a/django_async_redis/util.py
+++ b/django_async_redis/util.py
@@ -1,19 +1,11 @@
-# noinspection PyProtectedMember
-from aioredis.util import _converters
-
-
 class CacheKey(str):
     """
     A stub string class that we can use to check if a key was created already.
     """
 
-    def original_key(self):
+    def original_key(self) -> str:
         return self.rsplit(":", 1)[1]
 
 
-# Monkeypatch so that CacheKey works with aioredis encoding
-_converters[CacheKey] = lambda x: x.encode()
-
-
-def default_reverse_key(key):
+def default_reverse_key(key: str) -> str:
     return key.split(":", 2)[2]
diff --git a/requirements.txt b/requirements.txt
new file mode 100644
index 0000000..11d5cf4
--- /dev/null
+++ b/requirements.txt
@@ -0,0 +1,25 @@
+django>=3.2
+redis>=4.2.0
+hiredis==2.1.0
+
+lz4==4.3.2
+msgpack==1.0.4
+pyzstd==0.15.3
+
+pip==22.3.1
+bump2version==1.0.1
+wheel==0.38.4
+watchdog==2.2.1
+tox==4.2.3
+tox-current-env==0.0.10
+black==22.12.0  # https://github.com/ambv/black
+isort==5.11.4  # https://github.com/timothycrosley/isort/
+coverage==7.0.3
+Sphinx==6.0.0
+twine==4.0.2
+pre-commit==2.21.0
+
+pytest==7.2.0
+pytest-django==4.5.2
+pytest-asyncio==0.20.3
+pytest-sugar==0.9.6
diff --git a/requirements_dev.txt b/requirements_dev.txt
deleted file mode 100644
index 8fc8eae..0000000
--- a/requirements_dev.txt
+++ /dev/null
@@ -1,19 +0,0 @@
-django>=3
-aioredis>=1.3.1
-
-pip==20.2.3
-bump2version==0.5.11
-wheel==0.33.6
-watchdog==0.9.0
-tox==3.14.0
-tox-current-env==0.0.2
-black==19.10b0  # https://github.com/ambv/black
-isort==4.3.21  # https://github.com/timothycrosley/isort/
-coverage==4.5.4
-Sphinx==1.8.5
-twine==1.14.0
-
-pytest==6.1.0
-pytest-django==3.10.0
-pytest-asyncio==0.14.0
-pytest-sugar==0.9.4
diff --git a/setup.cfg b/setup.cfg
index af3e0e7..e2826c5 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -27,3 +27,4 @@ collect_ignore = ['setup.py']
 [isort]
 LINES_AFTER_IMPORTS=2
 src_paths=django_async_redis,tests
+profile=black
diff --git a/setup.py b/setup.py
index 48d2278..bd7e107 100644
--- a/setup.py
+++ b/setup.py
@@ -11,7 +11,7 @@
 with open("HISTORY.rst") as history_file:
     history = history_file.read()
 
-requirements = ["django>=3", "aioredis"]
+requirements = ["django>=3.2", "redis>=4.2.0"]
 
 setup_requirements = []
 
@@ -20,16 +20,20 @@
 setup(
     author="Andrew Chen Wang",
     author_email="acwangpython@gmail.com",
-    python_requires=">=3.6",
+    python_requires=">=3.7",
     classifiers=[
         "Development Status :: 2 - Pre-Alpha",
         "Intended Audience :: Developers",
         "License :: OSI Approved :: Apache Software License",
         "Natural Language :: English",
         "Programming Language :: Python :: 3",
-        "Programming Language :: Python :: 3.6",
         "Programming Language :: Python :: 3.7",
         "Programming Language :: Python :: 3.8",
+        "Programming Language :: Python :: 3.9",
+        "Programming Language :: Python :: 3.10",
+        "Programming Language :: Python :: 3.11",
+        "Framework :: Django :: 3.2",
+        "Framework :: Django :: 4.1",
     ],
     description="Full featured async Redis cache backend for Django.",
     install_requires=requirements,
@@ -43,6 +47,6 @@
     test_suite="tests",
     tests_require=test_requirements,
     url="https://github.com/Andrew-Chen-Wang/django-async-redis",
-    version="0.1.0",
+    version="0.2.0",
     zip_safe=False,
 )
diff --git a/tox.ini b/tox.ini
index 91af9c8..ee2cc0a 100644
--- a/tox.ini
+++ b/tox.ini
@@ -1,31 +1,35 @@
 [tox]
 envlist =
     lint
-    py{36,37,38}-django30-aioredis{latest,master}
-    py{36,37,38}-django31-aioredis{latest,master}
-    py{36,37,38}-djangomaster-aioredis{latest,master}
+    py{37,38,39}-django32-redis{latest,main}
+    py{38,39,310,311}-django41-redis{latest,main}
+    py{38,39,310,311}-djangomain-redis{latest,main}
 
 [pytest]
 DJANGO_SETTINGS_MODULE = tests.settings.base
 
 [testenv]
 deps =
-    django30: Django>=3.0,<3.1
-    django31: Django>=3.1,<3.2
-    djangomaster: https://github.com/django/django/archive/master.tar.gz
-    aioredislatest: aioredis
-    aioredismaster: https://github.com/aio-libs/aioredis/archive/master.tar.gz
-    pytest == 6.1.0
-    pytest-django == 3.10.0
-    pytest-asyncio == 0.14.0
+    django32: Django>=3.2,<4.0
+    django41: Django>=4.1,<4.2
+    djangomain: https://github.com/django/django/archive/main.tar.gz
+    redislatest: redis
+    redismain: https://github.com/redis/redis-py/archive/main.tar.gz
+    pytest == 7.2.0
+    pytest-django == 4.5.2
+    pytest-asyncio == 0.20.3
+    hiredis == 2.1.0
+    lz4 == 4.3.2
+    msgpack == 1.0.4
+    pyzstd == 0.15.3
 commands =
     pytest
 
 [testenv:lint]
 basepython = python3
 deps =
-    black == 19.10b0
-    isort == 4.3.21
+    black == 22.12.0
+    isort == 5.11.4
 commands =
     black --target-version py38 --check --diff django_async_redis tests
-    isort --settings-path=setup.cfg -rc --check-only --diff django_async_redis tests
+    isort --settings-path=setup.cfg --check-only --diff django_async_redis tests


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index 5b84108..bbb436e 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,16 +1,25 @@
+import asyncio
 from sys import version_info
 
 import pytest
-from django.core.cache import cache
+import pytest_asyncio
+from django.core.cache import cache as default_cache
 
 from django_async_redis.cache import RedisCache
 
 
-@pytest.fixture(autouse=True)
-async def cache_setup() -> RedisCache:
-    await cache.clear_async()
-    yield cache
-    await cache.client.close_connections()
+@pytest.fixture(scope="session")
+def event_loop():
+    policy = asyncio.get_event_loop_policy()
+    loop = policy.new_event_loop()
+    yield loop
+    loop.close()
+
+
+@pytest_asyncio.fixture()
+async def cache() -> RedisCache:
+    yield default_cache
+    await default_cache.aclear()
 
 
 if version_info >= (3, 8):
diff --git a/tests/test_backend.py b/tests/test_backend.py
index 392f3e9..fa5cdac 100644
--- a/tests/test_backend.py
+++ b/tests/test_backend.py
@@ -5,18 +5,17 @@
 
 import pytest
 from django.conf import settings
-from django.core.cache import cache, caches
+from django.core.cache import caches
 from django.test import override_settings
 
 from django_async_redis import pool
 from django_async_redis.cache import DJANGO_ASYNC_REDIS_SCAN_ITERSIZE, RedisCache
-from django_async_redis.serializers import JSONSerializer
+from django_async_redis.serializers.json import JSONSerializer
 
 
 pytestmark = pytest.mark.asyncio
 
 # Type hints
-cache: RedisCache
 caches: Dict[str, RedisCache]
 
 
@@ -27,28 +26,28 @@
 async def test_connection_strings(conn_string):
     cf = pool.get_connection_factory(options={})
     res = cf.make_connection_params(conn_string)
-    assert res["address"] == conn_string
+    assert res["url"] == conn_string
 
 
 class TestEscapePrefix:
     OTHER_CACHE = "with_prefix"
 
-    async def test_delete_pattern(self):
-        assert await cache.set_async("a", "1") is True
-        assert await caches[self.OTHER_CACHE].set_async("b", "2") is True
-        await cache.delete_pattern_async("*")
-        assert await cache.has_key_async("a") is False
-        assert await caches[self.OTHER_CACHE].get_async("b") == "2"
-
-    async def test_iter_keys(self):
-        assert await cache.set_async("a", "1") is True
-        await caches[self.OTHER_CACHE].set_async("b", "2")
-        assert [x async for x in await cache.iter_keys_async("*")] == ["a"]
-
-    async def test_keys(self):
-        await cache.set_async("a", "1")
-        await caches[self.OTHER_CACHE].set_async("b", "2")
-        keys = await cache.keys_async("*")
+    async def test_delete_pattern(self, cache: RedisCache):
+        assert await cache.aset("a", "1") is True
+        assert await caches[self.OTHER_CACHE].aset("b", "2") is True
+        await cache.adelete_pattern("*")
+        assert await cache.ahas_key("a") is False
+        assert await caches[self.OTHER_CACHE].aget("b") == "2"
+
+    async def test_iter_keys(self, cache: RedisCache):
+        assert await cache.aset("a", "1") is True
+        await caches[self.OTHER_CACHE].aset("b", "2")
+        assert [x async for x in await cache.aiter_keys("*")] == ["a"]
+
+    async def test_keys(self, cache: RedisCache):
+        await cache.aset("a", "1")
+        await caches[self.OTHER_CACHE].aset("b", "2")
+        keys = await cache.akeys("*")
         assert "a" in keys
         assert "b" not in keys
 
@@ -62,7 +61,7 @@ def reverse_key(key):
 
 
 class TestCustomKeyFunction:
-    async def test_custom_key_function(self):
+    async def test_custom_key_function(self, cache: RedisCache):
         caches_settings = copy.deepcopy(settings.CACHES)
         caches_settings["default"]["KEY_FUNCTION"] = "tests.test_backend.make_key"
         caches_settings["default"][
@@ -72,12 +71,12 @@ async def test_custom_key_function(self):
         cm.enable()
         # Test
         for key in ["foo-aa", "foo-ab", "foo-bb", "foo-bc"]:
-            await cache.set_async(key, "foo")
+            await cache.aset(key, "foo")
 
-        res = await cache.delete_pattern_async("*foo-a*")
+        res = await cache.adelete_pattern("*foo-a*")
         assert bool(res) is True
 
-        keys = await cache.keys_async("foo*")
+        keys = await cache.akeys("foo*")
         assert set(keys) == {"foo-bb", "foo-bc"}
         # ensure our custom function was actually called
         assert {
@@ -87,77 +86,77 @@ async def test_custom_key_function(self):
 
 
 class TestBackend:
-    async def test_setnx(self):
+    async def test_setnx(self, cache: RedisCache):
         # we should ensure there is no test_key_nx in redis
-        await cache.delete_async("test_key_nx")
-        res = await cache.get_async("test_key_nx")
+        await cache.adelete("test_key_nx")
+        res = await cache.aget("test_key_nx")
         assert res is None
 
-        res = await cache.set_async("test_key_nx", 1, nx=True)
+        res = await cache.aset("test_key_nx", 1, nx=True)
         assert res is True
         # test that second set will have
-        res = await cache.set_async("test_key_nx", 2, nx=True)
+        res = await cache.aset("test_key_nx", 2, nx=True)
         assert res is False
-        res = await cache.get_async("test_key_nx")
+        res = await cache.aget("test_key_nx")
         assert res == 1
 
-        await cache.delete_async("test_key_nx")
-        res = await cache.get_async("test_key_nx")
+        await cache.adelete("test_key_nx")
+        res = await cache.aget("test_key_nx")
         assert res is None
 
-    async def test_setnx_timeout(self):
+    async def test_setnx_timeout(self, cache: RedisCache):
         # test that timeout still works for nx=True
-        res = await cache.set_async("test_key_nx", 1, timeout=2, nx=True)
+        res = await cache.aset("test_key_nx", 1, timeout=2, nx=True)
         assert res is True
         await asyncio.sleep(3)
-        res = await cache.get_async("test_key_nx")
+        res = await cache.aget("test_key_nx")
         assert res is None
 
         # test that timeout will not affect key, if it was there
-        await cache.set_async("test_key_nx", 1)
-        res = await cache.set_async("test_key_nx", 2, timeout=2, nx=True)
+        await cache.aset("test_key_nx", 1)
+        res = await cache.aset("test_key_nx", 2, timeout=2, nx=True)
         assert res is False
         await asyncio.sleep(3)
-        res = await cache.get_async("test_key_nx")
+        res = await cache.aget("test_key_nx")
         assert res == 1
 
-        await cache.delete_async("test_key_nx")
-        res = await cache.get_async("test_key_nx")
+        await cache.adelete("test_key_nx")
+        res = await cache.aget("test_key_nx")
         assert res is None
 
-    async def test_unicode_keys(self):
-        await cache.set_async("ключ", "value")
-        res = await cache.get_async("ключ")
+    async def test_unicode_keys(self, cache: RedisCache):
+        await cache.aset("ключ", "value")
+        res = await cache.aget("ключ")
         assert res == "value"
 
-    async def test_save_and_integer(self):
-        await cache.set_async("test_key", 2)
-        res = await cache.get_async("test_key", "Foo")
+    async def test_save_and_integer(self, cache: RedisCache):
+        await cache.aset("test_key", 2)
+        res = await cache.aget("test_key", "Foo")
 
         assert isinstance(res, int)
         assert res == 2
 
-    async def test_save_string(self):
-        await cache.set_async("test_key", "hello" * 1000)
-        res = await cache.get_async("test_key")
+    async def test_save_string(self, cache: RedisCache):
+        await cache.aset("test_key", "hello" * 1000)
+        res = await cache.aget("test_key")
 
         assert isinstance(res, str)
         assert res == "hello" * 1000
 
-        await cache.set_async("test_key", "2")
-        res = await cache.get_async("test_key")
+        await cache.aset("test_key", "2")
+        res = await cache.aget("test_key")
 
         assert isinstance(res, str)
         assert res == "2"
 
-    async def test_save_unicode(self):
-        await cache.set_async("test_key", "heló")
-        res = await cache.get_async("test_key")
+    async def test_save_unicode(self, cache: RedisCache):
+        await cache.aset("test_key", "heló")
+        res = await cache.aget("test_key")
 
         assert isinstance(res, str)
         assert res == "heló"
 
-    async def test_save_dict(self):
+    async def test_save_dict(self, cache: RedisCache):
         if isinstance(cache.client._serializer, (JSONSerializer,)):
             # JSONSerializer and MSGPackSerializer use the isoformat for
             # datetimes.
@@ -167,114 +166,114 @@ async def test_save_dict(self):
 
         test_dict = {"id": 1, "date": now_dt, "name": "Foo"}
 
-        await cache.set_async("test_key", test_dict)
-        res = await cache.get_async("test_key")
+        await cache.aset("test_key", test_dict)
+        res = await cache.aget("test_key")
 
         assert isinstance(res, dict)
         assert res["id"] == 1
         assert res["name"] == "Foo"
         assert res["date"] == now_dt
 
-    async def test_save_float(self):
+    async def test_save_float(self, cache: RedisCache):
         float_val = 1.345620002
 
-        await cache.set_async("test_key", float_val)
-        res = await cache.get_async("test_key")
+        await cache.aset("test_key", float_val)
+        res = await cache.aget("test_key")
 
         assert isinstance(res, float)
         assert res == float_val
 
-    async def test_timeout(self):
-        await cache.set_async("test_key", 222, timeout=3)
+    async def test_timeout(self, cache: RedisCache):
+        await cache.aset("test_key", 222, timeout=3)
         await asyncio.sleep(4)
 
-        res = await cache.get_async("test_key")
+        res = await cache.aget("test_key")
         assert res is None
 
-    async def test_timeout_0(self):
-        await cache.set_async("test_key", 222, timeout=0)
-        res = await cache.get_async("test_key")
+    async def test_timeout_0(self, cache: RedisCache):
+        await cache.aset("test_key", 222, timeout=0)
+        res = await cache.aget("test_key")
         assert res is None
 
-    async def test_timeout_parameter_as_positional_argument(self):
-        await cache.set_async("test_key", 222, -1)
-        res = await cache.get_async("test_key")
+    async def test_timeout_parameter_as_positional_argument(self, cache: RedisCache):
+        await cache.aset("test_key", 222, -1)
+        res = await cache.aget("test_key")
         assert res is None
 
-        await cache.set_async("test_key", 222, 1)
-        res1 = await cache.get_async("test_key")
+        await cache.aset("test_key", 222, 1)
+        res1 = await cache.aget("test_key")
         await asyncio.sleep(2)
-        res2 = await cache.get_async("test_key")
+        res2 = await cache.aget("test_key")
         assert res1 == 222
         assert res2 is None
 
         # nx=True should not overwrite expire of key already in db
-        await cache.set_async("test_key", 222, None)
-        await cache.set_async("test_key", 222, -1, nx=True)
-        res = await cache.get_async("test_key")
+        await cache.aset("test_key", 222, None)
+        await cache.aset("test_key", 222, -1, nx=True)
+        res = await cache.aget("test_key")
         assert res == 222
 
-    async def test_timeout_negative(self):
-        await cache.set_async("test_key", 222, timeout=-1)
-        res = await cache.get_async("test_key")
+    async def test_timeout_negative(self, cache: RedisCache):
+        await cache.aset("test_key", 222, timeout=-1)
+        res = await cache.aget("test_key")
         assert res is None
 
-        await cache.set_async("test_key", 222, timeout=None)
-        await cache.set_async("test_key", 222, timeout=-1)
-        res = await cache.get_async("test_key")
+        await cache.aset("test_key", 222, timeout=None)
+        await cache.aset("test_key", 222, timeout=-1)
+        res = await cache.aget("test_key")
         assert res is None
 
         # nx=True should not overwrite expire of key already in db
-        await cache.set_async("test_key", 222, timeout=None)
-        await cache.set_async("test_key", 222, timeout=-1, nx=True)
-        res = await cache.get_async("test_key")
+        await cache.aset("test_key", 222, timeout=None)
+        await cache.aset("test_key", 222, timeout=-1, nx=True)
+        res = await cache.aget("test_key")
         assert res == 222
 
-    async def test_timeout_tiny(self):
-        await cache.set_async("test_key", 222, timeout=0.00001)
-        res = await cache.get_async("test_key")
+    async def test_timeout_tiny(self, cache: RedisCache):
+        await cache.aset("test_key", 222, timeout=0.00001)
+        res = await cache.aget("test_key")
         assert res in (None, 222)
 
-    async def test_set_add(self):
-        await cache.set_async("add_key", "Initial value")
-        res = await cache.add_async("add_key", "New value")
+    async def test_set_add(self, cache: RedisCache):
+        await cache.aset("add_key", "Initial value")
+        res = await cache.aadd("add_key", "New value")
         assert res is False
 
-        res = await cache.get_async("add_key")
+        res = await cache.aget("add_key")
         assert res == "Initial value"
 
-        res = await cache.add_async("other_key", "New value")
+        res = await cache.aadd("other_key", "New value")
         assert res is True
 
-    async def test_get_many(self):
-        await cache.set_async("a", 1)
-        await cache.set_async("b", 2)
-        await cache.set_async("c", 3)
+    async def test_get_many(self, cache: RedisCache):
+        await cache.aset("a", 1)
+        await cache.aset("b", 2)
+        await cache.aset("c", 3)
 
-        res = await cache.get_many_async(["a", "b", "c"])
+        res = await cache.aget_many(["a", "b", "c"])
         assert res == {"a": 1, "b": 2, "c": 3}
 
-    async def test_get_many_unicode(self):
-        await cache.set_async("a", "1")
-        await cache.set_async("b", "2")
-        await cache.set_async("c", "3")
+    async def test_get_many_unicode(self, cache: RedisCache):
+        await cache.aset("a", "1")
+        await cache.aset("b", "2")
+        await cache.aset("c", "3")
 
-        res = await cache.get_many_async(["a", "b", "c"])
+        res = await cache.aget_many(["a", "b", "c"])
         assert res == {"a": "1", "b": "2", "c": "3"}
 
-    async def test_set_many(self):
-        await cache.set_many_async({"a": 1, "b": 2, "c": 3})
-        res = await cache.get_many_async(["a", "b", "c"])
+    async def test_set_many(self, cache: RedisCache):
+        await cache.aset_many({"a": 1, "b": 2, "c": 3})
+        res = await cache.aget_many(["a", "b", "c"])
         assert res == {"a": 1, "b": 2, "c": 3}
 
     @pytest.mark.xfail(reason="Can't figure out how to test with mock")
-    async def test_set_call_empty_pipeline(self, mocker):
+    async def test_set_call_empty_pipeline(self, cache: RedisCache, mocker):
         pipeline = (await cache.client.get_client(write=True)).pipeline()
         key = "key"
         value = "value"
 
         mocked_set = mocker.patch.object(pipeline, "set")
-        cache.set_async(key, value, client=pipeline)
+        cache.aset(key, value, client=pipeline)
         await pipeline.execute()
 
         (await mocked_set).assert_called_once_with(
@@ -285,269 +284,271 @@ async def test_set_call_empty_pipeline(self, mocker):
             xx=False,
         )
 
-    async def test_delete(self):
-        await cache.set_many_async({"a": 1, "b": 2, "c": 3})
-        res = await cache.delete_async("a")
+    async def test_delete(self, cache: RedisCache):
+        await cache.aset_many({"a": 1, "b": 2, "c": 3})
+        res = await cache.adelete("a")
         assert bool(res) is True
 
-        res = await cache.get_many_async(["a", "b", "c"])
+        res = await cache.aget_many(["a", "b", "c"])
         assert res == {"b": 2, "c": 3}
 
-        res = await cache.delete_async("a")
+        res = await cache.adelete("a")
         assert bool(res) is False
 
-    async def test_delete_many(self):
-        await cache.set_many_async({"a": 1, "b": 2, "c": 3})
-        res = await cache.delete_many_async(["a", "b"])
+    async def test_delete_many(self, cache: RedisCache):
+        await cache.aset_many({"a": 1, "b": 2, "c": 3})
+        res = await cache.adelete_many(["a", "b"])
         assert bool(res) is True
 
-        res = await cache.get_many_async(["a", "b", "c"])
+        res = await cache.aget_many(["a", "b", "c"])
         assert res == {"c": 3}
 
-        res = await cache.delete_many_async(["a", "b"])
+        res = await cache.adelete_many(["a", "b"])
         assert bool(res) is False
 
-    async def test_delete_many_generator(self):
-        await cache.set_many_async({"a": 1, "b": 2, "c": 3})
-        res = await cache.delete_many_async(key for key in ["a", "b"])
+    async def test_delete_many_generator(self, cache: RedisCache):
+        await cache.aset_many({"a": 1, "b": 2, "c": 3})
+        res = await cache.adelete_many(key for key in ["a", "b"])
         assert bool(res) is True
 
-        res = await cache.get_many_async(["a", "b", "c"])
+        res = await cache.aget_many(["a", "b", "c"])
         assert res == {"c": 3}
 
-        res = await cache.delete_many_async(["a", "b"])
+        res = await cache.adelete_many(["a", "b"])
         assert bool(res) is False
 
-    async def test_delete_many_empty_generator(self):
-        res = await cache.delete_many_async(key for key in [])
+    async def test_delete_many_empty_generator(self, cache: RedisCache):
+        res = await cache.adelete_many(key for key in [])
         assert bool(res) is False
 
-    async def test_incr(self):
-        await cache.set_async("num", 1)
+    async def test_incr(self, cache: RedisCache):
+        await cache.aset("num", 1)
 
-        await cache.incr_async("num")
-        res = await cache.get_async("num")
+        await cache.aincr("num")
+        res = await cache.aget("num")
         assert res == 2
 
-        await cache.incr_async("num", 10)
-        res = await cache.get_async("num")
+        await cache.aincr("num", 10)
+        res = await cache.aget("num")
         assert res == 12
 
         # max 64 bit signed int
-        await cache.set_async("num", 9223372036854775807)
+        await cache.aset("num", 9223372036854775807)
 
-        await cache.incr_async("num")
-        res = await cache.get_async("num")
+        await cache.aincr("num")
+        res = await cache.aget("num")
         assert res == 9223372036854775808
 
-        await cache.incr_async("num", 2)
-        res = await cache.get_async("num")
+        await cache.aincr("num", 2)
+        res = await cache.aget("num")
         assert res == 9223372036854775810
 
-        await cache.set_async("num", 3)
+        await cache.aset("num", 3)
 
-        await cache.incr_async("num", 2)
-        res = await cache.get_async("num")
+        await cache.aincr("num", 2)
+        res = await cache.aget("num")
         assert res == 5
 
-    async def test_incr_error(self):
+    async def test_incr_error(self, cache: RedisCache):
         with pytest.raises(ValueError):
             # key does not exist
-            await cache.incr_async("numnum")
+            await cache.aincr("numnum")
 
-    async def test_incr_ignore_check(self):
+    async def test_incr_ignore_check(self, cache: RedisCache):
         # key exists check will be skipped and the value will be incremented by
         # '1' which is the default delta
-        await cache.incr_async("num", ignore_key_check=True)
-        res = await cache.get_async("num")
+        await cache.aincr("num", ignore_key_check=True)
+        res = await cache.aget("num")
         assert res == 1
-        await cache.delete_async("num")
+        await cache.adelete("num")
 
-        # since key doesnt exist it is set to the delta value, 10 in this case
-        await cache.incr_async("num", 10, ignore_key_check=True)
-        res = await cache.get_async("num")
+        # since key doesn't exist it is set to the delta value, 10 in this case
+        await cache.aincr("num", 10, ignore_key_check=True)
+        res = await cache.aget("num")
         assert res == 10
-        await cache.delete_async("num")
+        await cache.adelete("num")
 
         # following are just regression checks to make sure it still works as
         # expected with incr max 64 bit signed int
-        await cache.set_async("num", 9223372036854775807)
+        await cache.aset("num", 9223372036854775807)
 
-        await cache.incr_async("num", ignore_key_check=True)
-        res = await cache.get_async("num")
+        await cache.aincr("num", ignore_key_check=True)
+        res = await cache.aget("num")
         assert res == 9223372036854775808
 
-        await cache.incr_async("num", 2, ignore_key_check=True)
-        res = await cache.get_async("num")
+        await cache.aincr("num", 2, ignore_key_check=True)
+        res = await cache.aget("num")
         assert res == 9223372036854775810
 
-        await cache.set_async("num", 3)
+        await cache.aset("num", 3)
 
-        await cache.incr_async("num", 2, ignore_key_check=True)
-        res = await cache.get_async("num")
+        await cache.aincr("num", 2, ignore_key_check=True)
+        res = await cache.aget("num")
         assert res == 5
 
-    async def test_get_set_bool(self):
-        await cache.set_async("bool", True)
-        res = await cache.get_async("bool")
+    async def test_get_set_bool(self, cache: RedisCache):
+        await cache.aset("bool", True)
+        res = await cache.aget("bool")
 
         assert isinstance(res, bool)
         assert res is True
 
-        await cache.set_async("bool", False)
-        res = await cache.get_async("bool")
+        await cache.aset("bool", False)
+        res = await cache.aget("bool")
 
         assert isinstance(res, bool)
         assert res is False
 
-    async def test_decr(self):
-        await cache.set_async("num", 20)
+    async def test_decr(self, cache: RedisCache):
+        await cache.aset("num", 20)
 
-        await cache.decr_async("num")
-        res = await cache.get_async("num")
+        await cache.adecr("num")
+        res = await cache.aget("num")
         assert res == 19
 
-        await cache.decr_async("num", 20)
-        res = await cache.get_async("num")
+        await cache.adecr("num", 20)
+        res = await cache.aget("num")
         assert res == -1
 
-        await cache.decr_async("num", 2)
-        res = await cache.get_async("num")
+        await cache.adecr("num", 2)
+        res = await cache.aget("num")
         assert res == -3
 
-        await cache.set_async("num", 20)
+        await cache.aset("num", 20)
 
-        await cache.decr_async("num")
-        res = await cache.get_async("num")
+        await cache.adecr("num")
+        res = await cache.aget("num")
         assert res == 19
 
         # max 64 bit signed int + 1
-        await cache.set_async("num", 9223372036854775808)
+        await cache.aset("num", 9223372036854775808)
 
-        await cache.decr_async("num")
-        res = await cache.get_async("num")
+        await cache.adecr("num")
+        res = await cache.aget("num")
         assert res == 9223372036854775807
 
-        await cache.decr_async("num", 2)
-        res = await cache.get_async("num")
+        await cache.adecr("num", 2)
+        res = await cache.aget("num")
         assert res == 9223372036854775805
 
-    async def test_version(self):
-        await cache.set_async("keytest", 2, version=2)
-        res = await cache.get_async("keytest")
+    async def test_version(self, cache: RedisCache):
+        await cache.aset("keytest", 2, version=2)
+        res = await cache.aget("keytest")
         assert res is None
 
-        res = await cache.get_async("keytest", version=2)
+        res = await cache.aget("keytest", version=2)
         assert res == 2
 
-    async def test_incr_version(self):
-        await cache.set_async("keytest", 2)
-        await cache.incr_version_async("keytest")
+    async def test_incr_version(self, cache: RedisCache):
+        await cache.aset("keytest", 2)
+        await cache.aincr_version("keytest")
 
-        res = await cache.get_async("keytest")
+        res = await cache.aget("keytest")
         assert res is None
 
-        res = await cache.get_async("keytest", version=2)
+        res = await cache.aget("keytest", version=2)
         assert res == 2
 
-    async def test_delete_pattern(self):
+    async def test_delete_pattern(self, cache: RedisCache):
         for key in ["foo-aa", "foo-ab", "foo-bb", "foo-bc"]:
-            await cache.set_async(key, "foo")
+            await cache.aset(key, "foo")
 
-        res = await cache.delete_pattern_async("*foo-a*")
+        res = await cache.adelete_pattern("*foo-a*")
         assert bool(res) is True
 
-        keys = await cache.keys_async("foo*")
+        keys = await cache.akeys("foo*")
         assert set(keys) == {"foo-bb", "foo-bc"}
 
-        res = await cache.delete_pattern_async("*foo-a*")
+        res = await cache.adelete_pattern("*foo-a*")
         assert bool(res) is False
 
     @pytest.mark.xfail(reason="Can't figure out how to test with mock")
-    async def test_delete_pattern_with_custom_count(self, mocker):
+    async def test_delete_pattern_with_custom_count(self, cache: RedisCache, mocker):
         client_mock = mocker.patch("django_async_redis.cache.RedisCache.client")
         for key in ["foo-aa", "foo-ab", "foo-bb", "foo-bc"]:
-            await cache.set_async(key, "foo")
+            await cache.aset(key, "foo")
 
-        await cache.delete_pattern_async("*foo-a*", itersize=2)
+        await cache.adelete_pattern("*foo-a*", itersize=2)
 
-        (await client_mock).delete_pattern_async.assert_called_once_with(
+        (await client_mock).adelete_pattern.assert_called_once_with(
             "*foo-a*", itersize=2
         )
 
     @pytest.mark.xfail(reason="Can't figure out how to test with mock")
-    async def test_delete_pattern_with_settings_default_scan_count(self, mocker):
+    async def test_delete_pattern_with_settings_default_scan_count(
+        self, cache: RedisCache, mocker
+    ):
         client_mock = mocker.patch("django_async_redis.cache.RedisCache.client")
         for key in ["foo-aa", "foo-ab", "foo-bb", "foo-bc"]:
-            await cache.set_async(key, "foo")
+            await cache.aset(key, "foo")
         expected_count = DJANGO_ASYNC_REDIS_SCAN_ITERSIZE
 
-        await cache.delete_pattern_async("*foo-a*")
+        await cache.adelete_pattern("*foo-a*")
 
-        (await client_mock).delete_pattern_async.assert_called_once_with(
+        (await client_mock).adelete_pattern.assert_awaited_once_with(
             "*foo-a*", itersize=expected_count
         )
 
-    async def test_close(self):
+    async def test_close(self, cache: RedisCache):
         _cache = caches["default"]
-        await _cache.set_async("f", "1")
-        await _cache.close_async()
+        await _cache.aset("f", "1")
+        await _cache.aclose()
 
-    async def test_ttl(self):
+    async def test_ttl(self, cache: RedisCache):
         _cache = caches["default"]
 
         # Test ttl
-        await _cache.set_async("foo", "bar", 10)
-        ttl = await _cache.ttl_async("foo")
+        await _cache.aset("foo", "bar", 10)
+        ttl = await _cache.attl("foo")
         assert ttl == 10
 
         # Test ttl None
-        await _cache.set_async("foo", "foo", timeout=None)
-        ttl = await _cache.ttl_async("foo")
+        await _cache.aset("foo", "foo", timeout=None)
+        ttl = await _cache.attl("foo")
         assert ttl is None
 
         # Test ttl with expired key
-        await _cache.set_async("foo", "foo", timeout=-1)
-        ttl = await _cache.ttl_async("foo")
+        await _cache.aset("foo", "foo", timeout=-1)
+        ttl = await _cache.attl("foo")
         assert ttl == 0
 
         # Test ttl with not existent key
-        ttl = await _cache.ttl_async("not-existent-key")
+        ttl = await _cache.attl("not-existent-key")
         assert ttl == 0
 
-    async def test_persist(self):
-        await cache.set_async("foo", "bar", timeout=20)
-        await cache.persist_async("foo")
+    async def test_persist(self, cache: RedisCache):
+        await cache.aset("foo", "bar", timeout=20)
+        await cache.apersist("foo")
 
-        ttl = await cache.ttl_async("foo")
+        ttl = await cache.attl("foo")
         assert ttl is None
 
-    async def test_expire(self):
-        await cache.set_async("foo", "bar", timeout=None)
-        await cache.expire_async("foo", 20)
-        ttl = await cache.ttl_async("foo")
+    async def test_expire(self, cache: RedisCache):
+        await cache.aset("foo", "bar", timeout=None)
+        await cache.aexpire("foo", 20)
+        ttl = await cache.attl("foo")
         assert pytest.approx(ttl) == 20
 
-    async def test_iter_keys(self):
+    async def test_iter_keys(self, cache: RedisCache):
         _cache = caches["default"]
 
-        await _cache.set_async("foo1", 1)
-        await _cache.set_async("foo2", 1)
-        await _cache.set_async("foo3", 1)
+        await _cache.aset("foo1", 1)
+        await _cache.aset("foo2", 1)
+        await _cache.aset("foo3", 1)
 
         # Test simple result
-        result = set([x async for x in await _cache.iter_keys_async("*")])
+        result = set([x async for x in await _cache.aiter_keys("*")])
         assert result == {"foo1", "foo2", "foo3"}
 
         # Test limited result
-        result = [x async for x in await _cache.iter_keys_async("foo*", itersize=2)]
+        result = [x async for x in await _cache.aiter_keys("foo*", itersize=2)]
         assert len(result) == 3
 
         # Test generator object
-        result = await _cache.iter_keys_async("foo*")
+        result = await _cache.aiter_keys("foo*")
         assert result.__anext__() is not None
 
-    async def test_primary_replica_switching(self):
+    async def test_primary_replica_switching(self, cache: RedisCache):
         _cache = caches["sample"]
         client = cache.client
         client._server = ["foo", "bar"]
@@ -556,46 +557,46 @@ async def test_primary_replica_switching(self):
         assert await client.get_client(write=True) == "Foo"
         assert await client.get_client(write=False) == "Bar"
 
-    async def test_touch_zero_timeout(self):
-        await cache.set_async("test_key", 222, timeout=10)
+    async def test_touch_zero_timeout(self, cache: RedisCache):
+        await cache.aset("test_key", 222, timeout=10)
 
-        assert await cache.touch_async("test_key", 0) is True
-        res = await cache.get_async("test_key")
+        assert await cache.atouch("test_key", 0) is True
+        res = await cache.aget("test_key")
         assert res is None
 
-    async def test_touch_positive_timeout(self):
-        await cache.set_async("test_key", 222, timeout=10)
+    async def test_touch_positive_timeout(self, cache: RedisCache):
+        await cache.aset("test_key", 222, timeout=10)
 
-        assert await cache.touch_async("test_key", 2) is True
-        assert await cache.get_async("test_key") == 222
+        assert await cache.atouch("test_key", 2) is True
+        assert await cache.aget("test_key") == 222
         await asyncio.sleep(3)
-        assert await cache.get_async("test_key") is None
+        assert await cache.aget("test_key") is None
 
-    async def test_touch_negative_timeout(self):
-        await cache.set_async("test_key", 222, timeout=10)
+    async def test_touch_negative_timeout(self, cache: RedisCache):
+        await cache.aset("test_key", 222, timeout=10)
 
-        assert await cache.touch_async("test_key", -1) is True
-        res = await cache.get_async("test_key")
+        assert await cache.atouch("test_key", -1) is True
+        res = await cache.aget("test_key")
         assert res is None
 
-    async def test_touch_missed_key(self):
-        assert await cache.touch_async("test_key_does_not_exist", 1) is False
+    async def test_touch_missed_key(self, cache: RedisCache):
+        assert await cache.atouch("test_key_does_not_exist", 1) is False
 
-    async def test_touch_forever(self):
-        await cache.set_async("test_key", "foo", timeout=1)
-        result = await cache.touch_async("test_key", None)
+    async def test_touch_forever(self, cache: RedisCache):
+        await cache.aset("test_key", "foo", timeout=1)
+        result = await cache.atouch("test_key", None)
         assert result is True
-        assert await cache.ttl_async("test_key") is None
+        assert await cache.attl("test_key") is None
         await asyncio.sleep(2)
-        assert await cache.get_async("test_key") == "foo"
+        assert await cache.aget("test_key") == "foo"
 
-    async def test_touch_forever_nonexistent(self):
-        result = await cache.touch_async("test_key_does_not_exist", None)
+    async def test_touch_forever_nonexistent(self, cache: RedisCache):
+        result = await cache.atouch("test_key_does_not_exist", None)
         assert result is False
 
-    async def test_touch_default_timeout(self):
-        await cache.set_async("test_key", "foo", timeout=1)
-        result = await cache.touch_async("test_key")
+    async def test_touch_default_timeout(self, cache: RedisCache):
+        await cache.aset("test_key", "foo", timeout=1)
+        result = await cache.atouch("test_key")
         assert result is True
         await asyncio.sleep(2)
-        assert await cache.get_async("test_key") == "foo"
+        assert await cache.aget("test_key") == "foo"