Lorenzo Gentile
12/22/2025, 12:16 PMMarvin
12/22/2025, 12:17 PMMarvin
12/22/2025, 12:21 PMimport 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
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?Lorenzo Gentile
12/22/2025, 1:06 PMMarvin
12/22/2025, 1:07 PMMarvin
12/22/2025, 1:12 PMcreate_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
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:
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.