Перейти к содержанию

Task

Task Schema.

Task dataclass

Task модель.

Parameters:

Name Type Description Default
status str

Статус.

 required
uuid UUID

UUID.

 required
priority int

Приоритет.

 required
task_name str

Название.

 required
args Tuple[str]

Аргументы типа args.

 required
kwargs Dict[str, Any]

Аргументы типа kwargs.

 required
created_at datetime

Дата создания.

 required
updated_at datetime

Дата обновления.

 required
returning str | None

Результат. По умолчанию: None.

 None
traceback str | None

Трассировка ошибок. По умолчанию: None.

 None
Source code in src/qtasks/schemas/task.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
@dataclass
class Task:
    """`Task` модель.

    Args:
        status (str): Статус.
        uuid (UUID): UUID.
        priority (int): Приоритет.
        task_name (str): Название.

        args (Tuple[str]): Аргументы типа args.
        kwargs (Dict[str, Any]): Аргументы типа kwargs.

        created_at (datetime): Дата создания.
        updated_at (datetime): Дата обновления.

        returning (str | None): Результат. По умолчанию: `None`.
        traceback (str | None): Трассировка ошибок. По умолчанию: `None`.
    """

    status: str
    uuid: UUID
    priority: int
    task_name: str

    args: Tuple[str]
    kwargs: Dict[str, Any]

    created_at: datetime
    updated_at: datetime

    returning: InitVar[Union[str, None]] = None
    traceback: InitVar[Union[str, None]] = None

    def wait_result(
        self,
        timeout: Annotated[
            Optional[float],
            Doc(
                """
                    Таймаут задачи.

                    Если указан, задача возвращается через `qtasks.results.SyncTask`.
                    """
            ),
        ] = None
    ) -> Union["Task", None]:
        """Ожидание результата задачи Синхронно.

        Args:
            timeout (Annotated[Optional[float], Doc], optional): Таймаут ожидания результата. По умолчанию: `None`.
        """
        return SyncResult(uuid=self.uuid).result(timeout=timeout)

    async def wait_result_async(
        self,
        timeout: Annotated[
            Optional[float],
            Doc(
                """
                    Таймаут задачи.

                    Если указан, задача возвращается через `qtasks.results.AsyncTask`.
                    """
            ),
        ] = None
    ) -> Union["Task", None]:
        """Ожидание результата задачи Асинхронно.

        Args:
            timeout (Annotated[Optional[float], Doc], optional): Таймаут ожидания результата. По умолчанию: `None`.
        """
        return await AsyncResult(uuid=self.uuid).result(timeout=timeout)

wait_result(timeout=None)

Ожидание результата задачи Синхронно.

Parameters:

Name Type Description Default
timeout Annotated[Optional[float], Doc]

Таймаут ожидания результата. По умолчанию: None.

 None
Source code in src/qtasks/schemas/task.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def wait_result(
    self,
    timeout: Annotated[
        Optional[float],
        Doc(
            """
                Таймаут задачи.

                Если указан, задача возвращается через `qtasks.results.SyncTask`.
                """
        ),
    ] = None
) -> Union["Task", None]:
    """Ожидание результата задачи Синхронно.

    Args:
        timeout (Annotated[Optional[float], Doc], optional): Таймаут ожидания результата. По умолчанию: `None`.
    """
    return SyncResult(uuid=self.uuid).result(timeout=timeout)

wait_result_async(timeout=None) async

Ожидание результата задачи Асинхронно.

Parameters:

Name Type Description Default
timeout Annotated[Optional[float], Doc]

Таймаут ожидания результата. По умолчанию: None.

 None
Source code in src/qtasks/schemas/task.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
async def wait_result_async(
    self,
    timeout: Annotated[
        Optional[float],
        Doc(
            """
                Таймаут задачи.

                Если указан, задача возвращается через `qtasks.results.AsyncTask`.
                """
        ),
    ] = None
) -> Union["Task", None]:
    """Ожидание результата задачи Асинхронно.

    Args:
        timeout (Annotated[Optional[float], Doc], optional): Таймаут ожидания результата. По умолчанию: `None`.
    """
    return await AsyncResult(uuid=self.uuid).result(timeout=timeout)