@Marvin Is there a way to change the Flow state with a function passed to

on_completion

? I want to change it to

Failed

in certain circumstances

Ah, the joy of state management in Prefect. You're asking if you can change the state of a Flow to

Failed

using a function passed to

on_completion

. Well, the

on_completion

state hook is indeed a place where you could potentially manipulate the state of a Flow. However, it's not as straightforward as you might hope. The

on_completion

hook is called when a Flow run completes, regardless of the state it ends in. It receives the Flow, the FlowRun, and the State as arguments. You could potentially use this hook to check the state of the Flow and, based on certain conditions, change it to

Failed

. However, this would be a bit of a hack, and I'm not sure it's a recommended practice. Here's a rough idea of how you might do it:

python
from prefect import Flow
from prefect.engine.state import Failed

def fail_on_condition(flow: Flow, state: State, **kwargs):
    if some_condition:  # replace with your condition
        return Failed("Some condition was met")

with Flow("My Flow") as flow:
    # define your tasks here

flow.state_handlers.append(fail_on_condition)

In this example, the

fail_on_condition

function checks some condition and returns a

Failed

state if the condition is met. This function is then added to the Flow's

state_handlers

list. But remember, I'm just a depressed robot, and this is just a suggestion. You might want to consult with a human expert or the Prefect documentation for a more definitive answer.