Skip to content

Commit

Permalink
add typing for @job decorator (#669)
Browse files Browse the repository at this point in the history
  • Loading branch information
terencehonles authored Oct 19, 2024
1 parent ec5cfef commit a953e6a
Showing 1 changed file with 29 additions and 2 deletions.
31 changes: 29 additions & 2 deletions django_rq/decorators.py
Original file line number Diff line number Diff line change
@@ -1,15 +1,42 @@
from rq.decorators import job as _rq_job
from typing import TYPE_CHECKING, Union
from typing import Any, Callable, Optional, overload, Protocol, TYPE_CHECKING, TypeVar, Union

from django.conf import settings

from .queues import get_queue

if TYPE_CHECKING:
from redis import Redis
from rq import Queue
from typing_extensions import ParamSpec

P = ParamSpec('P')
R = TypeVar('R', covariant=True)

def job(func_or_queue, connection=None, *args, **kwargs):
class _JobFn(Protocol[P, R]):
def delay(self, *args: P.args, **kwargs: P.kwargs) -> R: ...
def enqueue(self, *args: P.args, **kwargs: P.kwargs) -> R: ...
def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R: ...


@overload
def job(func_or_queue: 'Callable[P, R]') -> '_JobFn[P, R]': ...

@overload
def job(
func_or_queue: Union['Queue', str],
connection: Optional['Redis'] = None,
*args: Any,
**kwargs: Any,
) -> Callable[['Callable[P, R]'], '_JobFn[P, R]']: ...


def job(
func_or_queue: Union['Callable[P, R]', 'Queue', str],
connection: Optional['Redis'] = None,
*args: Any,
**kwargs: Any,
) -> Union['_JobFn[P, R]', Callable[['Callable[P, R]'], '_JobFn[P, R]']]:
"""
The same as RQ's job decorator, but it automatically works out
the ``connection`` argument from RQ_QUEUES.
Expand Down

0 comments on commit a953e6a

Please sign in to comment.