Merge branch 'types'

Author: PeterJCLaw

.circleci/config.yml

@@ -80,6 +80,23 @@ jobs:
             source $HOME/.poetry/env
             poetry run flake8
 
+  type-check:
+    <<: *parametrised-python-executor
+
+    steps:
+      - checkout
+
+      - *install-poetry
+      - *restore-dependencies-cache
+      - *install-dependencies
+      - *save-dependencies-cache
+
+      - run:
+          name: Run Mypy
+          command: |
+            source $HOME/.poetry/env
+            poetry run ./script/type-check
+
   deploy:
     executor:
       name: python
@@ -108,6 +125,9 @@ workflows:
       - lint:
           name: lint-py<< matrix.python-version >>-Django<< matrix.django-version >>
           <<: *version-matrix
+      - type-check:
+          name: type-check-py<< matrix.python-version >>-Django<< matrix.django-version >>
+          <<: *version-matrix
 
 
   build-test-deploy:
@@ -130,11 +150,21 @@ workflows:
             branches:
               ignore: /.*/
 
+      - type-check:
+          name: type-check-py<< matrix.python-version >>-Django<< matrix.django-version >>
+          <<: *version-matrix
+          filters:
+            tags:
+              only: /v[0-9]+(\.[0-9]+)*/
+            branches:
+              ignore: /.*/
+
       - deploy:
           context: thread-pypi
           requires:
             - build-test
             - lint
+            - type-check
           filters:
               tags:
                 only: /v[0-9]+(\.[0-9]+)*/

django_lightweight_queue/app_settings.py

@@ -1,40 +1,51 @@
+from typing import Dict, Union, Mapping, TypeVar, Callable, Sequence
+
 from django.conf import settings
 
 from . import constants
+from .types import Logger, QueueName
+
+T = TypeVar('T')
 
 
-def setting(suffix, default):
+def setting(suffix: str, default: T) -> T:
     attr_name = '{}{}'.format(constants.SETTING_NAME_PREFIX, suffix)
     return getattr(settings, attr_name, default)
 
 
-WORKERS = setting('WORKERS', {})
-BACKEND = setting('BACKEND', 'django_lightweight_queue.backends.synchronous.SynchronousBackend')
+WORKERS = setting('WORKERS', {})  # type: Dict[QueueName, int]
+BACKEND = setting(
+    'BACKEND',
+    'django_lightweight_queue.backends.synchronous.SynchronousBackend',
+)  # type: str
 
-LOGGER_FACTORY = setting('LOGGER_FACTORY', 'logging.getLogger')
+LOGGER_FACTORY = setting(
+    'LOGGER_FACTORY',
+    'logging.getLogger',
+)  # type: Union[str, Callable[[str], Logger]]
 
 # Allow per-queue overrides of the backend.
-BACKEND_OVERRIDES = setting('BACKEND_OVERRIDES', {})
+BACKEND_OVERRIDES = setting('BACKEND_OVERRIDES', {})  # type: Mapping[QueueName, str]
 
 MIDDLEWARE = setting('MIDDLEWARE', (
     'django_lightweight_queue.middleware.logging.LoggingMiddleware',
     'django_lightweight_queue.middleware.transaction.TransactionMiddleware',
-))
+))  # type: Sequence[str]
 
 # Apps to ignore when looking for tasks. Apps must be specified as the dotted
 # name used in `INSTALLED_APPS`. This is expected to be useful when you need to
 # have a file called `tasks.py` within an app, but don't want
 # django-lightweight-queue to import that file.
 # Note: this _doesn't_ prevent tasks being registered from these apps.
-IGNORE_APPS = setting('IGNORE_APPS', ())
+IGNORE_APPS = setting('IGNORE_APPS', ())  # type: Sequence[str]
 
 # Backend-specific settings
-REDIS_HOST = setting('REDIS_HOST', '127.0.0.1')
-REDIS_PORT = setting('REDIS_PORT', 6379)
-REDIS_PREFIX = setting('REDIS_PREFIX', '')
+REDIS_HOST = setting('REDIS_HOST', '127.0.0.1')  # type: str
+REDIS_PORT = setting('REDIS_PORT', 6379)  # type: int
+REDIS_PREFIX = setting('REDIS_PREFIX', '')  # type: str
 
-ENABLE_PROMETHEUS = setting('ENABLE_PROMETHEUS', False)
+ENABLE_PROMETHEUS = setting('ENABLE_PROMETHEUS', False)  # type: bool
 # Workers will export metrics on this port, and ports following it
-PROMETHEUS_START_PORT = setting('PROMETHEUS_START_PORT', 9300)
+PROMETHEUS_START_PORT = setting('PROMETHEUS_START_PORT', 9300)  # type: int
 
-ATOMIC_JOBS = setting('ATOMIC_JOBS', True)
+ATOMIC_JOBS = setting('ATOMIC_JOBS', True)  # type: bool

django_lightweight_queue/apps.py

@@ -6,5 +6,5 @@
 class DjangoLightweightQueueConfig(AppConfig):
     name = 'django_lightweight_queue'
 
-    def ready(self):
+    def ready(self) -> None:
         load_all_tasks()

django_lightweight_queue/backends/base.py

@@ -1,23 +1,41 @@
 from abc import ABCMeta, abstractmethod
+from typing import Tuple, TypeVar, Optional
 
 from ..job import Job
+from ..types import QueueName, WorkerNumber
+from ..progress_logger import ProgressLogger, NULL_PROGRESS_LOGGER
+
+# Work around https://github.com/python/mypy/issues/9914. Name needs to match
+# that in progress_logger.py.
+T = TypeVar('T')
 
 
 class BaseBackend(metaclass=ABCMeta):
-    def startup(self, queue: str) -> None:
+    def startup(self, queue: QueueName) -> None:
         pass
 
     @abstractmethod
-    def enqueue(self, job: Job, queue: str) -> None:
+    def enqueue(self, job: Job, queue: QueueName) -> None:
         raise NotImplementedError()
 
     @abstractmethod
-    def dequeue(self, queue: str, worker_num: int, timeout: float) -> Job:
+    def dequeue(self, queue: QueueName, worker_num: WorkerNumber, timeout: int) -> Optional[Job]:
         raise NotImplementedError()
 
     @abstractmethod
-    def length(self, queue: str) -> int:
+    def length(self, queue: QueueName) -> int:
         raise NotImplementedError()
 
-    def processed_job(self, queue: str, worker_num: int, job: Job) -> None:
+    def processed_job(self, queue: QueueName, worker_num: WorkerNumber, job: Job) -> None:
         pass
+
+
+class BackendWithDeduplicate(BaseBackend, metaclass=ABCMeta):
+    @abstractmethod
+    def deduplicate(
+        self,
+        queue: QueueName,
+        *,
+        progress_logger: ProgressLogger = NULL_PROGRESS_LOGGER
+    ) -> Tuple[int, int]:
+        raise NotImplementedError()

django_lightweight_queue/backends/debug_web.py

@@ -1,9 +1,11 @@
-import urllib
+import urllib.parse
 
 from django.conf import settings
 from django.shortcuts import reverse
 
+from ..job import Job
 from .base import BaseBackend
+from ..types import QueueName, WorkerNumber
 
 
 class DebugWebBackend(BaseBackend):
@@ -17,14 +19,14 @@ class DebugWebBackend(BaseBackend):
     See the docstring of that view for information (and limitations) about it.
     """
 
-    def enqueue(self, job, queue):
+    def enqueue(self, job: Job, queue: QueueName) -> None:
         path = reverse('django-lightweight-queue:debug-run')
         query_string = urllib.parse.urlencode({'job': job.to_json()})
         url = "{}{}?{}".format(settings.SITE_URL, path, query_string)
         print(url)
 
-    def dequeue(self, queue, worker_num, timeout):
+    def dequeue(self, queue: QueueName, worker_num: WorkerNumber, timeout: float) -> None:
         pass
 
-    def length(self, queue):
+    def length(self, queue: QueueName) -> int:
         return 0

django_lightweight_queue/backends/redis.py

@@ -1,36 +1,39 @@
+from typing import Optional
+
 import redis
 
 from .. import app_settings
 from ..job import Job
 from .base import BaseBackend
+from ..types import QueueName, WorkerNumber
 
 
 class RedisBackend(BaseBackend):
     """
     This backend has at-most-once semantics.
     """
 
-    def __init__(self):
+    def __init__(self) -> None:
         self.client = redis.StrictRedis(
             host=app_settings.REDIS_HOST,
             port=app_settings.REDIS_PORT,
         )
 
-    def enqueue(self, job, queue):
+    def enqueue(self, job: Job, queue: QueueName) -> None:
         self.client.lpush(self._key(queue), job.to_json().encode('utf-8'))
 
-    def dequeue(self, queue, worker_num, timeout):
-        try:
-            _, data = self.client.brpop(self._key(queue), timeout)
+    def dequeue(self, queue: QueueName, worker_num: WorkerNumber, timeout: int) -> Optional[Job]:
+        raw = self.client.brpop(self._key(queue), timeout)
+        if raw is None:
+            return None
 
-            return Job.from_json(data.decode('utf-8'))
-        except TypeError:
-            pass
+        _, data = raw
+        return Job.from_json(data.decode('utf-8'))
 
-    def length(self, queue):
+    def length(self, queue: QueueName) -> int:
         return self.client.llen(self._key(queue))
 
-    def _key(self, queue):
+    def _key(self, queue: QueueName) -> str:
         if app_settings.REDIS_PREFIX:
             return '{}:django_lightweight_queue:{}'.format(
                 app_settings.REDIS_PREFIX,

django_lightweight_queue/backends/reliable_redis.py

@@ -1,13 +1,20 @@
+from typing import Dict, List, Tuple, TypeVar, Optional
+
 import redis
 
 from .. import app_settings
 from ..job import Job
-from .base import BaseBackend
+from .base import BackendWithDeduplicate
+from ..types import QueueName, WorkerNumber
 from ..utils import get_worker_numbers
-from ..progress_logger import NULL_PROGRESS_LOGGER
+from ..progress_logger import ProgressLogger, NULL_PROGRESS_LOGGER
+
+# Work around https://github.com/python/mypy/issues/9914. Name needs to match
+# that in progress_logger.py.
+T = TypeVar('T')
 
 
-class ReliableRedisBackend(BaseBackend):
+class ReliableRedisBackend(BackendWithDeduplicate):
     """
     This backend manages a per-queue-per-worker 'processing' queue. E.g. if we
     had a queue called 'django_lightweight_queue:things', and two workers, we
@@ -29,13 +36,13 @@ class ReliableRedisBackend(BaseBackend):
     This backend has at-least-once semantics.
     """
 
-    def __init__(self):
+    def __init__(self) -> None:
         self.client = redis.StrictRedis(
             host=app_settings.REDIS_HOST,
             port=app_settings.REDIS_PORT,
         )
 
-    def startup(self, queue):
+    def startup(self, queue: QueueName) -> None:
         main_queue_key = self._key(queue)
 
         pattern = self._prefix_key(
@@ -53,7 +60,7 @@ def startup(self, queue):
         )
         processing_queue_keys = current_processing_queue_keys - expected_processing_queue_keys
 
-        def move_processing_jobs_to_main(pipe):
+        def move_processing_jobs_to_main(pipe: redis.client.Pipeline) -> None:
             # Collect all the data we need to add, before adding the data back
             # to the main queue of and clearing the processing queues
             # atomically, so if this crashes, we don't lose jobs
@@ -79,10 +86,10 @@ def move_processing_jobs_to_main(pipe):
             *processing_queue_keys,
         )
 
-    def enqueue(self, job, queue):
+    def enqueue(self, job: Job, queue: QueueName) -> None:
         self.client.lpush(self._key(queue), job.to_json().encode('utf-8'))
 
-    def dequeue(self, queue, worker_number, timeout):
+    def dequeue(self, queue: QueueName, worker_number: WorkerNumber, timeout: int) -> Optional[Job]:
         main_queue_key = self._key(queue)
         processing_queue_key = self._processing_key(queue, worker_number)
 
@@ -104,7 +111,9 @@ def dequeue(self, queue, worker_number, timeout):
         if data:
             return Job.from_json(data.decode('utf-8'))
 
-    def processed_job(self, queue, worker_number, job):
+        return None
+
+    def processed_job(self, queue: QueueName, worker_number: WorkerNumber, job: Job) -> None:
         data = job.to_json().encode('utf-8')
 
         self.client.lrem(
@@ -113,10 +122,15 @@ def processed_job(self, queue, worker_number, job):
             value=data,
         )
 
-    def length(self, queue):
+    def length(self, queue: QueueName) -> int:
         return self.client.llen(self._key(queue))
 
-    def deduplicate(self, queue, *, progress_logger=NULL_PROGRESS_LOGGER):
+    def deduplicate(
+        self,
+        queue: QueueName,
+        *,
+        progress_logger: ProgressLogger = NULL_PROGRESS_LOGGER
+    ) -> Tuple[int, int]:
         """
         Deduplicate the given queue by comparing the jobs in a manner which
         ignores their created timestamps.
@@ -137,7 +151,7 @@ def deduplicate(self, queue, *, progress_logger=NULL_PROGRESS_LOGGER):
 
         # A mapping of job_identity -> list of raw_job data; the entries in the
         # latter list are ordered from newest to oldest
-        jobs = {}
+        jobs = {}  # type: Dict[str, List[bytes]]
 
         progress_logger.info("Collecting jobs")
 
@@ -163,20 +177,20 @@ def deduplicate(self, queue, *, progress_logger=NULL_PROGRESS_LOGGER):
 
         return original_size, self.client.llen(main_queue_key)
 
-    def _key(self, queue):
+    def _key(self, queue: QueueName) -> str:
         key = 'django_lightweight_queue:{}'.format(queue)
 
         return self._prefix_key(key)
 
-    def _processing_key(self, queue, worker_number):
+    def _processing_key(self, queue: QueueName, worker_number: WorkerNumber) -> str:
         key = 'django_lightweight_queue:{}:processing:{}'.format(
             queue,
             worker_number,
         )
 
         return self._prefix_key(key)
 
-    def _prefix_key(self, key):
+    def _prefix_key(self, key: str) -> str:
         if app_settings.REDIS_PREFIX:
             return '{}:{}'.format(
                 app_settings.REDIS_PREFIX,