Lifecycle Methods

Lifecycle methods can be registered through the bootstrap method.

Lifecycle hooks can be set using the hooks parameter of the bootstrap method:

bootstrap(hook={
    # dictionary of hook names and a list of their hooks
})

on_task_start

Runs as a task is starting.

Parameters

Name	Type	Default	Description
task	Optional[Task]	None	The task about to run.

Example

from cadence import bootstrap, Task, TaskPipelineHooks, TaskResult

def on_task_start(task: Task):
    print(f"Task {task.id} started...")

if __name__ == "__main__":
    hooks: TaskPipelineHooks = {
        "on_pipeline_start": [on_pipeline_start],
    }
    bootstrap(hooks=hooks, tasks=[
        #...
    ])

on_task_finish

Runs after a task has successfully finished.

Parameters

Name	Type	Default	Description
result	Optional[TaskResult]	None	The result of the task.

Example

from cadence import bootstrap, Task, TaskPipelineHooks, TaskResult

def on_task_finish(result: TaskResult):
    print(f"Task {result.task.id} {'passed' if result.passed else 'failed'}")

if __name__ == "__main__":
    hooks: TaskPipelineHooks = {
        "on_task_finish": [on_task_finish],
    }
    bootstrap(hooks=hooks, tasks=[
        #...
    ])

tip

The Task object can be accessed from the returned TaskResult#task field.

on_task_fail

Runs after a task has failed.

Parameters

Name	Type	Default	Description
result	Optional[TaskResult]	None	The result of the task.
exc	Optional[Exception]	None	The exception raised if applicable.

Example

from cadence import bootstrap, Task, TaskPipelineHooks, TaskResult

def on_task_fail(result: TaskResult, exc: Exception):
    print(f"Task {result.task.id} failed: {exc}")

if __name__ == "__main__":
    hooks: TaskPipelineHooks = {
        "on_task_fail": [on_task_fail],
    }
    bootstrap(hooks=hooks, tasks=[
        #...
    ])

on_queue_start

Runs as the task queue is starting.

Parameters

N/A

Example

from cadence import bootstrap, Task, TaskPipelineHooks, TaskResult

def on_queue_start():
    print("Queue started")

if __name__ == "__main__":
    hooks: TaskPipelineHooks = {
        "on_queue_start": [on_queue_start],
    }
    bootstrap(hooks=hooks, tasks=[
        #...
    ])

on_queue_stop

Runs as the task queue stops.

Parameters

N/A

Example

from cadence import bootstrap, Task, TaskPipelineHooks, TaskResult

def on_queue_stop():
    print("Queue stopped")

if __name__ == "__main__":
    hooks: TaskPipelineHooks = {
        "on_queue_stop": [on_queue_stop],
    }
    bootstrap(hooks=hooks, tasks=[
        #...
    ])

on_pipeline_start

Runs as the task pipeline is starting.

Parameters

N/A

Example

from cadence import bootstrap, Task, TaskPipelineHooks, TaskResult

def on_pipeline_start():
    print("Pipeline is starting")

if __name__ == "__main__":
    hooks: TaskPipelineHooks = {
        "on_pipeline_start": [on_pipeline_start],
    }
    bootstrap(hooks=hooks, tasks=[
        #...
    ])

on_pipeline_stop

Runs as the task pipeline is stopping.

Parameters

N/A

Example

from cadence import bootstrap, Task, TaskPipelineHooks, TaskResult

def on_pipeline_stop():
    print("Pipeline is stopping")

if __name__ == "__main__":
    hooks: TaskPipelineHooks = {
        "on_pipeline_stop": [on_pipeline_stop],
    }
    bootstrap(hooks=hooks, tasks=[
        #...
    ])