test type annotations with mypy (#668)

Author: terencehonles

.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/py.typed

@@ -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')