Linting and hints

django_q/brokers/redis_broker.py

@@ -1,4 +1,5 @@
 import redis
+from redis import Redis
 
 from django_q.brokers import Broker
 from django_q.conf import Conf, logger
@@ -10,7 +11,7 @@ except ImportError:
 
 
 class Redis(Broker):
-    def __init__(self, list_key=Conf.PREFIX):
+    def __init__(self, list_key: str = Conf.PREFIX):
         super(Redis, self).__init__(list_key=f"django_q:{list_key}:q")
 
     def enqueue(self, task):
@@ -30,33 +31,33 @@ class Redis(Broker):
     def purge_queue(self):
         return self.connection.ltrim(self.list_key, 1, 0)
 
-    def ping(self):
+    def ping(self) -> bool:
         try:
             return self.connection.ping()
         except redis.ConnectionError as e:
             logger.error("Can not connect to Redis server.")
             raise e
 
-    def info(self):
+    def info(self) -> str:
         if not self._info:
             info = self.connection.info("server")
             self._info = f"Redis {info['redis_version']}"
         return self._info
 
-    def set_stat(self, key, value, timeout):
+    def set_stat(self, key: str, value: str, timeout: int):
         self.connection.set(key, value, timeout)
 
-    def get_stat(self, key):
+    def get_stat(self, key: str):
         if self.connection.exists(key):
             return self.connection.get(key)
 
-    def get_stats(self, pattern):
+    def get_stats(self, pattern: str):
         keys = self.connection.keys(pattern=pattern)
         if keys:
             return self.connection.mget(keys)
 
     @staticmethod
-    def get_connection(list_key=Conf.PREFIX):
+    def get_connection(list_key: str = Conf.PREFIX) -> Redis:
         if django_redis and Conf.DJANGO_REDIS:
             return django_redis.get_redis_connection(Conf.DJANGO_REDIS)
         if isinstance(Conf.REDIS, str):