Skip to content

Commit

Permalink
Add type annotations to most public functions
Browse files Browse the repository at this point in the history
Except for `django_rq.job`, which is too complicated to type properly.
  • Loading branch information
SpecLad committed Mar 25, 2024
1 parent 025a53f commit 208cef5
Show file tree
Hide file tree
Showing 2 changed files with 22 additions and 13 deletions.
33 changes: 21 additions & 12 deletions django_rq/queues.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,9 @@
import warnings
from typing import Any, Callable, Optional, Type, Union

import redis
from redis.sentinel import Sentinel
from rq.job import Job
from rq.queue import Queue
from rq.utils import import_attribute

Expand Down Expand Up @@ -138,7 +140,10 @@ def get_redis_connection(config, use_strict_redis=False):
)


def get_connection(name='default', use_strict_redis=False):
def get_connection(
name: str = 'default',
use_strict_redis: bool = False,
) -> redis.Redis:
"""
Returns a Redis connection to use based on parameters in settings.RQ_QUEUES
"""
Expand All @@ -148,16 +153,16 @@ def get_connection(name='default', use_strict_redis=False):


def get_queue(
name='default',
default_timeout=None,
is_async=None,
autocommit=None,
connection=None,
queue_class=None,
job_class=None,
serializer=None,
name: str = 'default',
default_timeout: Optional[int] = None,
is_async: Optional[bool] = None,
autocommit: Optional[bool] = None,
connection: Optional[redis.Redis] = None,
queue_class: Optional[Union[str, Type[DjangoRQ]]] = None,
job_class: Optional[Union[str, Type[Job]]] = None,
serializer: Any = None,
**kwargs
):
) -> DjangoRQ:
"""
Returns an rq Queue using parameters defined in ``RQ_QUEUES``
"""
Expand Down Expand Up @@ -280,7 +285,7 @@ def get_queues(*queue_names, **kwargs):
return queues


def enqueue(func, *args, **kwargs):
def enqueue(func: Callable, *args, **kwargs) -> Job:
"""
A convenience function to put a job in the default queue. Usage::
Expand Down Expand Up @@ -333,7 +338,11 @@ def _create_job(self, *args, **kwargs):

return super(DjangoScheduler, self)._create_job(*args, **kwargs)

def get_scheduler(name='default', queue=None, interval=60):
def get_scheduler(
name: str = 'default',
queue: Optional[DjangoRQ] = None,
interval: int = 60,
) -> DjangoScheduler:
"""
Returns an RQ Scheduler instance using parameters defined in
``RQ_QUEUES``
Expand Down
2 changes: 1 addition & 1 deletion django_rq/workers.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@ def get_worker_class(worker_class=None):
return worker_class


def get_worker(*queue_names, **kwargs):
def get_worker(*queue_names: str, **kwargs) -> Worker:
"""
Returns a RQ worker for all queues or specified ones.
"""
Expand Down

0 comments on commit 208cef5

Please sign in to comment.