Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions tasktiger/_internal.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand Down
8 changes: 5 additions & 3 deletions tasktiger/schedule.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
import datetime
from typing import Callable, Optional, Tuple
from typing import Optional

from .types import ScheduleStrategy

__all__ = ["periodic", "cron_expr"]

Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
8 changes: 3 additions & 5 deletions tasktiger/task.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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,
Expand Down Expand Up @@ -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
Expand Down
14 changes: 8 additions & 6 deletions tasktiger/tasktiger.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"]
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down
10 changes: 8 additions & 2 deletions tasktiger/types.py
Original file line number Diff line number Diff line change
@@ -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]]
Loading