test type annotations with mypy

.github/workflows/test.yml

@@ -39,3 +39,24 @@ jobs:
     - name: Run Test
       run: |
         `which django-admin` test django_rq --settings=django_rq.tests.settings --pythonpath=.
+
+  mypy:
+    runs-on: ubuntu-latest
+    name: Type check
+
+    steps:
+    - uses: actions/checkout@v3
+
+    - name: Set up Python 3.8
+      uses: actions/[email protected]
+      with:
+        python-version: "3.8"
+
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install django-stubs[compatible-mypy] rq types-redis
+
+    - name: Run Test
+      run: |
+        mypy django_rq

django_rq/admin.py

@@ -1,4 +1,8 @@
+from typing import Any, Dict, Optional
+
 from django.contrib import admin
+from django.http.request import HttpRequest
+from django.http.response import HttpResponse
 
 from . import views, settings, models
 
@@ -9,10 +13,10 @@ class QueueAdmin(admin.ModelAdmin):
     def has_add_permission(self, request):
         return False  # Hide the admin "+ Add" link for Queues
 
-    def has_change_permission(self, request):
+    def has_change_permission(self, request: HttpRequest, obj: Optional[Any] = None) -> bool:
         return True
 
-    def has_module_permission(self, request):
+    def has_module_permission(self, request: HttpRequest):
         """
         return True if the given request has any permission in the given
         app label.
@@ -23,9 +27,9 @@ def has_module_permission(self, request):
         does not restrict access to the add, change or delete views. Use
         `ModelAdmin.has_(add|change|delete)_permission` for that.
         """
-        return request.user.has_module_perms('django_rq')
+        return request.user.has_module_perms('django_rq')  # type: ignore[union-attr]
 
-    def changelist_view(self, request):
+    def changelist_view(self, request: HttpRequest, extra_context: Optional[Dict[str, Any]] = None) -> HttpResponse:
         """The 'change list' admin view for this model."""
         # proxy request to stats view
         return views.stats(request)

django_rq/decorators.py

@@ -1,9 +1,13 @@
 from rq.decorators import job as _rq_job
+from typing import TYPE_CHECKING, Union
 
 from django.conf import settings
 
 from .queues import get_queue
 
+if TYPE_CHECKING:
+    from rq import Queue
+
 
 def job(func_or_queue, connection=None, *args, **kwargs):
     """
@@ -18,7 +22,7 @@ def job(func_or_queue, connection=None, *args, **kwargs):
     """
     if callable(func_or_queue):
         func = func_or_queue
-        queue = 'default'
+        queue: Union['Queue', str] = 'default'
     else:
         func = None
         queue = func_or_queue
@@ -30,13 +34,17 @@ def job(func_or_queue, connection=None, *args, **kwargs):
                 connection = queue.connection
         except KeyError:
             pass
+    else:
+        if connection is None:
+            connection = queue.connection
 
     RQ = getattr(settings, 'RQ', {})
     default_result_ttl = RQ.get('DEFAULT_RESULT_TTL')
     if default_result_ttl is not None:
         kwargs.setdefault('result_ttl', default_result_ttl)
 
-    decorator = _rq_job(queue, connection=connection, *args, **kwargs)
+    kwargs['connection'] = connection
+    decorator = _rq_job(queue, *args, **kwargs)
     if func:
         return decorator(func)
     return decorator

django_rq/management/commands/rqenqueue.py

@@ -22,9 +22,7 @@ def handle(self, *args, **options):
         Queues the function given with the first argument with the
         parameters given with the rest of the argument list.
         """
-        verbosity = int(options.get('verbosity', 1))
-        timeout = options.get('timeout')
-        queue = get_queue(options.get('queue'))
-        job = queue.enqueue_call(args[0], args=args[1:], timeout=timeout)
-        if verbosity:
+        queue = get_queue(options['queue'])
+        job = queue.enqueue_call(args[0], args=args[1:], timeout=options['timeout'])
+        if options['verbosity']:
             print('Job %s created' % job.id)

django_rq/management/commands/rqscheduler.py

@@ -36,7 +36,7 @@ def handle(self, *args, **options):
                 fp.write(str(os.getpid()))
 
         # Verbosity is defined by default in BaseCommand for all commands
-        verbosity = options.get('verbosity')
+        verbosity: int = options['verbosity']
         if verbosity >= 2:
             level = 'DEBUG'
         elif verbosity == 0:
@@ -46,5 +46,5 @@ def handle(self, *args, **options):
         setup_loghandlers(level)
 
         scheduler = get_scheduler(
-            name=options.get('queue'), interval=options.get('interval'))
+            name=options['queue'], interval=options['interval'])
         scheduler.run()

django_rq/management/commands/rqstats.py

@@ -11,6 +11,7 @@ class Command(BaseCommand):
     Print RQ statistics
     """
     help = __doc__
+    _separator: str
 
     def add_arguments(self, parser):
         # TODO: convert this to @click.command like rq does
@@ -89,7 +90,7 @@ def handle(self, *args, **options):
                 raise CommandError("PyYAML is not installed.") from ex
 
             # Disable YAML alias
-            yaml.Dumper.ignore_aliases = lambda *args: True
+            yaml.Dumper.ignore_aliases = lambda *args: True  # type: ignore[method-assign]
             click.echo(yaml.dump(get_statistics(), default_flow_style=False))
             return
 

django_rq/management/commands/rqworker-pool.py

@@ -4,6 +4,7 @@
 from rq.serializers import resolve_serializer
 from rq.worker_pool import WorkerPool
 from rq.logutils import setup_loghandlers
+from typing import cast
 
 from django.core.management.base import BaseCommand
 
@@ -67,7 +68,7 @@ def handle(self, *args, **options):
                 fp.write(str(os.getpid()))
 
         # Verbosity is defined by default in BaseCommand for all commands
-        verbosity = options.get('verbosity')
+        verbosity: int = options['verbosity']
         if verbosity >= 2:
             logging_level = 'DEBUG'
         elif verbosity == 0:

django_rq/management/commands/rqworker.py

@@ -60,7 +60,7 @@ def handle(self, *args, **options):
                 fp.write(str(os.getpid()))
 
         # Verbosity is defined by default in BaseCommand for all commands
-        verbosity = options.get('verbosity')
+        verbosity = options['verbosity']
         if verbosity >= 2:
             level = 'DEBUG'
         elif verbosity == 0:

django_rq/queues.py

@@ -105,10 +105,10 @@ def get_redis_connection(config, use_strict_redis=False):
         cache = caches[config['USE_REDIS_CACHE']]
         # We're using django-redis-cache
         try:
-            return cache._client
+            return cache._client  # type: ignore[attr-defined]
         except AttributeError:
             # For django-redis-cache > 0.13.1
-            return cache.get_master_client()
+            return cache.get_master_client()  # type: ignore[attr-defined]
 
     if 'UNIX_SOCKET_PATH' in config:
         return redis_cls(unix_socket_path=config['UNIX_SOCKET_PATH'], db=config['DB'])
@@ -161,7 +161,7 @@ def get_queue(
     queue_class: Optional[Union[str, Type[DjangoRQ]]] = None,
     job_class: Optional[Union[str, Type[Job]]] = None,
     serializer: Any = None,
-    **kwargs
+    **kwargs: Any,
 ) -> DjangoRQ:
     """
     Returns an rq Queue using parameters defined in ``RQ_QUEUES``
@@ -366,5 +366,5 @@ def get_scheduler(
 
 except ImportError:
 
-    def get_scheduler(*args, **kwargs):
+    def get_scheduler(*args, **kwargs):  # type: ignore[misc]
         raise ImproperlyConfigured('rq_scheduler not installed')

django_rq/settings.py

@@ -1,4 +1,5 @@
 from operator import itemgetter
+from typing import Any, cast, Dict, List, Optional
 
 from django.conf import settings
 from django.core.exceptions import ImproperlyConfigured
@@ -7,11 +8,11 @@
 
 SHOW_ADMIN_LINK = getattr(settings, 'RQ_SHOW_ADMIN_LINK', False)
 
-QUEUES = getattr(settings, 'RQ_QUEUES', None)
+QUEUES = cast(Dict[str, Any], getattr(settings, 'RQ_QUEUES', None))
 if QUEUES is None:
     raise ImproperlyConfigured("You have to define RQ_QUEUES in settings.py")
 NAME = getattr(settings, 'RQ_NAME', 'default')
-BURST = getattr(settings, 'RQ_BURST', False)
+BURST: bool = getattr(settings, 'RQ_BURST', False)
 
 # All queues in list format so we can get them by index, includes failed queues
 QUEUES_LIST = []
@@ -21,7 +22,7 @@
     QUEUES_MAP[key] = len(QUEUES_LIST) - 1
 
 # Get exception handlers
-EXCEPTION_HANDLERS = getattr(settings, 'RQ_EXCEPTION_HANDLERS', [])
+EXCEPTION_HANDLERS: List[str] = getattr(settings, 'RQ_EXCEPTION_HANDLERS', [])
 
 # Token for querying statistics
-API_TOKEN = getattr(settings, 'RQ_API_TOKEN', '')
+API_TOKEN: str = getattr(settings, 'RQ_API_TOKEN', '')

django_rq/tests/fixtures.py

@@ -28,7 +28,9 @@ class DummyScheduler(Scheduler):
 
 
 def access_self():
-    return get_current_job().id
+    job = get_current_job()
+    assert job
+    return job.id
 
 
 def failing_job():

django_rq/tests/settings.py

@@ -20,13 +20,6 @@
     except ImportError:
         REDIS_CACHE_TYPE = 'none'
 
-try:
-    from django.utils.log import NullHandler
-
-    nullhandler = 'django.utils.log.NullHandler'
-except:
-    nullhandler = 'logging.NullHandler'
-
 INSTALLED_APPS = [
     'django.contrib.contenttypes',
     'django.contrib.admin',
@@ -86,7 +79,7 @@
         },
         'null': {
             'level': 'DEBUG',
-            'class': nullhandler,
+            'class': 'logging.NullHandler',
         },
     },
     'loggers': {

django_rq/tests/test_views.py

@@ -69,6 +69,7 @@ def test_job_details_with_results(self):
         result = job.results()[0]
         url = reverse('rq_job_detail', args=[queue_index, job.id])
         response = self.client.get(url)
+        assert result.id
         self.assertContains(response, result.id)
 
     def test_job_details_on_deleted_dependency(self):
@@ -185,6 +186,7 @@ def test_enqueue_jobs(self):
 
         # Check that job is updated correctly
         last_job = queue.fetch_job(last_job.id)
+        assert last_job
         self.assertEqual(last_job.get_status(), JobStatus.QUEUED)
         self.assertIsNotNone(last_job.enqueued_at)
 
@@ -407,7 +409,7 @@ def test_action_stop_jobs(self):
         self.assertEqual(len(canceled_job_registry), len(job_ids))
 
         for job_id in job_ids:
-            self.assertIn(job_id, canceled_job_registry)
+            self.assertIn(job_id, canceled_job_registry)  # type: ignore[arg-type]
 
     def test_scheduler_jobs(self):
         # Override testing RQ_QUEUES

django_rq/tests/tests.py

@@ -1,6 +1,7 @@
 import sys
 import datetime
 import time
+from typing import Any, cast, Dict, List
 from unittest import skipIf, mock
 from unittest.mock import patch, PropertyMock, MagicMock
 from uuid import uuid4
@@ -34,7 +35,7 @@
 )
 from django_rq import thread_queue
 from django_rq.templatetags.django_rq import force_escape, to_localtime
-from django_rq.tests.fixtures import DummyJob, DummyQueue, DummyWorker
+from django_rq.tests.fixtures import access_self, DummyJob, DummyQueue, DummyWorker
 from django_rq.utils import get_jobs, get_statistics, get_scheduler_pid
 from django_rq.workers import get_worker, get_worker_class
 
@@ -50,10 +51,6 @@
 QUEUES = settings.RQ_QUEUES
 
 
-def access_self():
-    return get_current_job().id
-
-
 def divide(a, b):
     return a / b
 
@@ -271,7 +268,7 @@ def test_pass_queue_via_commandline_args(self):
         Checks that passing queues via commandline arguments works
         """
         queue_names = ['django_rq_test', 'django_rq_test2']
-        jobs = []
+        jobs: List[Any] = []
         for queue_name in queue_names:
             queue = get_queue(queue_name)
             jobs.append(
@@ -288,7 +285,7 @@ def test_pass_queue_via_commandline_args(self):
             self.assertIn(job['job'].id, job['finished_job_registry'].get_job_ids())
 
         # Test with rqworker-pool command
-        jobs = []
+        jobs: List[Any] = []
         for queue_name in queue_names:
             queue = get_queue(queue_name)
             jobs.append(
@@ -441,8 +438,7 @@ def test_async(self):
         # Make sure old keyword argument 'async' works for backwards
         # compatibility with code expecting older versions of rq or django-rq.
         # Note 'async' is a reserved keyword in Python >= 3.7.
-        kwargs = {'async': False}
-        default_queue_async = get_queue('default', **kwargs)
+        default_queue_async = get_queue('default', **cast(Dict[str, Any], {'async': False}))
         self.assertFalse(default_queue_async._is_async)
 
         # Make sure is_async setting works
@@ -722,7 +718,7 @@ def test_commandline_verbosity_affects_logging_level(self, setup_loghandlers_moc
             setup_loghandlers_mock.assert_called_once_with(expected_level[verbosity])
 
     @override_settings(RQ={'SCHEDULER_CLASS': 'django_rq.tests.fixtures.DummyScheduler'})
-    def test_scheduler_default_timeout(self):
+    def test_scheduler_default(self):
         """
         Scheduler class customization.
         """

django_rq/tests/urls.py

@@ -1,13 +1,13 @@
 from django.contrib import admin
 from django.urls import path
 
-from django_rq.urls import urlpatterns
+from django_rq.urls import urlpatterns as django_rq_urlpatterns
 
 from . import views
 
 urlpatterns = [
     path('admin/', admin.site.urls),
     path('success/', views.success, name='success'),
     path('error/', views.error, name='error'),
-    path('django-rq/', (urlpatterns, '', 'django_rq')),
+    path('django-rq/', (django_rq_urlpatterns, '', 'django_rq')),
 ]