orangeqs.juice.task

Task definitions for services.

Package Contents

Classes

Task	Base class for tasks that can be executed by a service.
TaskFuture	Future that resolves to the result of a task.
TaskResultOk	Result information of a successfully completed task.
TaskResultError	Result information of a task that raised an error.

Data

TaskResult

Union of all possible task results. Either TaskResultOk or TaskResultError.

API

class orangeqs.juice.task.Task(/, **data: Any)

Bases: pydantic.BaseModel

Base class for tasks that can be executed by a service.

classmethod type() → str

Return the unique name of the task type.

This is used to validate the model on the service side and route the task to the correct handler. Defaults to the class name.

Warning: The name of the type should be unique per service. Can be overridden to prevent name clashes.

property hidden_fields: set[str]

Return a set of field names to exclude from the display name.

By default, this excludes the parallel class variable.

property display_name: str

Return a human-readable name for the task.

Defaults to the task type and all fields except those in hidden_fields. For example: MyTask: param1=42, param2='foo'.

Can be overridden to provide a more descriptive name.

parallel: ClassVar[bool]

False

Whether this task can be executed in parallel with other tasks.

Tasks are executed sequentially by default. Classes that set this to True will be executed as soon as they are received, without waiting for other tasks to finish.

class orangeqs.juice.task.TaskFuture(task_id: str)

Bases: asyncio.Future[orangeqs.juice.task._task._T]

Future that resolves to the result of a task.

task_id: str

None

Unique identifier of the task.

orangeqs.juice.task.TaskResult

None

Union of all possible task results. Either TaskResultOk or TaskResultError.

class orangeqs.juice.task.TaskResultOk(/, **data: Any)

Bases: pydantic.BaseModel

Result information of a successfully completed task.

status: Literal[ok]

‘ok’

id: str

None

The unique identifier of the task.

result: Any

None

The result of the task execution.

class orangeqs.juice.task.TaskResultError(/, **data: Any)

Bases: pydantic.BaseModel

Result information of a task that raised an error.

status: Literal[error]

‘error’

id: str

None

The unique identifier of the task.

ename: str

None

The name of the error type.

evalue: str

None

The error message.

traceback: str

None

The traceback of the error.

exception orangeqs.juice.task.TaskExecutionError(error: orangeqs.juice.task._schemas.TaskResultError)

Bases: Exception

Task execution error, raised if a task fails.

Raised by request() and execute() when a task fails with check=True

error: orangeqs.juice.task._schemas.TaskResultError

None

The wrapped error details.

property traceback: str

The traceback of the error.