Task Run

It can be imported from fluid.scheduler:

from fluid.scheduler import TaskRun

fluid.scheduler.TaskRun pydantic-model

Bases: BaseModel, Generic[TP, TD]

A TaskRun contains all the data generated by a Task run

This model is never initialized directly, it is created by the TaskManager

Fields:

• id (str)
• task (Task)
• priority (TaskPriority)
• params (TP)
• state (TaskState)
• task_manager (TaskManager)
• queued (datetime | None)
• start (datetime | None)
• end (datetime | None)
• execute_after (datetime | None)
• rate_limit_attempt (int)
• retry_attempt (int)
• from_run_id (str)
• root_run_id (str)

id pydantic-field

id

Unique task run id

task pydantic-field

task

Task to be executed

priority pydantic-field

priority

Task priority

params pydantic-field

params

Task parameters

state pydantic-field

state = TaskState.init

Task state

task_manager pydantic-field

task_manager

queued pydantic-field

queued = None

start pydantic-field

start = None

end pydantic-field

end = None

execute_after pydantic-field

execute_after = None

Do not execute before this UTC timestamp. Set by retry logic.

rate_limit_attempt pydantic-field

rate_limit_attempt = 0

Number of rate-limit retries already consumed.

retry_attempt pydantic-field

retry_attempt = 0

Number of failure retries already consumed.

from_run_id pydantic-field

from_run_id = ''

ID of the task run that queued this one, empty when the run was not queued from another task run.

root_run_id pydantic-field

root_run_id = ''

ID of the first task run in the chain, shared by every run queued from it, directly or indirectly. Empty for a run that starts a chain.

logger property

logger

in_queue property

in_queue

duration property

duration

duration_ms property

duration_ms

total property

total

name property

name

name_id property

name_id

is_done property

is_done

is_failure property

is_failure

deps property

deps

Dependencies of the TaskManager running this task.

Annotate the second type parameter to type them, for example TaskRun[MyParams, MyDeps]. It defaults to Any, therefore TaskRun and TaskRun[MyParams] keep working unchanged.

abort

abort(reason='')

Abort the task run by raising TaskAbortedError.

Source code in fluid/scheduler/models.py

def abort(self, reason: str = "") -> None:
    """Abort the task run by raising
    [TaskAbortedError][fluid.scheduler.errors.TaskAbortedError].
    """
    raise TaskAbortedError(reason) from None

set_state

set_state(state, state_time=None)

Set the state of the task run, with proper handling of timestamps and state transitions.

This method is called by the task consumer and should not be called directly by the task executor.

Source code in fluid/scheduler/models.py

def set_state(
    self,
    state: TaskState,
    state_time: datetime | None = None,
) -> None:
    """Set the state of the task run, with proper handling of timestamps
    and state transitions.

    This method is called by the task consumer and should not be called directly
    by the task executor.
    """
    if self.state == state:
        return
    state_time = as_utc(state_time)
    match (self.state, state):
        case (TaskState.init, TaskState.queued):
            self.queued = state_time
            self.state = state
            self._dispatch()
        case (TaskState.init, _):
            self.set_state(TaskState.queued, state_time)
            self.set_state(state, state_time)
        case (TaskState.queued, TaskState.running):
            self.start = state_time
            self.state = state
            self._dispatch()
        case (
            TaskState.queued,
            TaskState.success
            | TaskState.aborted
            | TaskState.rate_limited
            | TaskState.failure,
        ):
            self.set_state(TaskState.running, state_time)
            self.set_state(state, state_time)
        case (
            TaskState.running,
            TaskState.success
            | TaskState.aborted
            | TaskState.rate_limited
            | TaskState.failure
            | TaskState.interrupted,
        ):
            self.end = state_time
            self.state = state
            self._dispatch()
        case _:
            raise TaskRunError(f"invalid state transition {self.state} -> {state}")

queue async

queue(task, *, run_id='', priority=None, **params)

Queue another task from within this task run.

The new run records this run in from_run_id and inherits its root_run_id, so a chain of tasks queued this way can be traced back to the run that started it.

This returns as soon as the run is on the queue, it does not wait for it to execute.

PARAMETER	DESCRIPTION
task	The task or task name, if a task name it must be registered with the task manager. TYPE: str | Task
run_id	Unique ID for the task run. If not provided a new UUID is generated. TYPE: str DEFAULT: ''
priority	Override the default task priority if provided TYPE: TaskPriority | None DEFAULT: None
**params	The optional parameters for the task run. They must match the task params model TYPE: Any DEFAULT: {}

Source code in fluid/scheduler/models.py

async def queue(
    self,
    task: Annotated[
        str | Task,
        Doc(
            "The task or task name,"
            " if a task name it must be registered with the task manager."
        ),
    ],
    *,
    run_id: Annotated[
        str,
        Doc("Unique ID for the task run. If not provided a new UUID is generated."),
    ] = "",
    priority: Annotated[
        TaskPriority | None, Doc("Override the default task priority if provided")
    ] = None,
    **params: Annotated[
        Any,
        Doc(
            "The optional parameters for the task run. "
            "They must match the task params model"
        ),
    ],
) -> TaskRun:
    """Queue another task from within this task run.

    The new run records this run in
    [from_run_id][fluid.scheduler.TaskRun.from_run_id] and inherits its
    [root_run_id][fluid.scheduler.TaskRun.root_run_id], so a chain of tasks
    queued this way can be traced back to the run that started it.

    This returns as soon as the run is on the queue, it does not wait for it
    to execute.
    """
    return await self.task_manager.queue(
        task,
        run_id=run_id,
        priority=priority,
        from_task_run=self,
        **params,
    )

lock

lock(timeout=None, name=None)

Get a lock for this task run

Source code in fluid/scheduler/models.py

def lock(self, timeout: float | None = None, name: str | None = None) -> Lock:
    """Get a lock for this task run"""
    lock_name = f"tasks:{self.name}"
    if name:
        lock_name = f"{lock_name}:{name}"
    return self.task_manager.broker.lock(lock_name, timeout=timeout)

queue_dump_json

queue_dump_json()

Serialize the task run for the task queue

Params are dumped with secret values revealed so they survive the round-trip through the queue - all other dumps keep secrets masked.

Source code in fluid/scheduler/models.py

def queue_dump_json(self) -> bytes:
    """Serialize the task run for the task queue

    Params are dumped with secret values revealed so they survive the
    round-trip through the queue - all other dumps keep secrets masked.
    """
    data = self.model_dump(mode="json", exclude={"params"})
    data["params"] = params_dump(self.params)
    return to_json(data)