Is there a way to manually set upstream tasks in prefect core 1?

user_19603379 0
Viewed 80

I have a task that ensures some log files are empty (task4 below), but with my current flow it is running first as a root task even when I built logic into try to run it last:

with Flow('Pipeline') as flow:
    bool1 = task1.run()
    bool2 = task2.run()
    bool3 = task3.run()

    if bool1 and bool2 and bool3:
        task4.run()

So I get the sense that prefect doesn't care about the conditional logic. Is there a way to make tasks 1,2,3 upstream tasks for task4 ? The only way I can think of is introducing the bool variables as parameters to task4?

EDIT: https://github.com/PrefectHQ/prefect/blob/master/examples/conditional.py found this to be helpful

prefect etl python
1 Answers

there's a couple things to note here:

  • In general, you shouldn't need to call the .run() method of tasks within the flow context (i.e. within the Flow context manager)
  • You can use the case block to conditionally run tasks like this
  • In prefect 1.0, you cannot interact with the task result directly in the flow context. Instead, you can do something like this:
@task
def and(bools: Tuple[bool]) -> bool:
    return all(bools)

with Flow('Pipeline') as flow:
    bools = (
       task1(), 
       task2(),
       task3()
    )
    
    all_true = and(bools)

    with case(all_true, True):
       task4()

Note that you can also pass any task a kwarg called upstream_tasks as a list of task results to specify dependencies, like:

with Flow('Pipeline') as flow:
   result_a = task_a()
   task_b(upstream_tasks=[result_a])

I'd recommend checking out prefect>=2.0b10 and it's concepts, since it is more intuitive for those familiar with Python!

user_84679520
Related