|
1 | 1 | from rq.decorators import job as _rq_job
|
2 |
| -from typing import TYPE_CHECKING, Union |
| 2 | +from typing import Any, Callable, Optional, overload, Protocol, TYPE_CHECKING, TypeVar, Union |
3 | 3 |
|
4 | 4 | from django.conf import settings
|
5 | 5 |
|
6 | 6 | from .queues import get_queue
|
7 | 7 |
|
8 | 8 | if TYPE_CHECKING:
|
| 9 | + from redis import Redis |
9 | 10 | from rq import Queue
|
| 11 | + from typing_extensions import ParamSpec |
10 | 12 |
|
| 13 | + P = ParamSpec('P') |
| 14 | + R = TypeVar('R', covariant=True) |
11 | 15 |
|
12 |
| -def job(func_or_queue, connection=None, *args, **kwargs): |
| 16 | + class _JobFn(Protocol[P, R]): |
| 17 | + def delay(self, *args: P.args, **kwargs: P.kwargs) -> R: ... |
| 18 | + def enqueue(self, *args: P.args, **kwargs: P.kwargs) -> R: ... |
| 19 | + def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R: ... |
| 20 | + |
| 21 | + |
| 22 | +@overload |
| 23 | +def job(func_or_queue: 'Callable[P, R]') -> '_JobFn[P, R]': ... |
| 24 | + |
| 25 | +@overload |
| 26 | +def job( |
| 27 | + func_or_queue: Union['Queue', str], |
| 28 | + connection: Optional['Redis'] = None, |
| 29 | + *args: Any, |
| 30 | + **kwargs: Any, |
| 31 | +) -> Callable[['Callable[P, R]'], '_JobFn[P, R]']: ... |
| 32 | + |
| 33 | + |
| 34 | +def job( |
| 35 | + func_or_queue: Union['Callable[P, R]', 'Queue', str], |
| 36 | + connection: Optional['Redis'] = None, |
| 37 | + *args: Any, |
| 38 | + **kwargs: Any, |
| 39 | +) -> Union['_JobFn[P, R]', Callable[['Callable[P, R]'], '_JobFn[P, R]']]: |
13 | 40 | """
|
14 | 41 | The same as RQ's job decorator, but it automatically works out
|
15 | 42 | the ``connection`` argument from RQ_QUEUES.
|
|
0 commit comments