<@ULVA73B9P> How should i manage deployment schedu...
# ask-marvin
l
@Marvin How should i manage deployment schedules using the pythhon sdk? imagine that i want to programmatically add and remove scheduels from a deployment
m
thought for 291.7 seconds
Great question! In Prefect 3.x you manage deployment schedules through the orchestration client. You can programmatically add (create), list, disable/update, and remove schedules for any deployment. Here’s a concise end-to-end flow using the Python SDK. Links to reference: - Schedules models: prefect.client.schemas.schedules - Deployment client methods: _deployments client (create/read/update/delete schedules) Assumptions: Prefect 3.x and async code (I included a sync variant too). Async example (add, list, delete)
Copy code
import asyncio
from datetime import timedelta
from prefect.client.orchestration import get_client
from prefect.client.schemas.schedules import CronSchedule, IntervalSchedule

FLOW_DEPLOYMENT_NAME = "my_flow/my_deployment"  # format: <FLOW_NAME>/<DEPLOYMENT_NAME>

async def main():
    async with get_client() as client:
        # 1) Look up the deployment ID
        deployment = await client.read_deployment_by_name(FLOW_DEPLOYMENT_NAME)
        deployment_id = deployment.id

        # 2) Add schedules: pass a list of tuples (schedule_obj, is_active)
        created = await client.create_deployment_schedules(
            deployment_id=deployment_id,
            schedules=[
                (CronSchedule(cron="0 2 * * *", timezone="UTC"), True),             # daily 02:00 UTC
                (IntervalSchedule(interval=timedelta(hours=6)), False),             # disabled by default
            ],
        )
        print(f"Created {len(created)} schedules")

        # 3) List schedules
        schedules = await client.read_deployment_schedules(deployment_id)
        for s in schedules:
            print(f"- {s.id} active={s.active} type={type(s.schedule).__name__}")

        # 4) Remove a schedule (by id)
        if schedules:
            to_delete = schedules[0].id
            await client.delete_deployment_schedule(deployment_id, to_delete)
            print(f"Deleted schedule {to_delete}")

        # Optional: disable (instead of deleting)
        # if schedules:
        #     await client.update_deployment_schedule(deployment_id, schedules[0].id, active=False)

asyncio.run(main())
Sync variant
Copy code
from datetime import timedelta
from prefect.client.orchestration import get_client
from prefect.client.schemas.schedules import CronSchedule, IntervalSchedule

with get_client(sync_client=True) as client:
    dep = client.read_deployment_by_name("my_flow/my_deployment")
    client.create_deployment_schedules(
        deployment_id=dep.id,
        schedules=[
            (CronSchedule(cron="0 9 * * MON-FRI", timezone="America/New_York"), True),
            (IntervalSchedule(interval=timedelta(days=1)), True),
        ],
    )
    for s in client.read_deployment_schedules(dep.id):
        print(s.id, s.active, type(s.schedule).__name__)
    # Delete one
    # client.delete_deployment_schedule(dep.id, some_schedule_id)
Notes and tips - The SDK methods you’ll use: -
read_deployment_by_name("<FLOW>/<DEPLOYMENT>")
→ get the deployment and its
id
-
create_deployment_schedules(deployment_id, schedules=[(schedule_obj, active_bool), ...])
-
read_deployment_schedules(deployment_id)
-
update_deployment_schedule(deployment_id, schedule_id, active=..., schedule=...)
-
delete_deployment_schedule(deployment_id, schedule_id)
- Supported schedule types:
CronSchedule
,
IntervalSchedule
,
RRuleSchedule
(and
NoSchedule
). - Removal requires a schedule ID; list schedules first to obtain it. - If you want to disable schedules without deleting them, set
active=False
via
update_deployment_schedule
. Want me to tailor this for your exact flow/deployment names and a specific schedule (cron/interval/rrule)? Also, do you prefer async or sync code?
l
is it possible to add a slug? ui allows it . @Marvin
m
thought for 309.6 seconds
Short answer: Yes, but not with the high-level helper you used. The UI writes a slug on schedules; you can do the same from Python, you just need to call the API with the full schedule payload. Details: - The convenient helper
create_deployment_schedules(deployment_id, schedules=[(schedule_obj, active_bool), ...])
only accepts tuples and only sends schedule+active. It cannot set slug/parameters/max_scheduled_runs. - Slug, parameters, and max_scheduled_runs are supported by the underlying API models: - DeploymentScheduleCreate: schedule, active, slug, parameters, max_scheduled_runs - DeploymentScheduleUpdate: schedule, active, slug, parameters, max_scheduled_runs - The SDK’s update helper exposes only active and schedule; to set slug/parameters/max_scheduled_runs you need to use a lower-level request. Two workable patterns: A) Create schedules with slug via a direct API call using the client
Copy code
from uuid import UUID
from datetime import timedelta
from prefect.client.orchestration import get_client
from prefect.client.schemas.actions import DeploymentScheduleCreate
from prefect.client.schemas.schedules import CronSchedule, IntervalSchedule

with get_client(sync_client=True) as client:
    deployment_id = UUID("your-deployment-id")

    create_payload = [
        DeploymentScheduleCreate(
            schedule=CronSchedule(cron="0 9 * * *", timezone="UTC"),
            active=True,
            slug="daily-morning",
            parameters={"env": "prod"},
            max_scheduled_runs=365,
        ).model_dump(mode="json"),
        DeploymentScheduleCreate(
            schedule=IntervalSchedule(interval=timedelta(hours=6)),
            active=False,
            slug="every-6-hours",
            parameters={"fast": True},
        ).model_dump(mode="json"),
    ]

    resp = client.request(
        "POST",
        "/deployments/{id}/schedules",
        path_params={"id": deployment_id},
        json=create_payload,
    )
    created = resp.json()
    print(created)
B) Update an existing schedule to add/change a slug and parameters - The high-level helper only exposes
active
and
schedule
. To set slug/parameters/max_scheduled_runs, call the PATCH endpoint directly:
Copy code
from uuid import UUID
from prefect.client.orchestration import get_client

with get_client(sync_client=True) as client:
    deployment_id = UUID("your-deployment-id")
    schedule_id = UUID("your-schedule-id")

    # This payload mirrors DeploymentScheduleUpdate (slug/parameters/max_scheduled_runs supported)
    resp = client.request(
        "PATCH",
        "/deployments/{deployment_id}/schedules/{schedule_id}",
        path_params={"deployment_id": deployment_id, "schedule_id": schedule_id},
        json={
            "slug": "new-slug",
            "parameters": {"env": "staging"},
            "max_scheduled_runs": 50,
            "active": True,
            # "schedule": {...}  # include only if you want to change timing
        },
    )
Gotchas: - Slugs must be unique per deployment; you’ll get a 409 if you reuse one. - If you prefer sticking to strictly typed SDK objects, build the payloads with
DeploymentScheduleCreate/Update
and send
model_dump(mode="json")
as shown above. If you share your deployment name, I can draft a ready-to-run snippet that looks up the deployment ID and adds a slugged schedule end-to-end.