Prefect 2.0 | Accessing current flow state Hi! :wave: Possibly simple question which I can’t seem to...

Dominik Wagner

07/26/2022, 1:28 PM

Prefect 2.0 | Accessing current flow state Hi! 👋 Possibly simple question which I can’t seem to find an answer for: Can I access the state of the current flow from within that flow? In particular I’d like to check if the flow is retrying, and if so pass an extra argument to my task(s) i.e.

Copy code

@flow
def my_flow():
  if <somehow_get_state_of_this_flow?> == "Retrying":
    do_something()

✅ 1

Anna Geller

07/26/2022, 1:32 PM

when running tasks, you can do:

Copy code

if task.get_state().is_failed():

Anna Geller

07/26/2022, 1:32 PM

or rather:

Copy code

if task.get_state().is_retrying():

Anna Geller

07/26/2022, 1:33 PM

this assumes that task is submitted to task runner

Anna Geller

07/26/2022, 1:34 PM

Copy code

@task
def do_sth():
   ...

@flow
def my_flow():
  if do_sth.submit().get_state().is_retrying():
    do_something()

Dominik Wagner

07/26/2022, 1:37 PM

Thanks, but maybe it wasn’t clear, I’m retrying the flow, not a task (added in 2.0b8). All my tasks are using

shell_run_command()

from

prefect_shell

, and I don’t think I can add retries there?

Anna Geller

07/26/2022, 1:38 PM

for flow there is no way currently, but will be after GA release

🦜 1

Anna Geller

07/26/2022, 1:39 PM

for 2.0b8, this syntax works for tasks https://discourse.prefect.io/t/how-to-take-action-on-a-state-change-of-a-task-run-task-level-state-handler/82

Dominik Wagner

07/26/2022, 1:41 PM

Okay, thanks! Will hold off until tomorrow then 😁

🙏 1

🙌 1

2 Views

Open in Slack

Previous Next

Prefect Community

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.