Callback blocks

Requires version 0.8.61 or greater.

Callback blocks are part of your pipeline but don’t run as individual steps in your pipeline like data loader blocks, transformer blocks, etc. However, callback blocks are associated to other blocks within the pipeline (e.g. data loaders, transformers, data exporters, etc). After those other blocks are executed, any associated callback blocks will also be executed depending on the status of the parent block.

Example

Your pipeline has the following blocks:
1. load_data_from_api (data loader)
2. clean_column_names (transformer)
3. save_data (data exporter)

Then, you add the following callback block to your pipeline:

@callback('success')
def hello_world(parent_block_data, **kwargs):
    # some code...


@callback('failure')
def alert_me(parent_block_data, **kwargs):
    # some code...

@callback('cancelled')
def notify_cancelled(parent_block_data, **kwargs):
    # some code...

You associate the above callback block to the following 2 blocks:
- load_data_from_api (data loader)
- clean_column_names (transformer)
When you run the pipeline, it’ll execute the load_data_from_api (data loader) block first. It successfully completes. Then, the callback block will run and execute the function hello_world.
Next, the pipeline will execute the clean_column_names (transformer) block. It fails due to some error. The callback block will run and execute the function alert_me.

How to add callbacks to your pipeline

Create a new pipeline or open an existing pipeline.
Edit the pipeline.
On the right side of the page, click the Add-ons icon in the navigation. If you don’t see it, try expanding the right area of the page.
Click the button Callbacks.
Click the button + Callback block.
In the callback block’s code, add a function and decorate it with @callback. You can have as many functions in a callback block as you want.
Use @callback or @callback('success') to decorate a function that should only run when the parent block successfully completes.
Use @callback('failure') to decorate a function that should only run when the parent block fails.

Supported callbacks

`success`

Execute callback function when block runs successfully.

@callback('success')
def only_run_this_function_on_success(parent_block_data, **kwargs):
    pass

`failure`

Execute callback function when block fails.

@callback('failure')
def only_run_this_function_on_failure(parent_block_data, **kwargs):
    pass

`cancelled`

Available in versions >= 0.9.77.

Execute callback function when block is cancelled (manually or by system interruption).

@callback('cancelled')
def only_run_this_function_on_cancelled(parent_block_data, **kwargs):
    pass

Mage Pro callback features

Pipeline-level callbacks

In Mage Pro, callback blocks can be attached to the pipeline itself instead of only to individual parent blocks. Pipeline-level callbacks run once after a pipeline run reaches one of these terminal statuses:

completed
failed
cancelled

Block-level callbacks continue to run from their parent block status. Pipeline-level callbacks are separate and wait for the full pipeline run outcome. To configure pipeline-level callbacks in the editor:

Open the pipeline editor.
Create at least one callback block from the Add-ons panel.
In the callback block footer selector, choose the current pipeline UUID under Pipeline.

The selector groups targets under Pipeline and Blocks. The Pipeline group contains only the current pipeline UUID. If a callback block still uses block-level-only function signatures with required positional block input, such as def success_callback(parent_block_data, **kwargs), the editor still shows the current pipeline target. Saving that pipeline target shows a compatibility error because pipeline callbacks do not pass parent_block_data or other block inputs. To use the callback at the pipeline level, make block input arguments optional, for example def success_callback(parent_block_data=None, **kwargs), use documented pipeline context keyword arguments, or use a kwargs-only signature such as def success_callback(**kwargs). The original required-argument signature remains valid for block-level callbacks. You can also configure the same behavior in the pipeline metadata.yaml file or through the pipeline create and update API:

callbacks:
- name: notify stakeholders
  type: callback
  uuid: notify_stakeholders

pipeline_callbacks:
- callback_block_uuid: notify_stakeholders
  enabled: true

The association controls whether the callback block runs once for the pipeline. The callback block code controls terminal-status-specific behavior by defining matching callback functions:

@callback('success')
def notify_pipeline_completed(**kwargs):
    pipeline_run_id = kwargs.get('pipeline_run_id')
    terminal_status = kwargs.get('terminal_status')
    print(f'Pipeline run {pipeline_run_id} finished with {terminal_status}.')


@callback('failure')
def notify_pipeline_failed(**kwargs):
    error_context = kwargs.get('terminal_status')


@callback('cancelled')
def notify_pipeline_cancelled(**kwargs):
    pipeline_run_id = kwargs.get('pipeline_run_id')

Pipeline-level callbacks receive a safe pipeline-run context instead of upstream block output. Available keyword arguments are intentionally minimal:

Name	Description
`pipeline_run_id`	ID of the pipeline run that triggered the callback.
`terminal_status`	Terminal pipeline run status: `completed`, `failed`, or `cancelled`.

Use pipeline_run_id with existing authorized APIs or models if callback code needs additional pipeline-run details. Run details show each dispatched pipeline callback as a block run with its status and any user-code error captured by the normal block-run execution path. Callback statuses can be initial, queued, running, completed, or failed.

Advanced parameters

In Mage Pro, you can pass additional parameters to the @callback decorator.

`fetch_inputs`

By default, Mage will fetch upstream inputs before executing a callback. If your callback doesn’t need input data (e.g. sending alerts, posting logs), you can disable input fetching to improve performance.

@callback('failure', fetch_inputs=False)
def alert_me_without_inputs(*args, **kwargs):
    # Runs faster since upstream block inputs are not fetched
    send_alert("Parent block failed!")

fetch_inputs=True (default) → Fetch parent block inputs before executing callback
fetch_inputs=False → Skip fetching inputs, speeding up execution

Positional arguments

The 1st positional argument is the data output of the callback block’s parent block.

Positional arguments are only supported in Mage versions >= 0.8.81.

Keyword arguments

Here are the keyword arguments that are available in each callback function:

Name	Description	Sample value
`block_uuid`	Callback block UUID.	`'fireball_callback'`
`parent_block_uuid`	The block UUID of the parent block.	`'parent_block'`
`ds`	Date string when the parent block started executing.	`'2023-12-25'`
`event`	A dictionary containing metadata from an event triggered pipeline.	`{}`
`execution_date`	Python datetime object for when the parent block started executing.	`datetime.datetime(2023, 4, 26, 20, 28, 17, 335254, tzinfo=datetime.timezone.utc)`
`execution_partition`	Partition used for the parent block when it was executed.	`'207/20230426T202817'`
`hr`	Hour string when the parent block started executing.	`'20'`
`pipeline_run`	Python pipeline run object associated to the current run.	`PipelineRun(id=2357, pipeline_uuid=fire_etl, execution_date=2023-04-26 20:28:17.335254+00:00)`
`pipeline_uuid`	UUID of the current pipeline.	`'fire_etl'`
`__input`	For version `>= 0.8.81`. See below for more details.	`{ 'some_block_uuid': {} }`

`__input` keyword argument

For versions >= 0.8.81.

If your callback function is defined to only accept keyword arguments and no positional arguments, you can access the callback block’s parent block’s data output by accessing the key labeled __input in the callback function’s keyword arguments. The value of the __input key is a dictionary. The keys in that dictionary are named after the callback block’s parent block’s UUID. For example, if the callback block’s parent block’s UUID is load_api_data, then this is how you would access load_api_data’s data output:

@callback('success')
def callback_for_load_api_data(**kwargs):
    data_from_parent_block = kwargs['__input']['load_api_data']

Data integration pipelines only

The following keyword arguments are only available in callback functions with parent blocks in a data integration pipelines:

Name	Description	Sample value
`destination_table`	The table name that the source data is being exported to.	`'dim_users_v1'`
`index`	The order in which the parent block executed compared to other blocks of the same type.	`0`
`is_last_block_run`	Boolean value for whether or not the parent block was the last block ran for the block type in the pipeline.	`True` or `False`
`stream`	The name of the source stream (or table) that the parent block is syncing for.	`'stripe_transactions'`

Legacy

For versions < 0.8.61.

Adding a callback to your block

Create a block as normal.
In the top right of the block, click the triple dot icon which should open up a menu with more options. Click “Add callback”.
You should now see another code editor under the main editor for your block.
You can add an on_success callback and/or a on_failure callback. These callbacks will run after your block run completes or fails. When you run your block in the pipeline edit page, it will also run the callback depending on the status of the block.
The callback function will be passed your pipeline’s runtime variables as keyword arguments, so you can use the same kwargs.get('<variable>') syntax in your callback.

Adventure starts here

Concepts

Project

Contributing

About

Callback blocks

Example

How to add callbacks to your pipeline

Supported callbacks

`success`

`failure`

`cancelled`

Mage Pro callback features

Pipeline-level callbacks

Advanced parameters

`fetch_inputs`

Positional arguments

Keyword arguments

`__input` keyword argument

Data integration pipelines only

Legacy

Adding a callback to your block

​Example

​How to add callbacks to your pipeline

​Supported callbacks

​success

​failure

​cancelled

​Mage Pro callback features

​Pipeline-level callbacks

​Advanced parameters

​fetch_inputs

​Positional arguments

​Keyword arguments

​__input keyword argument

​Data integration pipelines only

​Legacy

​Adding a callback to your block

Example

How to add callbacks to your pipeline

Supported callbacks

`success`

`failure`

`cancelled`

Mage Pro callback features

Pipeline-level callbacks

Advanced parameters

`fetch_inputs`

Positional arguments

Keyword arguments

`__input` keyword argument

Data integration pipelines only

Legacy

Adding a callback to your block