diff --git a/.github/release-drafter.yml b/.github/release-drafter.yml new file mode 100755 index 0000000..8d3df05 --- /dev/null +++ b/.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 \ No newline at end of file diff --git a/.github/workflows/release_log.yml b/.github/workflows/release_log.yml new file mode 100644 index 0000000..8c62eee --- /dev/null +++ b/.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 }} \ No newline at end of file diff --git a/django_q/brokers/__init__.py b/django_q/brokers/__init__.py index 167ff62..5bef988 100644 --- a/django_q/brokers/__init__.py +++ b/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) diff --git a/django_q/brokers/aws_sqs.py b/django_q/brokers/aws_sqs.py index 535259a..1a6deda 100644 --- a/django_q/brokers/aws_sqs.py +++ b/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) diff --git a/django_q/brokers/disque.py b/django_q/brokers/disque.py index ce8f7ed..09c6c4b 100644 --- a/django_q/brokers/disque.py +++ b/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 diff --git a/django_q/brokers/ironmq.py b/django_q/brokers/ironmq.py index bf14dd8..fb148d6 100644 --- a/django_q/brokers/ironmq.py +++ b/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) diff --git a/django_q/brokers/mongo.py b/django_q/brokers/mongo.py index c1b4e41..a130113 100644 --- a/django_q/brokers/mongo.py +++ b/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 diff --git a/django_q/brokers/orm.py b/django_q/brokers/orm.py index da5b289..345eff1 100644 --- a/django_q/brokers/orm.py +++ b/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 = [] diff --git a/django_q/brokers/redis_broker.py b/django_q/brokers/redis_broker.py index f224aa7..eafda41 100644 --- a/django_q/brokers/redis_broker.py +++ b/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): diff --git a/django_q/cluster.py b/django_q/cluster.py index e9dac34..1b3c993 100644 --- a/django_q/cluster.py +++ b/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. diff --git a/django_q/conf.py b/django_q/conf.py index 3c0b6a1..52f2261 100644 --- a/django_q/conf.py +++ b/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)) diff --git a/django_q/core_signing.py b/django_q/core_signing.py index 6674df0..e2ceb9d 100644 --- a/django_q/core_signing.py +++ b/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 diff --git a/django_q/queues.py b/django_q/queues.py index 978b068..9844fe2 100644 --- a/django_q/queues.py +++ b/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 diff --git a/django_q/signing.py b/django_q/signing.py index 474d0cf..3c89fab 100644 --- a/django_q/signing.py +++ b/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) diff --git a/django_q/status.py b/django_q/status.py index 7f0fa8f..00eefa9 100644 --- a/django_q/status.py +++ b/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. diff --git a/docs/examples.rst b/docs/examples.rst index 7d7b5a1..62b0811 100644 --- a/docs/examples.rst +++ b/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]) diff --git a/requirements.txt b/requirements.txt index 6c7da3a..3c4a993 100644 --- a/requirements.txt +++ b/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