diff --git a/tasktiger/_internal.py b/tasktiger/_internal.py index 01c913e..dc22bb6 100644 --- a/tasktiger/_internal.py +++ b/tasktiger/_internal.py @@ -15,13 +15,13 @@ Iterable, List, Optional, - Tuple, Type, TypedDict, Union, ) from .exceptions import TaskImportError +from .types import SerializedRetryStrategy if TYPE_CHECKING: from .task import Task @@ -134,7 +134,7 @@ def reversed_dotted_parts(s: str) -> Iterable[str]: yield s[:idx] -def serialize_retry_method(retry_method: Any) -> Tuple[str, Tuple]: +def serialize_retry_method(retry_method: Any) -> SerializedRetryStrategy: if callable(retry_method): return (serialize_func_name(retry_method), ()) else: diff --git a/tasktiger/schedule.py b/tasktiger/schedule.py index 35aa85f..dae11bb 100644 --- a/tasktiger/schedule.py +++ b/tasktiger/schedule.py @@ -1,5 +1,7 @@ import datetime -from typing import Callable, Optional, Tuple +from typing import Optional + +from .types import ScheduleStrategy __all__ = ["periodic", "cron_expr"] @@ -40,7 +42,7 @@ def periodic( weeks: int = 0, start_date: Optional[datetime.datetime] = None, end_date: Optional[datetime.datetime] = None, -) -> Tuple[Callable[..., Optional[datetime.datetime]], Tuple]: +) -> ScheduleStrategy: """ Periodic task schedule: Use to schedule a task to run periodically, starting from start_date (or None to be active immediately) until end_date @@ -94,7 +96,7 @@ def cron_expr( expr: str, start_date: Optional[datetime.datetime] = None, end_date: Optional[datetime.datetime] = None, -) -> Tuple[Callable[..., Optional[datetime.datetime]], Tuple]: +) -> ScheduleStrategy: """ Periodic task schedule via cron expression: Use to schedule a task to run periodically, starting from start_date (or None to be active immediately) until end_date diff --git a/tasktiger/task.py b/tasktiger/task.py index d5b5bb5..be4de25 100644 --- a/tasktiger/task.py +++ b/tasktiger/task.py @@ -32,7 +32,7 @@ ) from .exceptions import QueueFullException, TaskImportError, TaskNotFound from .runner import BaseRunner, get_runner_class -from .types import RetryStrategy +from .types import RetryStrategy, SerializedRetryStrategy if TYPE_CHECKING: from . import TaskTiger @@ -55,9 +55,7 @@ def __init__( lock_key: Optional[Collection[str]] = None, retry: Optional[bool] = None, retry_on: Optional[Collection[Type[BaseException]]] = None, - retry_method: Optional[ - Union[Callable[[int], float], Tuple[Callable[..., float], Tuple]] - ] = None, + retry_method: Optional[Union[Callable[[int], float], RetryStrategy]] = None, max_queue_size: Optional[int] = None, max_stored_executions: Optional[int] = None, runner_class: Optional[Type["BaseRunner"]] = None, @@ -237,7 +235,7 @@ def unique_key(self) -> Optional[str]: return self._data.get("unique_key") @property - def retry_method(self) -> Optional[RetryStrategy]: + def retry_method(self) -> Optional[SerializedRetryStrategy]: if "retry_method" in self._data: retry_func, retry_args = self._data["retry_method"] return retry_func, retry_args diff --git a/tasktiger/tasktiger.py b/tasktiger/tasktiger.py index aeb611b..4820f3b 100644 --- a/tasktiger/tasktiger.py +++ b/tasktiger/tasktiger.py @@ -36,6 +36,7 @@ from .retry import fixed from .runner import BaseRunner from .task import Task +from .types import RetryStrategy, ScheduleStrategy from .worker import LOCK_REDIS_KEY, Worker __all__ = ["TaskTiger"] @@ -312,10 +313,13 @@ def task( lock_key: Optional[Collection[str]] = None, retry: Optional[bool] = None, retry_on: Optional[Collection[Type[BaseException]]] = None, - retry_method: Optional[ - Union[Callable[[int], float], Tuple[Callable[..., float], Tuple]] + retry_method: Optional[Union[Callable[[int], float], RetryStrategy]] = None, + schedule: Optional[ + Union[ + Callable[[datetime.datetime], Optional[datetime.datetime]], + ScheduleStrategy, + ] ] = None, - schedule: Optional[Callable] = None, batch: bool = False, max_queue_size: Optional[int] = None, max_stored_executions: Optional[int] = None, @@ -441,9 +445,7 @@ def delay( when: Optional[Union[datetime.datetime, datetime.timedelta]] = None, retry: Optional[bool] = None, retry_on: Optional[Collection[Type[BaseException]]] = None, - retry_method: Optional[ - Union[Callable[[int], float], Tuple[Callable[..., float], Tuple]] - ] = None, + retry_method: Optional[Union[Callable[[int], float], RetryStrategy]] = None, max_queue_size: Optional[int] = None, max_stored_executions: Optional[int] = None, runner_class: Optional[Type["BaseRunner"]] = None, diff --git a/tasktiger/types.py b/tasktiger/types.py index c0e3949..848386e 100644 --- a/tasktiger/types.py +++ b/tasktiger/types.py @@ -1,3 +1,9 @@ -from typing import Callable, Tuple +import datetime +from typing import Any, Callable, Optional, Sequence, Tuple -RetryStrategy = Tuple[Callable[..., float], Tuple] +RetryStrategy = Tuple[Callable[..., float], Tuple[Any, ...]] +ScheduleStrategy = Tuple[Callable[..., Optional[datetime.datetime]], Tuple[Any, ...]] + +# A retry strategy as stored with a task. The function is replaced with its +# import path, and the arguments come back as a list after a JSON round trip. +SerializedRetryStrategy = Tuple[str, Sequence[Any]]