Tony Yun
10/14/2021, 6:36 PMTony Yun
10/14/2021, 6:36 PMAnna Geller
from prefect.backend import set_key_value
key_value_uuid = set_key_value(key="fail_or_not", value="fail")
Then failing the flow based on the value from KV store:
from prefect import task, Flow
from prefect.backend import get_key_value
@task(log_stdout=True)
def fail_if_key_say_so():
value = get_key_value(key="fail_or_not")
if value == "fail":
raise ValueError("Failing the task...")
print("Success!")
with Flow("fail_based_on_key") as flow:
fail_if_key_say_so()
Changing the value should result in success upon restart:
from prefect.backend import set_key_value
key_value_uuid = set_key_value(key="fail_or_not", value="not")
But similarly to you, I had to click twice for the restart to take effect. There is a caveat though. This happens ONLY when the original flow is triggered through “Quick Run” in the UI. When I trigger the flow run from CLI or when the flow runs on schedule, the Restart button in the UI triggers a restart immediately.Kevin Kho
Marvin
10/14/2021, 7:50 PM