@Marvin can you show me an example of cancelling a flow via the API? Do I need to set state to Cancelled or Cancelling?

To cancel a flow run using Prefect, you can use either the Prefect CLI or the UI. 1. Cancel via the CLI: You can cancel a flow run from the command line in your execution environment with the following command, where you'll need to pass the ID of the flow run:

bash
   prefect flow-run cancel 'FLOW_RUN_ID'

Make sure to replace

'FLOW_RUN_ID'

with the actual ID of your flow run. 2. Cancel via the UI: Alternatively, you can cancel a flow run from the Prefect UI. To do this, navigate to the flow run's detail page and click the "Cancel" button located in the upper right corner. Regarding setting the state to "Cancelled" or "Cancelling," when you initiate the cancellation process as described above, Prefect will handle the transition to the appropriate state on your behalf. When you request cancellation, Prefect will likely move the flow run into the "Cancelling" state first, and once the cancellation completes, it will be marked as "Cancelled." The mention of

PREFECT_EXPERIMENTAL_ENABLE_ENHANCED_CANCELLATION=True

indicates that there may be an experimental feature related to cancellation. Ensure to set this configuration if enhanced cancellation features are needed. If you encounter any issues, it is suggested to reach out for support in the Prefect Slack or by creating a Github issue.

I do something like this if this helps, I couldn't find something in API to do it/had to utilize the CLI/SubProcess Python library

Thanks! I pulled up CLI code and it looks like it sets flow run state to Cancelling which I think we can do via the API: https://github.com/PrefectHQ/prefect/blob/main/src/prefect/cli/flow_run.py#L133

I think that just 'sets'/writes that to DB: https://docs.prefect.io/latest/api-ref/server/models/flow_runs/#prefect.server.models.flow_runs.set_flow_run_state I could be totally wrong, lemme know if it works, because I might switch my code to use that 😛

The function

prefect.server.models.flow_runs.set_flow_run_state

appears to be related to changing the state of a flow run within Prefect's server. In Prefect 2.x, when a flow run cancellation is requested via the CLI, UI, REST API, or Python client, the flow run is moved to a "Cancelling" state. This indicates that the cancellation process involves more than just updating the database; the agent actively monitors the state of flow runs and, upon detecting a cancellation request, sends a signal to terminate the run's infrastructure. This means that the actual cancellation of a flow run involves coordination with the agent to ensure that the run is terminated. The

set_flow_run_state

function then likely facilitates the change of the flow run state in the database to reflect this process. The Prefect agent and infrastructure work together to guarantee that the flow run is either gracefully shut down or forcefully terminated if it does not comply within a given grace period. The cancellation process is robust and ensures flow runs are halted as requested; this process requires that an agent is running to enforce the cancellation. This behavior suggests that

set_flow_run_state

would be part of the process to update the state in the system but not the only part of the cancellation mechanism.

Not sure the answer, maybe marvin will, but fwiw that's what your code is doing now because that's what the CLI does so I'd assume it works

Here's my dialogue with robot from 30 days ago, I think I tested set_flow_run_state, and it didn't cancel fully, but I can't remember 😕 https://prefect-community.slack.com/archives/C04DZJC94DC/p1699889019869549

Please note that this method will only set the state of the flow run in the Prefect database. It will not send a signal to the executing job to stop. The job will continue to run until completion, but its results will not be reported back to Prefect because the flow run state is set to Cancelled.

prefect.server.models.flow_runs.set_flow_run_state

is the server side function. You're looking for:

await client.set_flow_run_state(
    flow_run_id=id, state=cancelling_state
)

which will send the api call (which you can do w/o the python client if you'd like). The proper thing to do is set it to

Cancelling

and let the agent/worker perform any teardown before it will set it to

Cancelled

Thanks Jake 🙂 Learning new stuff every day