Short answer: not today. In Prefect 3.x there isn’t a prune/remove-missing flag on

prefect deploy

. The deploy command will create/update what’s defined in your

prefect.yaml

, but it won’t delete deployments you’ve removed from that file. Ways to clean up: - Manually delete specific deployments:

prefect deployment delete <FLOW_NAME>/<DEPLOYMENT_NAME>

Or by ID:

prefect deployment delete --id <UUID>

- Delete everything, then redeploy (use with care):

prefect deployment delete --all
  prefect deploy --all

- Scripted “prune”: list current deployments, parse the names defined in your

prefect.yaml

, compute the difference, and call

prefect deployment delete

for the extras. I can share a small script if you’d like. References: - CLI: deployment commands - How-to: prefect.yaml - REST API: delete deployment If you want, tell me how you organize your

prefect.yaml

and I’ll draft a quick “prune missing deployments” script for your setup.

Great — here’s a safe “prune missing deployments” script for Prefect 3.x. It compares the deployment names declared in your

prefect.yaml

to what exists in your workspace, then deletes any workspace deployments whose names are no longer declared. By default it’s dry-run (no deletions). Add

--apply

to actually delete. ``` #!/usr/bin/env python3 import argparse import sys from pathlib import Path try: import yaml except ImportError: print("PyYAML is required. Install with: pip install pyyaml", file=sys.stderr) sys.exit(1) from prefect.client.orchestration import SyncPrefectClient def load_declared_deployments(prefect_file: Path) -> set[str]: """ Return the set of deployment names declared in prefect.yaml """ if not prefect_file.exists(): raise FileNotFoundError(f"Could not find {prefect_file}") data = yaml.safe_load(prefect_file.read_text()) or {} deployments = data.get("deployments", []) or [] # Deployments are a list of dicts; handle unexpected shapes defensively names = set() for idx, d in enumerate(deployments): if isinstance(d, dict): name = d.get("name") if name: names.add(str(name)) else: print(f"Warning: deployments[{idx}] missing 'name' - skipping", file=sys.stderr) else: print(f"Warning: deployments[{idx}] is not a mapping - skipping", file=sys.stderr) return names def main(): parser = argparse.ArgumentParser(description="Prune Prefect deployments missing from prefect.yaml") parser.add_argument("--prefect-file", default="prefect.yaml", help="Path to prefect.yaml (default: prefect.yaml)") parser.add_argument("--apply", action="store_true", help="Perform deletions. Without this, runs in dry-run mode.") parser.add_argument("--name-prefix", default=None, help="Only consider deployments whose names start with this prefix (safety filter).") parser.add_argument("--include-entrypoint-mismatches", action="store_true", help="Also delete deployments that share an entrypoint file with this project but have different names (advanced; conservative by default).") args = parser.parse_args() prefect_file = Path(args.prefect_file) try: declared_names = load_declared_deployments(prefect_file) except Exception as e: print(f"Error loading {prefect_file}: {e}", file=sys.stderr) sys.exit(1) if not declared_names: print(f"No deployments found in {prefect_file}. Nothing to compare against.") print("If this is unexpected, ensure your 'deployments:' list has entries with 'name' fields.") sys.exit(0) print(f"Declared deployment names in {prefect_file}:") for n in sorted(declared_names): print(f" - {n}") to_delete = [] kept = [] with SyncPrefectClient() as client: # Read all deployments in the current workspace # Note: read_deployments returns a list of deployment objects (pydantic models) deployments = client.read_deployments() # Build optional entrypoint map for advanced pruning declared_entrypoints = set() if args.include_entrypoint_mismatches: # Best-effort: pull entrypoints from YAML if present try: data = yaml.safe_load(prefect_file.read_text()) or {} for d in data.get("deployments", []) or []: if isinstance(d, dict): ep = d.get("entrypoint") if ep: declared_entrypoints.add(str(ep).lstrip("./")) except Exception: pass for dep in deployments: dep_id = getattr(dep, "id", None) dep_name = getattr(dep, "name", None) # Optional safety filter by name prefix if args.name_prefix and (not dep_name or not dep_name.startswith(args.name_prefix)):

continue # Determine if this deployment should be kept keep = False # 1) Keep anything with a name declared in prefect.yaml if dep_name in declared_names: keep = True # 2) Optionally, exclude from keep if it collides by entrypoint but different name # If include_entrypoint_mismatches is set, we will delete mismatched-name deployments dep_entrypoint = getattr(dep, "entrypoint", None) if keep is False and args.include_entrypoint_mismatches and dep_entrypoint: norm_ep = str(dep_entrypoint).lstrip("./") if norm_ep in declared_entrypoints: # This deployment shares the same entrypoint as a declared deployment, # but has a different name => treat as removable keep = False if keep: kept.append((dep_id, dep_name, dep_entrypoint)) else: # Candidate for deletion if we are not keeping it to_delete.append((dep_id, dep_name, dep_entrypoint)) print("\nKept (present in prefect.yaml):") for dep_id, dep_name, dep_entrypoint in sorted(kept, key=lambda x: (x[1] or "", str(x[0]))): print(f" - {dep_name} (id={dep_id}, entrypoint={dep_entrypoint})") print("\nCandidates to delete (NOT declared in prefect.yaml):") if not to_delete: print(" - None. You're in sync!") sys.exit(0) for dep_id, dep_name, dep_entrypoint in sorted(to_delete, key=lambda x: (x[1] or "", str(x[0]))): print(f" - {dep_name} (id={dep_id}, entrypoint={dep_entrypoint})") if not args.apply: print("\nDry-run: no deletions performed. Re-run with --apply to delete the candidates above.") sys.exit(0) # Perform deletions errors = 0 with SyncPrefectClient() as client: for dep_id, dep_name, dep_entrypoint in to_delete: if not dep_id: print(f"Skipping {dep_name}: missing deployment id", file=sys.stderr) errors += 1 continue try: client.delete_deployment_by_id(dep_id) print(f"Deleted: {dep_name} (id={dep_id})") except Exception as e: print(f"Failed to delete {dep_name} (id={dep_id}): {e}", file=sys.stderr) errors += 1 if errors: print(f"\nPrune finished with {errors} error(s).", file=sys.stderr) sys.exit(1) else: print("\nPrune complete.") sys.exit(0) if name == "__main__": main()

How to use
- Preview (dry-run):

python prune_missing_deployments.py

- Actually delete:

python prune_missing_deployments.py --apply

- Extra safety: limit by name prefix (useful if your deployments use a consistent prefix):

python prune_missing_deployments.py --name-prefix myproj- --apply

- Advanced: also delete deployments that share an entrypoint with your project but have different names:

python prune_missing_deployments.py --include-entrypoint-mismatches --apply ``` Notes and references - This script uses SyncPrefectClient methods consistent with Prefect 3.x: - Listing:

read_deployments()

- Deleting by ID:

delete_deployment_by_id(...)

- Docs: - CLI: deployment commands - REST: read deployments - REST: delete deployment - prefect.yaml reference Want a version that prunes by comparing both flow name and deployment name derived from your entrypoints? If you share your

prefect.yaml

(or how your flows are structured), I can tweak this to be even stricter.