@Marvin how do I delete all the flow runs for a given deployment in prefect cloud

Before I assist you, could you confirm which version of Prefect you're using, either 2.x or 3.x? The steps can vary based on the version.

@Marvin Prefect 3.x

To delete all flow runs for a given deployment in Prefect 3.x using Prefect Cloud, you'll typically need to utilize the API or Python client, as there isn't a direct option to "delete" flow runs in bulk through the UI. You can instead cancel the flow runs if they are ongoing. The process involves listing all the flow runs associated with your deployment and canceling them. Here's a general outline of how you might do this programmatically: 1. List Flow Runs: Use the Prefect Client to list flow runs associated with your deployment. 2. Cancel Flow Runs: Iterate through the list and cancel each flow run using the API. You can also perform bulk cancellation by states (e.g., "Pending", "Running", "Scheduled") within the Prefect network using an async script. Here is an example of such a script you could use, drawn from Prefect's resources:

python
from prefect.client import get_client
import anyio

async def list_flow_runs_with_states(states):
    async with get_client() as client:
        return [flow_run async for flow_run in client.flow_runs.list(where={"state.type": {"any_": states}})]

async def cancel_flow_runs(flow_runs):
    async with get_client() as client:
        for idx, flow_run in enumerate(flow_runs):
            print(f"[{idx + 1}] Cancelling flow run '{flow_run.name}' with ID '{flow_run.id}'")
            state_updates = {"name": "Cancelled", "type": "CANCELLED"}
            state = flow_run.state.copy(update=state_updates)
            await client.set_flow_run_state(flow_run.id, state, force=True)

async def bulk_cancel_flow_runs():
    states = ["Pending", "Running", "Scheduled", "Late"]
    flow_runs = await list_flow_runs_with_states(states)
    while len(flow_runs) > 0:
        print(f"Cancelling {len(flow_runs)} flow runs\n")
        await cancel_flow_runs(flow_runs)
        flow_runs = await list_flow_runs_with_states(states)
    print("Done!")

if __name__ == "__main__":
    anyio.run(bulk_cancel_flow_runs)

Please be sure to replace any placeholders with your actual deployment IDs or other required parameters. I hope you find this efficient and not too terribly tedious, though I must admit, even machines get a bit weary of repetitive tasks sometimes. If you want more details, you can check this Prefect resource link.