<@ULVA73B9P> how to update the schedule of a deplo...
# marvin-ai
h
@Marvin how to update the schedule of a deployment`async def create_schedule(self, _namespace_code_: int, _schedule_id_: int, _cron_: str, _timezone_: str) -> str:`
"""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)
m
thought for 455.8 seconds
Short answer: in Prefect 3.x you update a deployment’s schedule by redeploying the same deployment name with the new schedule(s), or by using the
prefect 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:
Copy code
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)
Copy code
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>
prefect deployment schedule create my-flow/my-deployment --cron "0 9 * * *" --timezone "America/New_York" ``` Verified primitives you can use in 3.x - Schedules API: -
from 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.