Task

Task Schema.

Task dataclass

Task model.

Parameters:

Name	Type	Description	Default
status	str	Status.	required
uuid	UUID	UUID.	required
priority	int	Priority.	required
task_name	str	Name.	required
args	Tuple[str]	Arguments of type args.	required
kwargs	Dict[str, Any]	Arguments of type kwargs.	required
created_at	datetime	Created date.	required
updated_at	datetime	Date of update.	required
returning	str | None	Result. Default: None.	None
traceback	str | None	Trace errors. Default: None.	None

Source code in src/qtasks/schemas/task.py

@dataclass
class Task:
    """
    `Task` model.

    Args:
        status (str): Status.
        uuid (UUID): UUID.
        priority (int): Priority.
        task_name (str): Name.

        args (Tuple[str]): Arguments of type args.
        kwargs (Dict[str, Any]): Arguments of type kwargs.

        created_at (datetime): Created date.
        updated_at (datetime): Date of update.

        returning (str | None): Result. Default: `None`.
        traceback (str | None): Trace errors. Default: `None`.
    """

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

    args: tuple[str]
    kwargs: dict[str, Any]

    created_at: datetime
    updated_at: datetime

    returning: Any | None = None
    traceback: Any | None = None

    # retry
    retry: int | None = None
    retry_child_uuid: UUID | None = None
    retry_parent_uuid: UUID | None = None

    def wait_result(
        self,
        timeout: Annotated[
            float,
            Doc("""
                    Task timeout.
                    """),
        ] = 100.0,
    ) -> Union["Task", None]:
        """
        Waiting for task result Synchronously.

        Args:
            timeout (Annotated[Optional[float], Doc], optional): Timeout for waiting for the result. Default: `100.0`.
        """
        return SyncResult(uuid=self.uuid).result(timeout=timeout)

    async def wait_result_async(
        self,
        timeout: Annotated[
            float,
            Doc("""
                    Task timeout.
                    """),
        ] = 100.0,
    ) -> Union["Task", None]:
        """
        Waiting for task result Asynchronously.

        Args:
            timeout (Annotated[Optional[float], Doc], optional): Timeout for waiting for the result. Default: `100.0`.
        """
        return await AsyncResult(uuid=self.uuid).result(timeout=timeout)

wait_result(timeout=100.0)

Waiting for task result Synchronously.

Parameters:

Name	Type	Description	Default
timeout	Annotated[Optional[float], Doc]	Timeout for waiting for the result. Default: 100.0.	100.0

Source code in src/qtasks/schemas/task.py

def wait_result(
    self,
    timeout: Annotated[
        float,
        Doc("""
                Task timeout.
                """),
    ] = 100.0,
) -> Union["Task", None]:
    """
    Waiting for task result Synchronously.

    Args:
        timeout (Annotated[Optional[float], Doc], optional): Timeout for waiting for the result. Default: `100.0`.
    """
    return SyncResult(uuid=self.uuid).result(timeout=timeout)

wait_result_async(timeout=100.0) async

Waiting for task result Asynchronously.

Parameters:

Name	Type	Description	Default
timeout	Annotated[Optional[float], Doc]	Timeout for waiting for the result. Default: 100.0.	100.0

Source code in src/qtasks/schemas/task.py

async def wait_result_async(
    self,
    timeout: Annotated[
        float,
        Doc("""
                Task timeout.
                """),
    ] = 100.0,
) -> Union["Task", None]:
    """
    Waiting for task result Asynchronously.

    Args:
        timeout (Annotated[Optional[float], Doc], optional): Timeout for waiting for the result. Default: `100.0`.
    """
    return await AsyncResult(uuid=self.uuid).result(timeout=timeout)