Task

class Task:
    def execute(context: ExecutionContext) -> Awaitable[None] | None

    @staticmethod
    def identifier() -> tuple[str, str]

Copy

Base class for Tilebox workflows tasks. Inherit from this class to create a task. Inheriting also automatically applies the dataclass decorator.

Methods

def Task.execute(context: ExecutionContext) -> Awaitable[None] | None

Copy

The entry point for the execution of the task. Define it as either def execute(...) -> None or async def execute(...) -> None. The runner waits for an asynchronous method to complete.

@staticmethod
def Task.identifier() -> tuple[str, str]

Copy

Override a task identifier and specify its version. If not overridden, the identifier of a task defaults to the class name, and the version to v0.0.

Task Input Parameters

class MyTask(Task):
    message: str
    value: int
    data: dict[str, int]

Copy

Optional task input parameters, defined as annotated class attributes. See Python task inputs for the supported types and their package requirements.

Request example

Python

from tilebox.workflows import Task, ExecutionContext

class MyFirstTask(Task):
    def execute(self, context: ExecutionContext):
        print(f"Hello World!")

    @staticmethod
    def identifier() -> tuple[str, str]:
        return ("tilebox.workflows.MyTask", "v3.2")

class MyFirstParameterizedTask(Task):
    name: str
    greet: bool
    data: dict[str, str]

    def execute(self, context: ExecutionContext):
        if self.greet:
            print(f"Hello {self.name}!")

Copy