I have two tasks in my workflow:

@workflow
def data_processing_workflow(
    trigger_file_s3_uri: str,
    triggering_timestamp: str,
    hub_name: str,
    environment: str,
    poll_interval: int = 60,
    poll_timeout: int = 7200,
    sleep_time: int = 0
) -> None:

    wait_until_timestamp_task(
        triggering_timestamp=triggering_timestamp,
        sleep_time=sleep_time
    )

    should_launch, processing_params = validate_and_get_params_task(
        trigger_file_s3_uri=trigger_file_s3_uri,
        hub_name=hub_name,
        environment=environment
    )

...

Notice

validate_and_get_params_task

do not receive inputs from

wait_until_timestamp_task

, which outputs None.

def wait_until_timestamp_task(
    triggering_timestamp: str,
    sleep_time: int = 120
) -> None:
...

Currently

validate_and_get_params_task

and

wait_until_timestamp_task

runs in parallel inside

data_processing_workflow

How can I make sure

validate_and_get_params_task

runs only after

wait_until_timestamp_task

, even though the former do not receive inputs from the latter? It worked to alter

wait_until_timestamp_task

to output a dummy string then alter

validate_and_get_params_task

to receive this dummy string as input, but it seems hacky. Is there another way?

Hi Marcus! can consider the chaining operator (docs here)

@astonishing-eve-54331 thanks for the reply! How can I use the chaining operator taking into account the two outputs returned by

validate_and_get_params_task

?

should_launch, processing_params = validate_and_get_params_task(...)

These outputs are used as inputs for the following workflow tasks

That is a good question - I haven't done something like this before. Could you try something like this?

outputs = validate_and_get_params_task(...)

wait_until_timestamp_task >> outputs

should_launch, processing_params = outputs

OR

should_launch, processing_params = validate_and_get_params_task(...)

wait_until_timestamp_task >> processing_params