Get Prefect Deployment Summary

deployment_data.py

from typing import Optional

import pandas as pd
from prefect.client.orchestration import get_client
from pydantic import BaseModel

columns = {
    "flow_name": "Flow",
    "deployment_name": "Deployment",
    "schedule": "Schedule",
    "active": "Active",
    "work_pool": "Work Pool",
    "work_queue": "Work Queue",
    "infra_type": "Infrastructure Type",
    "image": "Image",
    "entrypoint": "Entrypoint",
    "tags": "Tags",
}


class DepData(BaseModel):
    flow_name: str
    deployment_name: str
    schedule: Optional[str]
    active: bool
    work_pool: str
    work_queue: str
    infra_type: str
    image: Optional[str]
    entrypoint: str
    tags: str


async with get_client() as client:
    deployments = await client.read_deployments()
    infra_ids = list(set(dep.infrastructure_document_id for dep in deployments))
    flow_ids = list(set(dep.flow_id for dep in deployments))
    infras = {infra_id: await client.read_block_document(infra_id) for infra_id in infra_ids}
    flows = {flow_id: await client.read_flow(flow_id) for flow_id in flow_ids}

data = []
for dep in deployments:
    flow = flows[dep.flow_id]
    infra = infras[dep.infrastructure_document_id]

    schedule = None
    if dep.schedule is not None:
        schedule = dep.schedule.cron

    image = None
    if infra.block_type.name == "Docker Container":
        image = infra.data["name"]

    data.append(
        DepData(
            flow_name=flow.name,
            deployment_name=dep.name,
            schedule=schedule,
            active=dep.is_schedule_active,
            work_pool=dep.work_pool_name,
            work_queue=dep.work_queue_name,
            infra_type=infra.block_type.name,
            image=image,
            entrypoint=dep.entrypoint,
            tags=";".join(dep.tags),
        )
        .dict()
    )

df = pd.DataFrame(data)
df.rename(columns=columns).to_csv("deployment_summaries.csv", index=False, quoting=1)