Task API Reference

boilermaker.task.Task

Bases: BaseModel

Represents a serializable task with retry policies and callback chains.

A Task encapsulates a function call with its arguments, retry configuration, and optional success/failure callbacks. Tasks are JSON-serializable and can be published to Azure Service Bus for asynchronous execution.

Attributes:

Name	Type	Description
task_id	TaskId	Unique UUID7 identifier for timestamp-ordered task identification
should_dead_letter	bool	Whether failed tasks should be dead-lettered (default: True)
acks_late	bool	Whether to acknowledge messages after processing (default: True)
function_name	str	Name of the registered function to execute
attempts	RetryAttempts	Retry attempt tracking and metadata
policy	RetryPolicy	Retry policy governing backoff and max attempts
payload	dict[str, Any]	Function arguments and keyword arguments (must be JSON-serializable)
_sequence_number	int | None	Service Bus sequence number (set after publishing)
on_success	Optional[Task]	Optional callback task to run on successful completion
on_failure	Optional[Task]	Optional callback task to run on failure

Example:

# Create task with default settings
task = Task.default("my_function", args=[1, 2], kwargs={"key": "value"})

# Create task with custom retry policy
policy = RetryPolicy(max_tries=5, backoff_mode="exponential")
task = Task.si(my_function, arg1, kwarg=value, policy=policy)

# Chain tasks with callbacks
task1 >> task2 >> task3  # Success chain
task1.on_failure = error_handler_task

acks_early property

Whether the task acknowledges messages before processing.

Returns:

Name	Type	Description
bool		True if acks_late is False, meaning messages are acknowledged immediately upon receipt

can_retry property

Whether the task can be retried based on current attempts.

Returns:

Name	Type	Description
bool		True if current attempts are less than or equal to the maximum tries allowed by the retry policy

diagnostic_id cached property

Get the OpenTelemetry diagnostic ID if available.

Returns:

Type	Description
str | None	str | None: The diagnostic ID for tracing, or None if not set

msg property writable

Get the associated Service Bus message if available.

Returns:

Type	Description
ServiceBusReceivedMessage | None	ServiceBusReceivedMessage | None: The associated Service Bus message, or None if not set

sequence_number cached property

Get the Service Bus sequence number if available.

Returns:

Type	Description
int | None	int | None: The sequence number of the associated Service Bus message, or None if not set

__hash__()

Hash based on unique task_id for use in sets and dicts.

Source code in boilermaker/task/task.py

def __hash__(self) -> int:
    """Hash based on unique task_id for use in sets and dicts."""
    return hash(self.task_id)

__lshift__(other)

Set success callback using << operator (left-shift chaining).

Creates a success callback chain where this task executes if the right-hand task completes successfully.

Parameters:

Name	Type	Description	Default
other	Task	Task that will trigger this task on success	required

Returns:

Name	Type	Description
Task	Task	This task, allowing for continued chaining

Example:

# If task3 succeeds, run task2
# If task2 succeeds, run task1
task1 << task2 << task3

Source code in boilermaker/task/task.py

def __lshift__(self, other: "Task") -> "Task":
    """Set success callback using << operator (left-shift chaining).

    Creates a success callback chain where this task executes
    if the right-hand task completes successfully.

    Args:
        other: Task that will trigger this task on success

    Returns:
        Task: This task, allowing for continued chaining

    Example:

        # If task3 succeeds, run task2
        # If task2 succeeds, run task1
        task1 << task2 << task3
    """
    other.on_success = self
    return self

__rshift__(other)

Set success callback using >> operator (right-shift chaining).

Creates a success callback chain where the right-hand task executes if this task completes successfully.

Parameters:

Name	Type	Description	Default
other	Task	Task to execute on success	required

Returns:

Name	Type	Description
Task	Task	The other task, allowing for continued chaining

Example:

# If task1 succeeds, run task2
# If task2 succeeds, run task3
task1 >> task2 >> task3

Source code in boilermaker/task/task.py

def __rshift__(self, other: "Task") -> "Task":
    """Set success callback using >> operator (right-shift chaining).

    Creates a success callback chain where the right-hand task
    executes if this task completes successfully.

    Args:
        other: Task to execute on success

    Returns:
        Task: The other task, allowing for continued chaining

    Example:

        # If task1 succeeds, run task2
        # If task2 succeeds, run task3
        task1 >> task2 >> task3
    """
    self.on_success = other
    return other

default(function_name, **kwargs) classmethod

Create a Task with default retry settings.

Convenience method to create a task with sensible defaults for retry attempts and policy. Additional keyword arguments override default values.

Parameters:

Name	Type	Description	Default
function_name	str	Name of the function to execute	required
**kwargs		Additional task attributes to override defaults	{}

Returns:

Name	Type	Description
Task		New task instance with default retry configuration

Example:

task = Task.default("process_data", payload={"data": [1, 2, 3]})

Source code in boilermaker/task/task.py

@classmethod
def default(cls, function_name: str, **kwargs):
    """Create a Task with default retry settings.

    Convenience method to create a task with sensible defaults for
    retry attempts and policy. Additional keyword arguments override
    default values.

    Args:
        function_name: Name of the function to execute
        **kwargs: Additional task attributes to override defaults

    Returns:
        Task: New task instance with default retry configuration

    Example:

        task = Task.default("process_data", payload={"data": [1, 2, 3]})
    """
    attempts = retries.RetryAttempts.default()
    policy = retries.RetryPolicy.default()
    if "policy" in kwargs:
        policy = kwargs.pop("policy")
    if "payload" in kwargs:
        payload = kwargs.pop("payload")
        if "args" not in payload:
            payload["args"] = []
        if "kwargs" not in payload:
            payload["kwargs"] = {}
    else:
        payload = {}
    return cls(
        attempts=attempts,
        function_name=function_name,
        policy=policy,
        payload=payload,
        **kwargs,
    )

get_next_delay()

Calculate the delay before the next retry attempt.

Uses the task's retry policy to determine the appropriate delay based on the current number of attempts.

Returns:

Name	Type	Description
int		Delay in seconds before next retry attempt

Source code in boilermaker/task/task.py

def get_next_delay(self):
    """Calculate the delay before the next retry attempt.

    Uses the task's retry policy to determine the appropriate
    delay based on the current number of attempts.

    Returns:
        int: Delay in seconds before next retry attempt
    """
    return self.policy.get_delay_interval(self.attempts.attempts)

mark_published(sequence_number)

Mark the task as published by setting the sequence number.

Parameters:

Name	Type	Description	Default
sequence_number	int	The Service Bus sequence number assigned upon publishing	required

Source code in boilermaker/task/task.py

def mark_published(self, sequence_number: int) -> None:
    """Mark the task as published by setting the sequence number.

    Args:
        sequence_number: The Service Bus sequence number assigned upon publishing
    """
    self._sequence_number = sequence_number

record_attempt()

Record a new execution attempt with current timestamp.

Increments the attempt counter and records the current UTC timestamp for tracking retry intervals and debugging.

Returns:

Name	Type	Description
RetryAttempts		Updated attempts object with incremented count

Source code in boilermaker/task/task.py

def record_attempt(self):
    """Record a new execution attempt with current timestamp.

    Increments the attempt counter and records the current UTC
    timestamp for tracking retry intervals and debugging.

    Returns:
        RetryAttempts: Updated attempts object with incremented count
    """
    now = datetime.datetime.now(datetime.UTC)
    return self.attempts.inc(now)

si(fn, *fn_args, should_dead_letter=True, acks_late=True, policy=None, **fn_kwargs) classmethod

Create an immutable signature task from a function and arguments.

Creates a task bound to specific function arguments, useful for preparing tasks with callbacks or custom settings before publishing. The function arguments are captured at creation time.

Note: This implementation creates immutable signatures only. Future versions may support mutable signatures where task outputs are passed to subsequent tasks in a chain.

Parameters:

Name	Type	Description	Default
fn	TaskHandler	The function to be executed	required
*fn_args		Positional arguments for the function	()
should_dead_letter	bool	Whether to dead-letter failed tasks (default: True)	True
acks_late	bool	Whether to acknowledge after processing (default: True)	True
policy	RetryPolicy | None	Custom retry policy (uses default if None)	None
**fn_kwargs		Keyword arguments for the function	{}

Returns:

Name	Type	Description
Task	Task	New task with bound function signature

Example:

def process_data(data, format="json"):
    return f"Processed {data} as {format}"

# Arguments are bound to the task **immutably** at creation time
task = Task.si(process_data, [1, 2, 3], format="xml")

Source code in boilermaker/task/task.py

@classmethod
def si(
    cls,
    fn: TaskHandler,
    *fn_args,
    should_dead_letter: bool = True,
    acks_late: bool = True,
    policy: retries.RetryPolicy | None = None,
    **fn_kwargs,
) -> "Task":
    """Create an immutable signature task from a function and arguments.

    Creates a task bound to specific function arguments, useful for
    preparing tasks with callbacks or custom settings before publishing.
    The function arguments are captured at creation time.

    Note: This implementation creates immutable signatures only.
    Future versions may support mutable signatures where task outputs
    are passed to subsequent tasks in a chain.

    Args:
        fn: The function to be executed
        *fn_args: Positional arguments for the function
        should_dead_letter: Whether to dead-letter failed tasks (default: True)
        acks_late: Whether to acknowledge after processing (default: True)
        policy: Custom retry policy (uses default if None)
        **fn_kwargs: Keyword arguments for the function

    Returns:
        Task: New task with bound function signature

    Example:

        def process_data(data, format="json"):
            return f"Processed {data} as {format}"

        # Arguments are bound to the task **immutably** at creation time
        task = Task.si(process_data, [1, 2, 3], format="xml")
    """
    attempts = retries.RetryAttempts.default()
    policy = policy or retries.RetryPolicy.default()

    return cls(
        should_dead_letter=should_dead_letter,
        acks_late=acks_late,
        attempts=attempts,
        function_name=fn.__name__,
        policy=policy,
        payload={
            "args": fn_args,
            "kwargs": fn_kwargs,
        },
    )

options: show_root_heading: true show_source: true heading_level: 2