fabricatio_core.models.task

This module defines the Task class, which represents a task with a status and output.

It includes methods to manage the task’s lifecycle, such as starting, finishing, cancelling, and failing the task.

Attributes

NameSpace

Classes

Task

A class representing a task with status management and output handling.

Module Contents

type fabricatio_core.models.task.NameSpace = str | List[str]

class fabricatio_core.models.task.Task[T](/, **data: Any)[source]

Bases: fabricatio_core.models.generic.WithBriefing, fabricatio_core.models.generic.ProposedAble, fabricatio_core.models.generic.WithDependency

A class representing a task with status management and output handling.

goals: List[str] = None: Objectives the task aims to achieve.

dependencies: List[str] = None: File paths necessarily needed to read or write to complete the task. Do add path(s) needed!

description: str = None: Detailed explanation of the task with 5W1H rule.

name: str = None: Concise and descriptive name of the task.

send_to: List[str] = None

List of namespace path components used to construct the target task queue.

The full queue path is formed as: <component1>::<component2>::…::*::Pending. For example: - with [‘work’] will be received by ‘work::::Pending’ - with [‘write’, ‘book’] will be received by ‘write::book::::Pending’

⚠️ The caller must ensure that the resulting namespace (e.g., ‘write::book’) exists. Sending to a non-existent namespace may result in task loss or an error.

property extra_init_context: Dict: Extra initialization context for the task, which is designed to override the one of the Workflow.

update_init_context(/, **kwargs) → Self[source]: Update the extra initialization context for the task.

property assembled_prompt: str: Assemble both dependencies and briefing prompt for the task.

move_to(new_namespace: NameSpace) → Self[source]

Move the task to a new namespace.

Parameters:: new_namespace (str|List[str]) – The new namespace to move the task to.
Returns:: The moved instance of the Task class
Return type:: Task

Example

task = Task(name="example_task", namespace=["example"]).move_to("work")
assert task.namespace == ["work"]

append_extra_description(description: str) → Self[source]

Append a description to the task.

Parameters:: description (str) – The description to append.
Returns:: The updated instance of the Task class.
Return type:: Task

Example

task = Task(name="example_task", description="This is an example task.")
task.append_extra_description("This task is important.")
assert task.description == "This is an example task.\nThis task is important."

update_task(*, goal: List[str] | str | None = None, description: str | None = None) → Self[source]

Update the goal and description of the task.

Parameters:

goal (str|List[str], optional) – The new goal of the task.
description (str, optional) – The new description of the task.

Returns:

The updated instance of the Task class.

Return type:

Task

Example

# Update both goal and description
task = Task(name="example_task", goals=["old_goal"], description="old description")
task.update_task(goal="new_goal", description="new description")
assert task.goals == ["new_goal"]
assert task.description == "new description"

# Update only the goal with a single string
task = Task(name="example_task", goals=["old_goal"])
task.update_task(goal="new_goal")
assert task.goals == ["new_goal"]

# Update goal with a list of strings
task = Task(name="example_task", goals=["old_goal"])
task.update_task(goal=["new_goal1", "new_goal2"])
assert task.goals == ["new_goal1", "new_goal2"]

# Update only the description
task = Task(name="example_task", description="old description")
task.update_task(description="new description")
assert task.description == "new description"

async get_output() → T | None[source]

Get the output of the task.

Returns:: The output of the task.
Return type:: T

Example

# Test basic output retrieval
task = Task(name="output_task")
await task.finish("success")
assert await task.get_output() == "success"

# Test output retrieval with multiple get calls
task2 = Task(name="multi_get_task")
await task2.finish(42)
assert await task2.get_output() == 42
# Second get should return same value
assert await task2.get_output() == 42

# Test output retrieval for cancelled task
task3 = Task(name="cancelled_task")
await task3.cancel()
assert await task3.get_output() is None

status_label(status: fabricatio_core.rust.TaskStatus) → str[source]

Return a formatted status label for the task.

Parameters:: status (fabricatio.constants.TaskStatus) – The status of the task.
Returns:: The formatted status label.
Return type:: str

property pending_label: str[source]

Return the pending status label for the task.

Returns:: The pending status label.
Return type:: str

property running_label: str[source]

Return the running status label for the task.

Returns:: The running status label.
Return type:: str

property finished_label: str[source]

Return the finished status label for the task.

Returns:: The finished status label.
Return type:: str

property failed_label: str[source]

Return the failed status label for the task.

Returns:: The failed status label.
Return type:: str

property cancelled_label: str[source]

Return the cancelled status label for the task.

Returns:: The cancelled status label.
Return type:: str

async finish(output: T) → Self[source]

Mark the task as finished and set the output.

Parameters:: output (T) – The output of the task.
Returns:: The finished instance of the Task class.
Return type:: Task

async start() → Self[source]

Mark the task as running.

Returns:: The running instance of the Task class.
Return type:: Task

async cancel() → Self[source]

Mark the task as cancelled.

Returns:: The cancelled instance of the Task class.
Return type:: Task

async fail() → Self[source]

Mark the task as failed.

Returns:: The failed instance of the Task class.
Return type:: Task

publish(new_namespace: NameSpace | None = None, *, event: NameSpace | None = None) → Self[source]

Publish the task to the event bus.

Parameters:

new_namespace (EventLike, optional) – The new namespace to move the task to.
event (EventLike, optional) – The event to publish.

Returns:

The published instance of the Task class.

Return type:

Task

async delegate(new_namespace: NameSpace | None = None, *, event: NameSpace | None = None) → T | None[source]

Delegate the task to the event.

Parameters:

new_namespace (EventLike, optional) – The new namespace to move the task to.
event (EventLike, optional) – The event to publish, overrides the event in this instance and the new_namespace.

Returns:

The output of the task.

Return type:

T|None

delegate_blocking(new_namespace: NameSpace | None = None, *, event: NameSpace | None = None) → T | None[source]

Delegate the task to the event in a blocking manner.

Parameters:

new_namespace (EventLike, optional) – The new namespace to move the task to.
event (EventLike, optional) – The event to publish.

Returns:

The output of the task.

Return type:

T|None

property briefing: str

Return a briefing of the task including its goal.

Returns:: The briefing of the task.
Return type:: str

is_running() → bool[source]

Check if the task is running.

Returns:: True if the task is running, False otherwise.
Return type:: bool

is_finished() → bool[source]

Check if the task is finished.

Returns:: True if the task is finished, False otherwise.
Return type:: bool

is_failed() → bool[source]

Check if the task is failed.

Returns:: True if the task is failed, False otherwise.
Return type:: bool

is_cancelled() → bool[source]

Check if the task is cancelled.

Returns:: True if the task is cancelled, False otherwise.
Return type:: bool

is_pending() → bool[source]

Check if the task is pending.

Returns:: True if the task is pending, False otherwise.
Return type:: bool