Yes — you can do this with a Prefect flow that: 1. clones or downloads the git repo containing

prefect.yaml

2. parses the deployments listed in that file 3. queries your self-hosted Prefect Server for currently registered deployments 4. finds deployments that exist in the server but are not defined in

prefect.yaml

5. alerts by logging, failing the flow, and/or emitting a Prefect event For Prefect 3.6.29, the safest comparison key is:

<flow_name>/<deployment_name>

because Prefect deployments are typically addressed that way. Here’s a complete example. ```python from future import annotations import shutil import subprocess import tempfile from pathlib import Path from typing import Any import yaml from prefect import flow, task, get_run_logger from prefect.client.orchestration import get_client from prefect.cli.deploy._models import PrefectYamlModel from prefect.events import emit_event @task def clone_repo(repo_url: str, branch: str = "main", subdirectory: str | None = None) -> Path: logger = get_run_logger() temp_dir = Path(tempfile.mkdtemp(prefix="prefect-yaml-audit-")) logger.info(f"Cloning repo {repo_url} into {temp_dir}") subprocess.run( ["git", "clone", "--depth", "1", "--branch", branch, repo_url, str(temp_dir)], check=True, capture_output=True, text=True, ) repo_path = temp_dir / subdirectory if subdirectory else temp_dir if not repo_path.exists(): raise FileNotFoundError(f"Repository path does not exist: {repo_path}") return repo_path @task def load_prefect_yaml_deployments(repo_path: Path, prefect_yaml_path: str = "prefect.yaml") -> set[str]: logger = get_run_logger() yaml_file = repo_path / prefect_yaml_path if not yaml_file.exists(): raise FileNotFoundError(f"Could not find {prefect_yaml_path} at {yaml_file}") logger.info(f"Loading deployment definitions from {yaml_file}") with yaml_file.open("r") as f: raw = yaml.safe_load(f) model = PrefectYamlModel.model_validate(raw) deployment_keys = set() for dep in model.deployments or []: if not dep.name or not dep.flow_name: logger.warning(f"Skipping deployment with missing name or flow_name: {dep}") continue deployment_keys.add(f"{dep.flow_name}/{dep.name}") logger.info(f"Found {len(deployment_keys)} deployments defined in prefect.yaml") return deployment_keys @task async def load_server_deployments() -> list[dict[str, Any]]: logger = get_run_logger() async with get_client() as client: deployments = await client.read_deployments() results = [] for dep in deployments: key = f"{dep.flow_name}/{dep.name}" results.append( { "key": key, "id": str(dep.id), "flow_name": dep.flow_name, "deployment_name": dep.name, "work_pool_name": dep.work_pool_name, "paused": dep.paused, "version": dep.version, "updated": dep.updated.isoformat() if dep.updated else None, } ) logger.info(f"Found {len(results)} deployments registered in Prefect Server") return results @task def find_orphaned_deployments( yaml_keys: set[str], server_deployments: list[dict[str, Any]], ) -> list[dict[str, Any]]: logger = get_run_logger() orphaned = [dep for dep in server_deployments if dep["key"] not in yaml_keys] if orphaned: logger.warning(f"Found {len(orphaned)} deployments in server that are not in prefect.yaml") for dep in orphaned: logger.warning( f"Orphaned deployment: {dep['key']} " f"(id={dep['id']}, work_pool={dep['work_pool_name']}, paused={dep['paused']})" ) else: logger.info("No orphaned deployments found") return orphaned @task

def emit_orphaned_events(orphaned: list[dict[str, Any]]) -> None: logger = get_run_logger() for dep in orphaned: emit_event( event="deployment-audit.orphaned", resource={ "prefect.resource.id": f"prefect.deployment.{dep['id']}", "prefect.resource.name": dep["key"], }, payload=dep, ) logger.info(f"Emitted event for orphaned deployment: {dep['key']}") @task def cleanup_repo(repo_path: Path) -> None: root = repo_path while root.parent != root and not root.name.startswith("prefect-yaml-audit-"): root = root.parent if root.exists() and root.name.startswith("prefect-yaml-audit-"): shutil.rmtree(root, ignore_errors=True) @flow(name="audit-prefect-deployments-against-prefect-yaml") async def audit_prefect_yaml_deployments( repo_url: str, branch: str = "main", prefect_yaml_path: str = "prefect.yaml", subdirectory: str | None = None, fail_if_orphaned: bool = False, emit_events: bool = True, ) -> list[dict[str, Any]]: logger = get_run_logger() repo_path = clone_repo(repo_url=repo_url, branch=branch, subdirectory=subdirectory) try: yaml_keys = load_prefect_yaml_deployments(repo_path, prefect_yaml_path) server_deployments = await load_server_deployments() orphaned = find_orphaned_deployments(yaml_keys, server_deployments) if orphaned and emit_events: emit_orphaned_events(orphaned) if orphaned: logger.warning(f"Deployment audit failed: {len(orphaned)} orphaned deployment(s) found") if fail_if_orphaned: raise RuntimeError( f"Found {len(orphaned)} deployment(s) in Prefect Server that are not present in prefect.yaml" ) else: logger.info("Deployment audit passed") return orphaned finally: cleanup_repo(repo_path)

*How it works*

- `clone_repo` pulls your git repo containing `prefect.yaml`
- `load_prefect_yaml_deployments` parses `prefect.yaml` using Prefect’s internal `PrefectYamlModel`
- `load_server_deployments` queries your Prefect Server with `get_client()` and `read_deployments()`
- `find_orphaned_deployments` compares server deployments against the YAML-defined ones
- `emit_orphaned_events` emits a custom event you can use for automations
- the flow optionally fails if any orphaned deployments are found

*How to run it*

Make sure these are set in the environment where the flow runs:

```bash
export PREFECT_API_URL="<http://your-prefect-server:4200/api>"

If your self-hosted setup requires auth, also set the relevant auth environment variables for that environment. Then run the flow like:

python
if __name__ == "__main__":
    import asyncio

    asyncio.run(
        audit_prefect_yaml_deployments(
            repo_url="<https://github.com/your-org/your-repo.git>",
            branch="main",
            prefect_yaml_path="prefect.yaml",
            fail_if_orphaned=True,
            emit_events=True,
        )
    )

Important caveat This assumes the

prefect.yaml

entries include

flow_name

. If your

prefect.yaml

relies on values inferred during deploy-time and does not explicitly include

flow_name

, then matching becomes less reliable. In that case, I’d recommend either: - ensuring each deployment in

prefect.yaml

includes

flow_name

, or - matching by

entrypoint + deployment name

instead Alerting options You have a few choices: - Simplest: set

fail_if_orphaned=True

and put this on a schedule - Better: use

emit_events=True

and create an automation that listens for

deployment-audit.orphaned

- Also useful: send Slack/email/webhook directly from a task if you want immediate notification If you want, I can also give you: 1. a version that does not clone git directly and instead reads from a local checked-out repo

2. a version that sends a Slack webhook alert 3. a version that also detects deployments in prefect.yaml that are missing from the server 4. a deployment for this audit flow using

flow.from_source(...).deploy(...)

Related API areas used here: - Prefect client orchestration:

prefect.client.orchestration.get_client()

- Prefect YAML model:

prefect.cli.deploy._models.PrefectYamlModel

- Prefect events:

prefect.events.emit_event

If you want, I can turn this into a production-ready version with Slack alerting and a deployment spec next.