Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Prefect Community

Hi all... Is there a way to implement a dynamic `manual_only` trigger? I'm trying to do the following: read file into dataframe -&gt; run GreatExpectations validation on dataframe -&gt; if validation passes continue with next task, if not allow the next task to run when manually triggered

something like this
```df = extract_data_frame(
    input_file_path,
    sftp_hostname,
    sftp_password,
    sftp_file_path,
  )
  try:
    vt = validation_task(
      batch_kwargs={"dataset": df, "datasource": "appts"},
      expectation_suite_name="STLUKES_DEV_6K_APPTS.warning",
      context_root_dir=ge_ctx_root
    )
    nodes = extract_nodes(df, upstream_tasks=[vt])
  except FAIL:
    nodes = extract_nodes(df, upstream_tasks=[vt], trigger=manual_only)```


Hi <@U01J4AT3C7Q> You might want to consider adding a control flow task (maybe `case` for your validation task?) and an extra task to handle the manual only trigger which can be upstream of your extract_nodes task if validation fails.  (See the pause task at the end of the example here: <https://medium.com/the-prefect-blog/needs-approval-184f2512a3cf>) and docs on control flow tasks here:  <https://docs.prefect.io/api/latest/tasks/control_flow.html#functions>

hi <@U01J4AT3C7Q>, I think I had the same q the other day and <@UKVFX6N3B> suggested this: <https://github.com/PrefectHQ/prefect/issues/3937>

Thanks <@UL455076D>, I think thats more high level/run meta info… What I’m looking for is more of alternate paths based on previous task states… Though I’ve seen <@ULXMV9SD7>’s article before her response gave me an idea for a solution…

In my flow I have as follows:
```df = extract_data_frame(
    input_file_path,
    sftp_hostname,
    sftp_password,
    sftp_file_path,
  )
  vt = validation_task(
    batch_kwargs={"dataset": df, "datasource": "appts"},
    expectation_suite_name="STLUKES_DEV_6K_APPTS.warning",
    context_root_dir=ge_ctx_root,
    upstream_tasks=[df]
                  )
  vf = validation_fail(upstream_tasks=[vt])
  ua = user_approve(upstream_tasks=[vf])
  nodes = extract_nodes(df, upstream_tasks=[vt,ua])
  ```


```def autocomplete(task, old_state, new_state):
  new_state = Skipped() if new_state.is_failed() else new_state
  return new_state

@task(trigger=any_failed, state_handlers=[autocomplete])
def validation_fail():
  pass

@task(trigger=manual_only)
def user_approve():
  pass

@task(trigger=any_successful, skip_on_upstream_skip=False)
def extract_nodes(df: DataFrame) -&gt; List[ExternalAppointmentStruct]:```


The only slight flaw here is the need for an intermediary step `validation_fail` what would have been nice to have is to have a trigger that’s a combo of `manual_only` and `any_failed`

ie: `user_approved` would trigger on `manual_only_if_any_failed` or `[manual_only, any_failed]`