Requires version 0.8.93 or greater.

Conditional blocks are part of your pipeline but don’t run as individual steps in your pipeline like data loader blocks, transformer blocks, etc. However, conditional blocks are associated to other blocks within the pipeline (e.g. data loaders, transformers, data exporters, etc).

Before those other blocks are executed, any associated conditional blocks will be executed. If any conditional block returns False, the parent block will not be executed, and its status will be set as condition_failed.


Example

  1. Your pipeline has the following blocks:

    1. load_data_from_api (data loader)
    2. clean_column_names (transformer)
    3. save_data (data exporter)
  2. Then, you add the following conditional block to your pipeline:

    @condition
    def check_if_data_exists(*args, **kwargs) -> bool:
        # some code...
    
  3. You associate the above conditional block to the following block:

    • clean_column_names (transformer)
  4. When you run the pipeline, it’ll execute the load_data_from_api (data loader) block first. It completes, but the dataframe returned is empty.

  5. Next, the pipeline will execute the conditional block and execute the function check_if_data_exists. Since there is no data in the dataframe, the function returns False.

  6. The clean_column_names transformer block is not executed, and the status is set as condition_failed. The save_data data exporter block is also not executed, and is set as condition_failed.


How to add conditionals to your pipeline

  1. Create a new pipeline or open an existing pipeline.
  2. Edit the pipeline.
  3. 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.
  4. Click the button Conditionals.
  5. Click the button + Conditional block.
  6. In the conditional block’s code, add a function that returns a boolean value and decorate it with @condition.

Positional arguments

The positional arguments to the conditional block will be the same as the positional arguments to the parent block. You can use the positional arguments to get the output of the upstream blocks.

Keyword arguments

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

NameDescriptionSample value
block_uuidConditional block UUID.'fireball_conditional'
dsDate string when the parent block started executing.'2023-12-25'
eventA dictionary containing metadata from an event triggered pipeline.{}
execution_datePython datetime object for when the trigger run started executing.datetime.datetime(2023, 4, 26, 20, 28, 17, 335254, tzinfo=datetime.timezone.utc)
execution_partitionPartition used for the trigger run when it was executed.'207/20230426T202817'
hrHour string when the trigger run started executing.'20'
pipeline_runPython 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_uuidUUID of the current pipeline.'fire_etl'