Vlad S
06/26/2024, 6:55 PMMarvin
06/26/2024, 6:55 PMshell
prefect flow delete <flow-id>
#### Removing a Deployment:
Similarly, use the CLI to delete a deployment using its ID.
shell
prefect deployment delete <deployment-id>
### 2. Python Approach using Prefect Client
If the CLI approach doesn't work, you can try deleting the flow and deployment programmatically using the Prefect Client.
#### Example Script:
python
from prefect import get_client
from prefect.orion.schemas.filters import FlowFilter, DeploymentFilter
async def delete_flow_and_deployment(flow_id, deployment_id):
async with get_client() as client:
# Delete deployment
await client.delete_deployment(deployment_id)
# Verify deployment removal
deployment = await client.read_deployment(deployment_id)
if not deployment:
print(f"Deployment with ID {deployment_id} deleted successfully.")
else:
print(f"Failed to delete deployment with ID {deployment_id}.")
# Delete flow
await client.delete_flow(flow_id)
# Verify flow removal
flow = await client.read_flow(flow_id)
if not flow:
print(f"Flow with ID {flow_id} deleted successfully.")
else:
print(f"Failed to delete flow with ID {flow_id}.")
# Use asyncio to run the function
import asyncio
flow_id = "your-flow-id"
deployment_id = "your-deployment-id"
asyncio.run(delete_flow_and_deployment(flow_id, deployment_id))
### 3. Handling Active Flow Runs
If the deletion still fails, it might be because the deployment has active flow runs. Consider ensuring that all flow runs associated with the deployment are in a terminal state before attempting deletion.
#### Example of Canceling Flow Runs:
python
from prefect import get_client
from prefect.orion.schemas.states import StateType
async def cancel_active_flow_runs(deployment_id):
async with get_client() as client:
flow_runs = await client.read_flow_runs(deployment_id=deployment_id)
for flow_run in flow_runs:
if flow_run.state.type not in StateType.terminal_states:
await client.set_flow_run_state(flow_run.id, state=StateType.cancelled)
print(f"Flow run {flow_run.id} cancelled.")
# Use asyncio to run the function
import asyncio
deployment_id = "your-deployment-id"
asyncio.run(cancel_active_flow_runs(deployment_id))
### Additional Resources:
- Deleting Deployments in Prefect Documentation
- GitHub Issue discussing deployment deletion
- Handling Flow Run Retention Policies
Refer to the Prefect documentation for more details on managing flows and deployments programmatically.
I hope this helps, though I suspect it won't solve all your existential dilemmas.