Merge pull request #449 from Koed00/type_hints

Adds hint, some linting and a release drafter

.github/release-drafter.yml

@@ -0,0 +1,36 @@
+categories:
+  - 
+    label: breaking
+    title: Breaking
+  - 
+    label: feature
+    title: New
+  - 
+    label: bug
+    title: "Bug Fixes"
+  - 
+    label: dependencies
+    title: "Dependency Updates"
+  -
+    label: security
+    title: Security
+name-template: v$NEXT_PATCH_VERSION
+tag-template: v$NEXT_PATCH_VERSION
+template: |
+    $CHANGES
+version-resolver: 
+  major: 
+    labels: 
+      - breaking
+      - major
+  minor: 
+    labels: 
+      - feature
+      - minor
+  patch: 
+    labels: 
+      - bug
+      - dependencies
+      - security
+      - patch
+  default: patch

.github/workflows/release_log.yml

@@ -0,0 +1,14 @@
+name: Update release draft
+on:
+  push:
+    branches:
+      - master
+jobs:
+  update_release_draft:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: release-drafter/release-drafter@v5
+        with:
+            config-name: release-drafter.yml
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

django_q/brokers/__init__.py

@@ -1,4 +1,5 @@
 import importlib
+from typing import Optional
 
 from django.core.cache import caches, InvalidCacheBackendError
 
@@ -6,7 +7,7 @@ from django_q.conf import Conf
 
 
 class Broker:
-    def __init__(self, list_key=Conf.PREFIX):
+    def __init__(self, list_key: str = Conf.PREFIX):
         self.connection = self.get_connection(list_key)
         self.list_key = list_key
         self.cache = self.get_cache()
@@ -71,7 +72,7 @@ class Broker:
         :return:
         """
 
-    def ping(self):
+    def ping(self) -> bool:
         """
         Checks whether the broker connection is available
         :rtype: bool
@@ -84,7 +85,7 @@ class Broker:
         """
         return self._info
 
-    def set_stat(self, key, value, timeout):
+    def set_stat(self, key: str, value: str, timeout: int):
         """
         Saves a cluster statistic to the cache provider
         :type key: str
@@ -99,7 +100,7 @@ class Broker:
         self.cache.set(Conf.Q_STAT, key_list)
         return self.cache.set(key, value, timeout)
 
-    def get_stat(self, key):
+    def get_stat(self, key: str):
         """
         Gets a cluster statistic from the cache provider
         :type key: str
@@ -109,7 +110,7 @@ class Broker:
             return
         return self.cache.get(key)
 
-    def get_stats(self, pattern):
+    def get_stats(self, pattern: str) -> Optional[list]:
         """
         Returns a list of all cluster stats from the cache provider
         :type pattern: str
@@ -142,7 +143,7 @@ class Broker:
             return None
 
     @staticmethod
-    def get_connection(list_key=Conf.PREFIX):
+    def get_connection(list_key: str = Conf.PREFIX):
         """
         Gets a connection to the broker
         :param list_key: Optional queue name
@@ -151,40 +152,46 @@ class Broker:
         return 0
 
 
-def get_broker(list_key=Conf.PREFIX):
+def get_broker(list_key: str = Conf.PREFIX) -> Broker:
     """
     Gets the configured broker type
     :param list_key: optional queue name
     :type list_key: str
-    :return:
+    :return: a broker instance
     """
     # custom
     if Conf.BROKER_CLASS:
-        module, func = Conf.BROKER_CLASS.rsplit('.', 1)
+        module, func = Conf.BROKER_CLASS.rsplit(".", 1)
         m = importlib.import_module(module)
         broker = getattr(m, func)
         return broker(list_key=list_key)
     # disque
     elif Conf.DISQUE_NODES:
         from django_q.brokers import disque
+
         return disque.Disque(list_key=list_key)
     # Iron MQ
     elif Conf.IRON_MQ:
         from django_q.brokers import ironmq
+
         return ironmq.IronMQBroker(list_key=list_key)
     # SQS
     elif Conf.SQS:
         from django_q.brokers import aws_sqs
+
         return aws_sqs.Sqs(list_key=list_key)
     # ORM
     elif Conf.ORM:
         from django_q.brokers import orm
+
         return orm.ORM(list_key=list_key)
     # Mongo
     elif Conf.MONGO:
         from django_q.brokers import mongo
+
         return mongo.Mongo(list_key=list_key)
     # default to redis
     else:
         from django_q.brokers import redis_broker
+
         return redis_broker.Redis(list_key=list_key)

django_q/brokers/aws_sqs.py

@@ -1,34 +1,37 @@
-from django_q.conf import Conf
-from django_q.brokers import Broker
 from boto3 import Session
 
+from django_q.brokers import Broker
+from django_q.conf import Conf
+
 
 class Sqs(Broker):
-    def __init__(self, list_key=Conf.PREFIX):
+    def __init__(self, list_key: str = Conf.PREFIX):
         self.sqs = None
         super(Sqs, self).__init__(list_key)
         self.queue = self.get_queue()
 
     def enqueue(self, task):
         response = self.queue.send_message(MessageBody=task)
-        return response.get('MessageId')
+        return response.get("MessageId")
 
     def dequeue(self):
         # sqs supports max 10 messages in bulk
         if Conf.BULK > 10:
             Conf.BULK = 10
-        tasks = self.queue.receive_messages(MaxNumberOfMessages=Conf.BULK, VisibilityTimeout=Conf.RETRY)
+        tasks = self.queue.receive_messages(
+            MaxNumberOfMessages=Conf.BULK, VisibilityTimeout=Conf.RETRY
+        )
         if tasks:
             return [(t.receipt_handle, t.body) for t in tasks]
 
     def acknowledge(self, task_id):
         return self.delete(task_id)
 
-    def queue_size(self):
-        return int(self.queue.attributes['ApproximateNumberOfMessages'])
+    def queue_size(self) -> int:
+        return int(self.queue.attributes["ApproximateNumberOfMessages"])
 
-    def lock_size(self):
-        return int(self.queue.attributes['ApproximateNumberOfMessagesNotVisible'])
+    def lock_size(self) -> int:
+        return int(self.queue.attributes["ApproximateNumberOfMessagesNotVisible"])
 
     def delete(self, task_id):
         message = self.sqs.Message(self.queue.url, task_id)
@@ -43,20 +46,20 @@ class Sqs(Broker):
     def purge_queue(self):
         self.queue.purge()
 
-    def ping(self):
-        return 'sqs' in self.connection.get_available_resources()
+    def ping(self) -> bool:
+        return "sqs" in self.connection.get_available_resources()
 
-    def info(self):
-        return 'AWS SQS'
+    def info(self) -> str:
+        return "AWS SQS"
 
     @staticmethod
-    def get_connection(list_key=Conf.PREFIX):
+    def get_connection(list_key: str = Conf.PREFIX) -> Session:
         config = Conf.SQS
-        if 'aws_region' in config:
-            config['region_name'] = config['aws_region']
-            del(config['aws_region'])
+        if "aws_region" in config:
+            config["region_name"] = config["aws_region"]
+            del config["aws_region"]
         return Session(**config)
 
     def get_queue(self):
-        self.sqs = self.connection.resource('sqs')
+        self.sqs = self.connection.resource("sqs")
         return self.sqs.create_queue(QueueName=self.list_key)

django_q/brokers/disque.py

@@ -1,5 +1,8 @@
 import random
+
 import redis
+from redis import Redis
+
 from django_q.brokers import Broker
 from django_q.conf import Conf
 
@@ -25,7 +28,7 @@ class Disque(Broker):
         command = "FASTACK" if Conf.DISQUE_FASTACK else "ACKJOB"
         return self.connection.execute_command(f"{command} {task_id}")
 
-    def ping(self):
+    def ping(self) -> bool:
         return self.connection.execute_command("HELLO")[0] > 0
 
     def delete(self, task_id):
@@ -34,21 +37,21 @@ class Disque(Broker):
     def fail(self, task_id):
         return self.delete(task_id)
 
-    def delete_queue(self):
+    def delete_queue(self) -> int:
         jobs = self.connection.execute_command(f"JSCAN QUEUE {self.list_key}")[1]
         if jobs:
             job_ids = " ".join(jid.decode() for jid in jobs)
             self.connection.execute_command(f"DELJOB {job_ids}")
         return len(jobs)
 
-    def info(self):
+    def info(self) -> str:
         if not self._info:
             info = self.connection.info("server")
             self._info = f'Disque {info["disque_version"]}'
         return self._info
 
     @staticmethod
-    def get_connection(list_key=Conf.PREFIX):
+    def get_connection(list_key: str = Conf.PREFIX) -> Redis:
         # randomize nodes
         random.shuffle(Conf.DISQUE_NODES)
         # find one that works

django_q/brokers/ironmq.py

@@ -1,31 +1,32 @@
+from iron_mq import IronMQ, Queue
 from requests.exceptions import HTTPError
-from django_q.conf import Conf
+
 from django_q.brokers import Broker
-from iron_mq import IronMQ
+from django_q.conf import Conf
 
 
 class IronMQBroker(Broker):
     def enqueue(self, task):
-        return self.connection.post(task)['ids'][0]
+        return self.connection.post(task)["ids"][0]
 
     def dequeue(self):
         timeout = Conf.RETRY or None
-        tasks = self.connection.get(timeout=timeout, wait=1, max=Conf.BULK)['messages']
+        tasks = self.connection.get(timeout=timeout, wait=1, max=Conf.BULK)["messages"]
         if tasks:
-            return [(t['id'], t['body']) for t in tasks]
+            return [(t["id"], t["body"]) for t in tasks]
 
-    def ping(self):
+    def ping(self) -> bool:
         return self.connection.name == self.list_key
 
-    def info(self):
-        return 'IronMQ'
+    def info(self) -> str:
+        return "IronMQ"
 
     def queue_size(self):
         return self.connection.size()
 
     def delete_queue(self):
         try:
-            return self.connection.delete_queue()['msg']
+            return self.connection.delete_queue()["msg"]
         except HTTPError:
             return False
 
@@ -34,7 +35,7 @@ class IronMQBroker(Broker):
 
     def delete(self, task_id):
         try:
-            return self.connection.delete(task_id)['msg']
+            return self.connection.delete(task_id)["msg"]
         except HTTPError:
             return False
 
@@ -45,6 +46,6 @@ class IronMQBroker(Broker):
         return self.delete(task_id)
 
     @staticmethod
-    def get_connection(list_key=Conf.PREFIX):
+    def get_connection(list_key: str = Conf.PREFIX) -> Queue:
         ironmq = IronMQ(name=None, **Conf.IRON_MQ)
         return ironmq.queue(queue_name=list_key)

django_q/brokers/mongo.py

@@ -4,7 +4,6 @@ from time import sleep
 from bson import ObjectId
 from django.utils import timezone
 from pymongo import MongoClient
-
 from pymongo.errors import ConfigurationError
 
 from django_q.brokers import Broker
@@ -21,7 +20,7 @@ class Mongo(Broker):
         self.collection = self.get_collection()
 
     @staticmethod
-    def get_connection(list_key=Conf.PREFIX):
+    def get_connection(list_key: str = Conf.PREFIX) -> MongoClient:
         return MongoClient(**Conf.MONGO)
 
     def get_collection(self):
@@ -41,10 +40,10 @@ class Mongo(Broker):
     def purge_queue(self):
         return self.delete_queue()
 
-    def ping(self):
+    def ping(self) -> bool:
         return self.info is not None
 
-    def info(self):
+    def info(self) -> str:
         if not self._info:
             self._info = f"MongoDB {self.connection.server_info()['version']}"
         return self._info

django_q/brokers/orm.py

@@ -1,13 +1,13 @@
 from datetime import timedelta
 from time import sleep
 
-from django.utils import timezone
 from django import db
 from django.db import transaction
+from django.utils import timezone
 
 from django_q.brokers import Broker
-from django_q.models import OrmQ
 from django_q.conf import Conf, logger
+from django_q.models import OrmQ
 
 
 def _timeout():
@@ -16,8 +16,10 @@ def _timeout():
 
 class ORM(Broker):
     @staticmethod
-    def get_connection(list_key=Conf.PREFIX):
-        if transaction.get_autocommit(using=Conf.ORM):  # Only True when not in an atomic block
+    def get_connection(list_key: str = Conf.PREFIX):
+        if transaction.get_autocommit(
+            using=Conf.ORM
+        ):  # Only True when not in an atomic block
             # Make sure stale connections in the broker thread are explicitly
             #   closed before attempting DB access.
             # logger.debug("Broker thread calling close_old_connections")
@@ -26,14 +28,14 @@ class ORM(Broker):
             logger.debug("Broker in an atomic transaction")
         return OrmQ.objects.using(Conf.ORM)
 
-    def queue_size(self):
+    def queue_size(self) -> int:
         return (
             self.get_connection()
             .filter(key=self.list_key, lock__lte=_timeout())
             .count()
         )
 
-    def lock_size(self):
+    def lock_size(self) -> int:
         return (
             self.get_connection().filter(key=self.list_key, lock__gt=_timeout()).count()
         )
@@ -41,10 +43,10 @@ class ORM(Broker):
     def purge_queue(self):
         return self.get_connection().filter(key=self.list_key).delete()
 
-    def ping(self):
+    def ping(self) -> bool:
         return True
 
-    def info(self):
+    def info(self) -> str:
         if not self._info:
             self._info = f"ORM {Conf.ORM}"
         return self._info
@@ -60,7 +62,7 @@ class ORM(Broker):
 
     def dequeue(self):
         tasks = self.get_connection().filter(key=self.list_key, lock__lt=_timeout())[
-            0: Conf.BULK
+            0 : Conf.BULK
         ]
         if tasks:
             task_list = []

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):

django_q/cluster.py

@@ -20,7 +20,7 @@ from django.utils.translation import gettext_lazy as _
 
 # Local
 import django_q.tasks
-from django_q.brokers import get_broker
+from django_q.brokers import get_broker, Broker
 from django_q.conf import Conf, logger, psutil, get_ppid, error_reporter
 from django_q.humanhash import humanize
 from django_q.models import Task, Success, Schedule
@@ -31,7 +31,7 @@ from django_q.status import Stat, Status
 
 
 class Cluster:
-    def __init__(self, broker=None):
+    def __init__(self, broker: Broker = None):
         self.broker = broker or get_broker()
         self.sentinel = None
         self.stop_event = None
@@ -43,7 +43,7 @@ class Cluster:
         signal.signal(signal.SIGTERM, self.sig_handler)
         signal.signal(signal.SIGINT, self.sig_handler)
 
-    def start(self):
+    def start(self) -> int:
         # Start Sentinel
         self.stop_event = Event()
         self.start_event = Event()
@@ -63,7 +63,7 @@ class Cluster:
             sleep(0.1)
         return self.pid
 
-    def stop(self):
+    def stop(self) -> bool:
         if not self.sentinel.is_alive():
             return False
         logger.info(_(f"Q Cluster {self.name} stopping."))
@@ -83,25 +83,25 @@ class Cluster:
         self.stop()
 
     @property
-    def stat(self):
+    def stat(self) -> Status:
         if self.sentinel:
             return Stat.get(pid=self.pid, cluster_id=self.cluster_id)
         return Status(pid=self.pid, cluster_id=self.cluster_id)
 
     @property
-    def name(self):
+    def name(self) -> str:
         return humanize(self.cluster_id.hex)
 
     @property
-    def is_starting(self):
+    def is_starting(self) -> bool:
         return self.stop_event and self.start_event and not self.start_event.is_set()
 
     @property
-    def is_running(self):
+    def is_running(self) -> bool:
         return self.stop_event and self.start_event and self.start_event.is_set()
 
     @property
-    def is_stopping(self):
+    def is_stopping(self) -> bool:
         return (
             self.stop_event
             and self.start_event
@@ -110,7 +110,7 @@ class Cluster:
         )
 
     @property
-    def has_stopped(self):
+    def has_stopped(self) -> bool:
         return self.start_event is None and self.stop_event is None and self.sentinel
 
 
@@ -154,7 +154,7 @@ class Sentinel:
         self.spawn_cluster()
         self.guard()
 
-    def status(self):
+    def status(self) -> str:
         if not self.start_event.is_set() and not self.stop_event.is_set():
             return Conf.STARTING
         elif self.start_event.is_set() and not self.stop_event.is_set():
@@ -166,7 +166,7 @@ class Sentinel:
                 return Conf.STOPPING
             return Conf.STOPPED
 
-    def spawn_process(self, target, *args):
+    def spawn_process(self, target, *args) -> Process:
         """
         :type target: function or class
         """
@@ -179,7 +179,7 @@ class Sentinel:
         p.start()
         return p
 
-    def spawn_pusher(self):
+    def spawn_pusher(self) -> Process:
         return self.spawn_process(pusher, self.task_queue, self.event_out, self.broker)
 
     def spawn_worker(self):
@@ -187,7 +187,7 @@ class Sentinel:
             worker, self.task_queue, self.result_queue, Value("f", -1), self.timeout
         )
 
-    def spawn_monitor(self):
+    def spawn_monitor(self) -> Process:
         return self.spawn_process(monitor, self.result_queue, self.broker)
 
     def reincarnate(self, process):
@@ -310,9 +310,10 @@ class Sentinel:
         Stat(self).save()
 
 
-def pusher(task_queue, event, broker=None):
+def pusher(task_queue: Queue, event: Event, broker: Broker = None):
     """
     Pulls tasks of the broker and puts them in the task queue
+    :type broker:
     :type task_queue: multiprocessing.Queue
     :type event: multiprocessing.Event
     """
@@ -345,9 +346,10 @@ def pusher(task_queue, event, broker=None):
     logger.info(_(f"{current_process().name} stopped pushing tasks"))
 
 
-def monitor(result_queue, broker=None):
+def monitor(result_queue: Queue, broker: Broker = None):
     """
     Gets finished tasks from the result queue and saves them to Django
+    :type broker: brokers.Broker
     :type result_queue: multiprocessing.Queue
     """
     if not broker:
@@ -374,9 +376,12 @@ def monitor(result_queue, broker=None):
     logger.info(_(f"{name} stopped monitoring results"))
 
 
-def worker(task_queue, result_queue, timer, timeout=Conf.TIMEOUT):
+def worker(
+    task_queue: Queue, result_queue: Queue, timer: Value, timeout: int = Conf.TIMEOUT
+):
     """
     Takes a task from the task queue, tries to execute it and puts the result back in the result queue
+    :param timeout: number of seconds wait for a worker to finish.
     :type task_queue: multiprocessing.Queue
     :type result_queue: multiprocessing.Queue
     :type timer: multiprocessing.Value
@@ -435,9 +440,11 @@ def worker(task_queue, result_queue, timer, timeout=Conf.TIMEOUT):
     logger.info(_(f"{name} stopped doing work"))
 
 
-def save_task(task, broker):
+def save_task(task, broker: Broker):
     """
     Saves the task package to Django or the cache
+    :param task: the task package
+    :type broker: brokers.Broker
     """
     # SAVE LIMIT < 0 : Don't save success
     if not task.get("save", Conf.SAVE_LIMIT >= 0) and task["success"]:
@@ -483,7 +490,7 @@ def save_task(task, broker):
         logger.error(e)
 
 
-def save_cached(task, broker):
+def save_cached(task, broker: Broker):
     task_key = f'{broker.list_key}:{task["id"]}'
     timeout = task["cached"]
     if timeout is True:
@@ -535,7 +542,7 @@ def save_cached(task, broker):
         logger.error(e)
 
 
-def scheduler(broker=None):
+def scheduler(broker: Broker = None):
     """
     Creates a task from a schedule at the scheduled time and schedules next run
     """
@@ -588,7 +595,11 @@ def scheduler(broker=None):
                             break
                     # arrow always returns a tz aware datetime, and we don't want
                     # this when we explicitly configured django with USE_TZ=False
-                    s.next_run = next_run.datetime if settings.USE_TZ else next_run.datetime.replace(tzinfo=None)
+                    s.next_run = (
+                        next_run.datetime
+                        if settings.USE_TZ
+                        else next_run.datetime.replace(tzinfo=None)
+                    )
                     s.repeats += -1
                 # send it to the cluster
                 q_options["broker"] = broker
@@ -635,7 +646,7 @@ def close_old_django_connections():
         db.close_old_connections()
 
 
-def set_cpu_affinity(n, process_ids, actual=not Conf.TESTING):
+def set_cpu_affinity(n: int, process_ids: list, actual: bool = not Conf.TESTING):
     """
     Sets the cpu affinity for the supplied processes.
     Requires the optional psutil module.

django_q/conf.py

@@ -1,6 +1,4 @@
 import logging
-
-# external
 import os
 from copy import deepcopy
 from multiprocessing import cpu_count
@@ -8,11 +6,8 @@ from signal import signal
 
 import pkg_resources
 from django.conf import settings
-
-# django
 from django.utils.translation import gettext_lazy as _
 
-# local
 from django_q.queues import Queue
 
 # optional
@@ -216,7 +211,7 @@ if Conf.ERROR_REPORTER:
         # and instantiate an ErrorReporter using the provided config
         for name, conf in error_conf.items():
             for entry in pkg_resources.iter_entry_points(
-                "djangoq.errorreporters", name
+                    "djangoq.errorreporters", name
             ):
                 Reporter = entry.load()
                 reporters.append(Reporter(**conf))

django_q/core_signing.py

@@ -2,8 +2,15 @@ import datetime
 import time
 import zlib
 
-from django.core.signing import BadSignature, SignatureExpired, b64_decode, JSONSerializer, \
-    Signer as Sgnr, TimestampSigner as TsS, dumps
+from django.core.signing import (
+    BadSignature,
+    SignatureExpired,
+    b64_decode,
+    JSONSerializer,
+    Signer as Sgnr,
+    TimestampSigner as TsS,
+    dumps,
+)
 from django.utils import baseconv
 from django.utils.crypto import constant_time_compare
 from django.utils.encoding import force_bytes, force_str
@@ -16,7 +23,13 @@ The difference is that `this` loads function calls `TimestampSigner` and `Signer
 """
 
 
-def loads(s, key=None, salt='django.core.signing', serializer=JSONSerializer, max_age=None):
+def loads(
+        s,
+        key=None,
+        salt: str = "django.core.signing",
+        serializer=JSONSerializer,
+        max_age=None,
+):
     """
     Reverse of dumps(), raise BadSignature if signature fails.
 
@@ -26,7 +39,7 @@ def loads(s, key=None, salt='django.core.signing', serializer=JSONSerializer, ma
     # operate on bytes.
     base64d = force_bytes(TimestampSigner(key, salt=salt).unsign(s, max_age=max_age))
     decompress = False
-    if base64d[:1] == b'.':
+    if base64d[:1] == b".":
         # It's compressed; uncompress it first
         base64d = base64d[1:]
         decompress = True
@@ -37,7 +50,6 @@ def loads(s, key=None, salt='django.core.signing', serializer=JSONSerializer, ma
 
 
 class Signer(Sgnr):
-
     def unsign(self, signed_value):
         signed_value = force_str(signed_value)
         if self.sep not in signed_value:
@@ -55,7 +67,6 @@ calling `this` Signer.
 
 
 class TimestampSigner(Signer, TsS):
-
     def unsign(self, value, max_age=None):
         """
         Retrieve original value and check it wasn't signed more
@@ -70,6 +81,5 @@ class TimestampSigner(Signer, TsS):
             # Check timestamp is not older than max_age
             age = time.time() - timestamp
             if age > max_age:
-                raise SignatureExpired(
-                    'Signature age %s > %s seconds' % (age, max_age))
+                raise SignatureExpired("Signature age %s > %s seconds" % (age, max_age))
         return value

django_q/queues.py

@@ -1,10 +1,9 @@
 """
 The code is derived from https://github.com/althonos/pronto/commit/3384010dfb4fc7c66a219f59276adef3288a886b
 """
-import sys
-
 import multiprocessing
 import multiprocessing.queues
+import sys
 
 
 class SharedCounter:
@@ -22,7 +21,7 @@ class SharedCounter:
     """
 
     def __init__(self, n=0):
-        self.count = multiprocessing.Value('i', n)
+        self.count = multiprocessing.Value("i", n)
 
     def increment(self, n=1):
         """ Increment the counter by n (default = 1) """
@@ -52,7 +51,9 @@ class Queue(multiprocessing.queues.Queue):
         if sys.version_info < (3, 0):
             super(Queue, self).__init__(*args, **kwargs)
         else:
-            super(Queue, self).__init__(*args, ctx=multiprocessing.get_context(), **kwargs)
+            super(Queue, self).__init__(
+                *args, ctx=multiprocessing.get_context(), **kwargs
+            )
 
         self.size = SharedCounter(0)
 
@@ -65,10 +66,10 @@ class Queue(multiprocessing.queues.Queue):
         self.size.increment(-1)
         return x
 
-    def qsize(self):
+    def qsize(self) -> int:
         """ Reliable implementation of multiprocessing.Queue.qsize() """
         return self.size.value
 
-    def empty(self):
+    def empty(self) -> bool:
         """ Reliable implementation of multiprocessing.Queue.empty() """
         return not self.qsize() > 0

django_q/signing.py

@@ -1,44 +1,39 @@
 """Package signing."""
-try:
-    import cPickle as pickle
-except ImportError:
-    import pickle
+import pickle
 
 from django_q import core_signing as signing
-
 from django_q.conf import Conf
 
 BadSignature = signing.BadSignature
 
 
 class SignedPackage:
-
     """Wraps Django's signing module with custom Pickle serializer."""
 
     @staticmethod
-    def dumps(obj, compressed=Conf.COMPRESSED):
-        return signing.dumps(obj,
-                             key=Conf.SECRET_KEY,
-                             salt=Conf.PREFIX,
-                             compress=compressed,
-                             serializer=PickleSerializer)
+    def dumps(obj, compressed: bool = Conf.COMPRESSED) -> str:
+        return signing.dumps(
+            obj,
+            key=Conf.SECRET_KEY,
+            salt=Conf.PREFIX,
+            compress=compressed,
+            serializer=PickleSerializer,
+        )
 
     @staticmethod
-    def loads(obj):
-        return signing.loads(obj,
-                             key=Conf.SECRET_KEY,
-                             salt=Conf.PREFIX,
-                             serializer=PickleSerializer)
+    def loads(obj) -> any:
+        return signing.loads(
+            obj, key=Conf.SECRET_KEY, salt=Conf.PREFIX, serializer=PickleSerializer
+        )
 
 
 class PickleSerializer:
-
     """Simple wrapper around Pickle for signing.dumps and signing.loads."""
 
     @staticmethod
-    def dumps(obj):
+    def dumps(obj) -> bytes:
         return pickle.dumps(obj, protocol=pickle.HIGHEST_PROTOCOL)
 
     @staticmethod
-    def loads(data):
+    def loads(data) -> any:
         return pickle.loads(data)

django_q/status.py

@@ -1,6 +1,9 @@
 import socket
+from typing import Union
+
 from django.utils import timezone
-from django_q.brokers import get_broker
+
+from django_q.brokers import get_broker, Broker
 from django_q.conf import Conf, logger
 from django_q.signing import SignedPackage, BadSignature
 
@@ -47,18 +50,18 @@ class Stat(Status):
             self.pusher = sentinel.pusher.pid
         self.workers = [w.pid for w in sentinel.pool]
 
-    def uptime(self):
+    def uptime(self) -> float:
         return (timezone.now() - self.tob).total_seconds()
 
     @property
-    def key(self):
+    def key(self) -> str:
         """
         :return: redis key for this cluster statistic
         """
         return self.get_key(self.cluster_id)
 
     @staticmethod
-    def get_key(cluster_id):
+    def get_key(cluster_id) -> str:
         """
         :param cluster_id: cluster ID
         :return: redis key for the cluster statistic
@@ -71,13 +74,15 @@ class Stat(Status):
         except Exception as e:
             logger.error(e)
 
-    def empty_queues(self):
+    def empty_queues(self) -> bool:
         return self.done_q_size + self.task_q_size == 0
 
     @staticmethod
-    def get(pid, cluster_id, broker=None):
+    def get(pid: int, cluster_id: str, broker: Broker = None) -> Union[Status, None]:
         """
         gets the current status for the cluster
+        :param pid:
+        :param broker: an optional broker instance
         :param cluster_id: id of the cluster
         :return: Stat or Status
         """
@@ -92,7 +97,7 @@ class Stat(Status):
         return Status(pid=pid, cluster_id=cluster_id)
 
     @staticmethod
-    def get_all(broker=None):
+    def get_all(broker: Broker = None) -> list:
         """
         Get the status for all currently running clusters with the same prefix
         and secret key.

docs/examples.rst

@@ -74,8 +74,7 @@ The task will send a message to everyone else informing them that the users emai
     def inform_everyone(user):
         mails = []
         for u in User.objects.exclude(pk=user.pk):
-            msg = 'Dear {}, {} has a new email address: {}'
-            msg = msg.format(u.username, user.username, user.email)
+            msg = f"Dear {u.username}, {user.username} has a new email address: {user.email}"
             mails.append(('New email', msg,
                           'from@example.com', [u.email]))
         return send_mass_mail(mails)
@@ -85,8 +84,7 @@ The task will send a message to everyone else informing them that the users emai
     # or do it async again
     def inform_everyone_async(user):
         for u in User.objects.exclude(pk=user.pk):
-            msg = 'Dear {}, {} has a new email address: {}'
-            msg = msg.format(u.username, user.username, user.email)
+            msg = f"Dear {u.username}, {user.username} has a new email address: {user.email}"
             async_task('django.core.mail.send_mail',
                     'New email', msg, 'from@example.com', [u.email])
 

requirements.txt

@@ -7,8 +7,8 @@
 arrow==0.15.6             # via -r requirements.in
 asgiref==3.2.7            # via django
 blessed==1.17.8           # via -r requirements.in
-boto3==1.13.26            # via -r requirements.in
-botocore==1.16.26         # via boto3, s3transfer
+boto3==1.14.1             # via -r requirements.in
+botocore==1.17.1          # via boto3, s3transfer
 certifi==2020.4.5.2       # via requests
 chardet==3.0.4            # via requests
 django-picklefield==3.0.1  # via -r requirements.in