Hi! I'm trying to find option similar to

depends_on_past

in airflow. We wanted next run of flow to start only when previous run is in Success state. Can anyone help?

@Suresh R you can achieve that (any many other state-related patterns) with a state handler: • this one will start a next flow run as soon as the previous flow run ended with Success • this one will skip any next flow run if there is any other flow run in progress of this flow

@Anna Geller Thanks,This is helpful!! I have one more query. We have option in prefect to to rerun a failed flow. Do we have option to clear state of succeeded task and rerun it.

Yes, on the task run page you can set a state of a task run.

@Anna Geller I am getting this error when trying to restart the run after setting it to failed state, is this due to my agent lost information of the task.

prefect.exceptions.ClientError: [{'path': ['get_or_create_task_run_info'], 'message': 'Expected type UUID!, found ""; Could not parse UUID:

Are you using some env variables when you register your flows? This error happened to one user when he was using env variables during flow registration, but those env variables didn’t exist on the agent during flow runtime. I would definitely cross-check if your registration environment and what you have on your agent matches.

ohh Let me check

no, you don’t. The flow run history and flow’s metadata is stored in the backend, not on the agent.

ok Thanks Again!!!

wait, do you want to only schedule this flow to run once per day, or do you want it to be running constantly one after the other?

I want to schedule it at 8AM (Once per day) and don't want to run it next day if it failed previous day.

ok, then your state handler must be different: it needs to query the flow runs in the descending order, limit to 1 and check if this state is a Success. Then for scheduling, you could use e.g. CronClock:

from prefect.schedules import Schedule
from prefect.schedules.clocks import CronClock

schedule = Schedule(clocks=[CronClock("0 8 * * *")])

print(schedule.next(5))

Can you point some examples of querying flow runs

Ok Got it

@Suresh R are you on Cloud? I was looking at this state handler and remembered that we have an Automation exactly for that use case 🤦‍♀️ here is how it works: if any flow run from YOUR FLOW NAME changes its state to Failed, then pause its schedule - which has an effect of blocking any future runs if any run ends in a Failed state.

@Anna Geller Great, We are on cloud, Automation will be easy for us. Thank you!!!