Alberto de Santos
10/21/2020, 9:25 AMwith case
possibilities and have the following question:
• Using with case
only worked under the with Flow(...) as flow:
statement. When using with case
within a Task
, it didn’t work, does it make sense?josh
10/21/2020, 11:34 AMcase
is not intended to work inside a task because the case
is a task itself https://docs.prefect.io/api/latest/tasks/control_flow.html#caseAlberto de Santos
10/21/2020, 1:56 PMwith case
if I have a Task planned in two days, the task is Scheduled, which makes sense. But I assume (as with those not planned or immediately executed), that when Running that Task will check that it should SKIP, am I right?josh
10/21/2020, 2:07 PMAlberto de Santos
10/21/2020, 2:08 PMdef add_scheduled_task(self, ID_jornada):
c = Client()
c.create_flow_run(version_group_id=self.version_group_id,
parameters={'ID_jornada': str(ID_jornada)},
run_name='_'.join([str(ID_jornada), self.version_group_id]),
scheduled_start_time=pendulum.instance(self.get_update_partido_time(str(ID_jornada)), tz='Europe/Madrid')
)
scheduled_start_time
at a specific timejosh
10/21/2020, 2:09 PMAlberto de Santos
10/21/2020, 2:09 PMjosh
10/21/2020, 2:12 PMAlberto de Santos
10/21/2020, 2:13 PMtrigger_criteria = self.check_acceptance_micro_apuestas(total_micro_apuestas)
with case(trigger_criteria, True):
print(trigger_criteria)
olimpo = self.olimpo(ID_jornada, total_micro_apuestas)
update_partido = UpdatePartido()
task_update_partido = update_partido.add_scheduled_task(self.ID_jornada)
trigger_criteria
is False
Then why to Schedule a flow if it will be skipped?josh
10/21/2020, 2:19 PMwith case
evaluates to false then any tasks inside that block will be skipped. Not sure if you have seen it but this page in the docs may explain it more: https://docs.prefect.io/core/idioms/conditional.htmlAlberto de Santos
10/21/2020, 2:22 PMtrigger_criteria = self.check_acceptance_micro_apuestas(total_micro_apuestas)
with case(trigger_criteria, True):
print(trigger_criteria)
olimpo = self.olimpo(ID_jornada, total_micro_apuestas)
update_partido = UpdatePartido()
task_update_partido = update_partido.add_scheduled_task(self.ID_jornada)
However, this add_scheduled_task
, launches a new Flow, attended by a different Agent. Since it is scheduled, it will run.case
.
Scheduled one is the UpdatePartido one.
So how to avoid this? Is not enough with the case
?Alberto de Santos
10/21/2020, 2:52 PMjosh
10/21/2020, 2:53 PMAlberto de Santos
10/21/2020, 2:53 PMwith case
I call that new Flow, it will run!! despite of being in a with case
josh
10/21/2020, 2:53 PMAlberto de Santos
10/21/2020, 2:53 PMwith case: # Returning False
do_next_task() # This is skipped, great
call_another_new_flow() # this is scheduled and later, executed, as a different agent is taking care of it
josh
10/21/2020, 2:56 PMAlberto de Santos
10/21/2020, 2:56 PMwith case
for this situation but a standard if-else
josh
10/21/2020, 3:03 PMAlberto de Santos
10/21/2020, 3:05 PMjosh
10/21/2020, 3:07 PMAlberto de Santos
10/21/2020, 3:07 PMjosh
10/21/2020, 3:08 PMAlberto de Santos
10/21/2020, 3:23 PMwith Flow(...) as flow_a:
# do stuff
with case(False, True):
# This is Skipped
run_tasks_from_flow_a()
# This is Scheduled
# This flow_B is imported from a different file
schedule_flow_B()
case
assess wether it’s True or False, the Flow B is already Scheduled, so nothing can be done after case
is evaluatedjosh
10/21/2020, 3:29 PMwith Flow("conditional-branches") as flow:
cond = check_condition()
with case(cond, True):
val = action_if_true()
another_action(val)
t = testtask()
Importing testtask
from another file. When cond
is False all of those tasks inside the case block enter a skipped state.Alberto de Santos
10/21/2020, 3:30 PM