I have a flow a la

A -> B -> C

. I want task A to be skippable via a command line argument to the Python script that implements that flow. What's the best way to do this? Can I mark A as skipped, set

skip_on_upstream_skip

on B and C (or just B?) to False and then run the flow as usual? Or is there a more idiomatic way of doing this?

Currently I am doing this in the following way, which completely eliminates task A from the flow:

with Flow('name') as flow:
  upstream_tasks_b = []
  if not skip_a:
    a_result = A()()
    upstream_tasks_b.append(a_result)
  b_result = b()(
    upstream_tasks=upstream_tasks_b
  )
  c_result = c()(b_result)

I’m personally satisfied with the result, as it keeps flow clean depending on my config, but this can get wonky pretty quick. I’m as much interested about hearing how people approach this problem of flow customization.

@emre’s solution is probably the best if your flow is being built and run in a single script and you want the flow’s construction to depend on the script parameters. However we might also consider this from a Prefect perspective, in which we ideally only want to build a flow once and run it many times. In this case, you could

raise prefect.engine.signals.SKIP()

inside task A depending on a

Parameter

passed to it at runtime. A would mark itself as skipped and B or C could respond appropriately (through

skip_on_upstream_skip=False

, for example)