Class Task

tasktiger/task.py:43–660

class Task:
    def __init__(
        self,
        tiger: "TaskTiger",
        func: Optional[Callable] = None,
        args: Optional[Any] = None,
        kwargs: Optional[Any] = None,
        queue: Optional[str] = None,
        hard_timeout: Optional[float] = None,
        unique: Optional[bool] = None,
        unique_key: Optional[Collection[str]] = None,
        lock: Optional[bool] = None,
        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,
        max_queue_size: Optional[int] = None,
        max_stored_executions: Optional[int] = None,
        runner_class: Optional[Type["BaseRunner"]] = None,
        # internal variables
        _data: Any = None,
        _state: Any = None,
        _ts: Any = None,
        _executions: Optional[List[Dict[str, Any]]] = None,
    ):
        """
        Queues a task. See README.rst for an explanation of the options.
        """

        if func and queue is None:
            queue = Task.queue_from_function(func, tiger)

        self.tiger = tiger
        self._func = func
        self._queue = queue
        self._state = _state
        self._ts = _ts
        self._executions = _executions or []

        # Internal initialization based on raw data.
        if _data is not None:
            self._data = _data
            return

        assert func

        serialized_name = serialize_func_name(func)

        if unique is None:
            unique = getattr(func, "_task_unique", False)

        if unique_key is None:
            unique_key = getattr(func, "_task_unique_key", None)

        if lock is None:
            lock = getattr(func, "_task_lock", False)