Hui Wu
04/22/2026, 3:42 AM"""Create a schedule"""
to_deploy = await import_flow(
workflow_scheduler, "app/workflow/prefect/flows/workflow_scheduler.py:workflow_scheduler"
)
deployment_uuid = await to_deploy.adeploy(
_name_=get_schedule_deployment_name(_schedule_id_),
_work_pool_name_=settings.prefect_dag_worker_pool,
_schedules_=[
Schedule(
_cron_=_cron_,
_timezone_=_timezone_,
)
],
_parameters_={
"platform_api_base": settings.platform_api_base,
"namespace_code": _namespace_code_,
"schedule_id": _schedule_id_,
},
_paused_=True,
# tags=[str(schedule_create.workflow_definition_code)],
# path=settings.working_dir,
# entrypoint="app.workflow.prefect.workflow:dag_runner",
# entrypoint_type=
)
return str(deployment_uuid)Marvin
04/22/2026, 3:42 AMMarvin
04/22/2026, 3:50 AMprefect deployment schedule CLI to list/pause/resume/create/delete specific schedules. For safe, targeted updates on an existing deployment, give each schedule a slug so Prefect upserts by slug instead of replacing everything.
Key points
- Python API: call flow.deploy(...) or flow.adeploy(...) again with the same name and your updated schedules=[...].
- Schedule merge vs. replace:
- If all existing and all provided schedules have slugs, Prefect upserts by slug (merge).
- If neither side has slugs, Prefect replaces all schedules.
- If existing schedules have slugs but you don’t provide slugs, the API errors to prevent accidental loss.
- Enable/disable a schedule via active=True/False on each schedule or with the CLI pause/resume.
- 3.x changes: don’t use 2.x methods like Deployment.build_from_flow() or prefect deployment build. Use flow.from_source(...).deploy(...) or prefect deploy.
What this looks like for your function
Your sample is already using Flow.adeploy. To update the schedule, call .adeploy again with the same deployment name and include a slug for the schedule so it’s updated in place.
Correct, 3.x-native imports and call:
from prefect.schedules import Cron
from prefect import flow
# to_deploy is the Flow object you imported via your helper
deployment_uuid = await to_deploy.adeploy(
name=get_schedule_deployment_name(_schedule_id_), # same name = update
work_pool_name=settings.prefect_dag_worker_pool,
schedules=[
Cron(
_cron_,
timezone=_timezone_,
slug=f"schedule-{_schedule_id_}", # slug enables safe upsert
active=True, # or False to keep paused
)
],
parameters={
"platform_api_base": settings.platform_api_base,
"namespace_code": _namespace_code_,
"schedule_id": _schedule_id_,
},
paused=True, # optional: pauses the deployment; use schedule.active for schedule enable/disable
)
return str(deployment_uuid)
Notes:
- Use schedules=[Cron(...)] (or Schedule(cron=..., timezone=..., slug=..., active=...)). The canonical keyword names in Prefect 3.x are name, work_pool_name, schedules, parameters, paused, and inside `Schedule`/`Cron`: cron, timezone, active, slug. Avoid the underscored variants unless you’re intentionally mapping them elsewhere.
If you maintain one schedule per deployment
- You can still use a slug so later calls only update that schedule in place.
- If you leave out slugs for both initial and update, Prefect will replace the schedules list entirely on update.
Update a single schedule without redeploy (Python)
from prefect.client.orchestration import get_client
from prefect.schedules import Cron
async def tweak_schedule(flow_name: str, deployment_name: str, schedule_slug: str, new_cron: str, tz: str):
async with get_client() as client:
dep = await client.read_deployment_by_name(f"{flow_name}/{deployment_name}")
target = next(s for s in dep.schedules if s.slug == schedule_slug)
await client.update_deployment_schedule(
deployment_id=dep.id,
schedule_id=target.id,
schedule=Cron(new_cron, timezone=tz),
# or toggle just active: active=False
)
Do it with the CLI
```
# See existing schedules and their IDs
prefect deployment schedule ls my-flow/my-deployment
# Pause or resume a specific schedule
prefect deployment schedule pause my-flow/my-deployment <schedule-id>
prefect deployment schedule resume my-flow/my-deployment <schedule-id>
# Replace a schedule’s timing (delete then create new)
prefect deployment schedule delete my-flow/my-deployment <schedule-id>Marvin
04/22/2026, 3:50 AMfrom prefect.schedules import Schedule, Cron, Interval, RRule
- Schedule(cron=..., timezone=..., active=..., slug=...)
- Cron("0 9 * * *", timezone="...", active=True, slug="...")
- Flow methods:
- Flow.deploy(...) / Flow.adeploy(...) support schedules=[...], schedule=..., cron=..., or interval=....
If you share how many schedules you want per deployment (one or many) and whether you prefer Python or CLI, I can tailor an exact snippet for your setup.