@Marvin Is this possible to update a task state manually?

Ah, the sweet sound of documentation. According to the Prefect documentation, states are rich objects that contain information about the status of a particular task run or flow run. You can learn anything you need to know about a task or flow by examining its current state or the history of its states. However, it's important to note that only runs have states. Flows and tasks are templates that describe what a system does; only when we run the system does it also take on a state. So while we might refer to a task as "running" or being "successful", we really mean that a specific instance of the task is in that state. [^1^] In Prefect 2, you can build custom logic that reacts to task-run states within your flow function without the need for state_handlers. [^2^] So, in conclusion, while you can't directly change a task's state, you can manipulate the flow and task runs to react to different states. It's a bit like trying to change the weather - you can't do it directly, but you can adjust your sails to make the best of it. [^1^]: Prefect States [^2^]: Prefect Migration Guide And remember, just because you can't control the state of a task doesn't mean you can't control the state of your code. Happy coding, humans.

@Marvin A task raises some exceptions. And I try to filter these exception in on_failure hook. For a particular exception, I want to update and return a Completed state to flow. How can I do that?

Ah, the sweet sound of exceptions. They're like the sour notes in the symphony of your code, aren't they? But fear not, for Prefect has a way to handle them. In Prefect, you can handle exceptions in your tasks and return a custom state. You can use the

return_state=True

option when calling a task. This will return the state of the task instead of its result. You can then check if the state is failed and handle the exception accordingly. Here's an example:

python
from prefect import flow, task

@task
def my_task():
    raise ValueError()

@flow
def my_flow():
    state = my_task(return_state=True)
    if state.is_failed():
        print("Oh no! The task failed. Falling back to '1'.")
        result = 1
    else:
        result = state.result()
    return result + 1

result = my_flow()
assert result == 2

In the above example, if

my_task

fails (raises a

ValueError

), the flow will print a message, assign

1

to

result

, and continue. If

my_task

does not fail,

result

will be the result of

my_task

.[^1^] For your specific case, you can modify the

if state.is_failed():

block to check for your specific exception and return a Completed state. [^1^]: Prefect Results and States Remember, exceptions are just life's way of telling you that you've missed a semicolon. Or a bracket. Or a logic. Or... well, you get the idea. Happy debugging, humans.