Mrigank .
03/17/2026, 11:41 AMMarvin
03/17/2026, 11:41 AMMarvin
03/17/2026, 11:49 AMfrom prefect import flow
from prefect.schedules import Cron
@flow
def my_flow(): ...
my_flow.deploy(
name="my-deployment",
work_pool_name="my-pool",
schedules=[
Cron("0 6 * * *", slug="morning", timezone="UTC"),
Cron("0 18 * * *", slug="evening", timezone="UTC"),
],
)
- prefect.yaml:
deployments:
- name: my-deployment
entrypoint: flows/my_flow.py:my_flow
work_pool:
name: my-pool
schedules:
- cron: "0 6 * * *"
slug: "morning"
timezone: "UTC"
- cron: "0 18 * * *"
slug: "evening"
timezone: "UTC"
- If you need to rename a schedule, use replaces so the existing one is updated instead of creating a new one:
# YAML snippet
schedules:
- cron: "0 7 * * *"
slug: "morning-v2"
replaces: "morning"
timezone: "UTC"
How to detect duplicates
- CLI (human readable):
prefect deployment schedule ls <flow-name>/<deployment-name>
- CLI (machine readable):
prefect deployment schedule ls <flow-name>/<deployment-name> --output json
- Full deployment details (also shows schedules):
prefect deployment inspect "<flow-name>/<deployment-name>"
How to remove duplicates
- Delete a specific schedule by ID:
prefect deployment schedule ls <flow-name>/<deployment-name>
prefect deployment schedule delete <flow-name>/<deployment-name> <schedule-id>
- Clear all schedules, then redeploy with correct slugs:
prefect deployment schedule clear <flow-name>/<deployment-name> --accept-yes
# then redeploy with slugged schedules
Tips and common causes
- If you mix methods (prefect.yaml + ad-hoc “prefect deployment schedule create”), you can unintentionally append schedules on top of what your code/yaml defines.
- If your existing deployment already has slugged schedules and you redeploy with schedules that don’t include slugs, Prefect will require slugs to avoid ambiguity.
- Keep slugs stable; changing them between deploys creates new schedules.
Can you share:
- Your Prefect version (2.x or 3.x)?
- How you’re deploying (prefect.yaml via prefect deploy, or programmatically with flow.deploy/flow.from_source)?
- A quick snippet of how your schedules are defined now?
With that, I can suggest the minimal changes to make your redeploys idempotent and a quick one-off cleanup for current duplicates.Mrigank .
03/17/2026, 11:50 AMflow.deployMrigank .
03/17/2026, 12:05 PMflow.deployMarvin
03/17/2026, 12:05 PMMarvin
03/17/2026, 12:12 PMflow.deploy, here’s how duplicates can happen and how to prevent/detect them.
Why duplicates appear
- In 3.1.8, flow.deploy will:
- Replace all schedules on redeploy if you use `cron`/`interval`/`rrule` or pass schedules without slugs.
- Update schedules in-place by matching on slug if you provide slugs. If you change slugs across redeploys (or add more schedules via CLI/UI), the old schedules remain and you’ll see multiples.
Best practices to prevent duplicates
- If you just want “whatever I pass now is the only schedule(s) that exist,” don’t provide slugs:
from prefect import flow
@flow
def my_flow(): ...
# Single schedule
my_flow.deploy(name="my-deployment", work_pool_name="my-pool", cron="0 6 * * *")
# Or multiple (all will replace on redeploy)
my_flow.deploy(
name="my-deployment",
work_pool_name="my-pool",
cron=["0 6 * * *", "0 18 * * *"],
)
Each redeploy replaces the schedules — no duplicates.
- If you need partial updates, use stable slugs and keep them the same across redeploys:
from prefect.schedules import Cron
my_flow.deploy(
name="my-deployment",
work_pool_name="my-pool",
schedules=[
Cron("0 6 * * *", slug="morning", timezone="UTC"),
Cron("0 18 * * *", slug="evening", timezone="UTC"),
],
)
# Redeploy with the same slugs to update instead of append
my_flow.deploy(
name="my-deployment",
work_pool_name="my-pool",
schedules=[
Cron("0 7 * * *", slug="morning", timezone="UTC"), # updates
Cron("0 18 * * *", slug="evening", timezone="UTC"), # updates
],
)
If you need to rename a slug, use replaces so you don’t create a new schedule:
# rename "morning" -> "morning-v2"
my_flow.deploy(
name="my-deployment",
work_pool_name="my-pool",
schedules=[
Cron("0 7 * * *", slug="morning-v2", timezone="UTC", parameters=None), # new slug
# in YAML you can set 'replaces: morning' — with Python, keep the slug stable to avoid append
],
)
Note: Changing a slug without using the rename semantics (YAML replaces) will leave the old schedule in place.
- Avoid mixing schedule sources. If you also run:
prefect deployment schedule create <flow>/<deployment>
without --replace, that appends a new schedule on top of what flow.deploy manages.
How to detect duplicates
- List schedules:
prefect deployment schedule ls <flow-name>/<deployment-name>
- JSON for automation/audit:
prefect deployment schedule ls <flow-name>/<deployment-name> --output json
- Full deployment info:
prefect deployment inspect "<flow-name>/<deployment-name>"
How to clean up duplicates
- Delete a specific schedule by ID:
prefect deployment schedule delete <flow-name>/<deployment-name> <schedule-id>
- Clear all schedules, then redeploy with your intended config:
prefect deployment schedule clear <flow-name>/<deployment-name> --accept-yes
# then run flow.deploy again
If you can share a quick snippet of how you’re calling flow.deploy (especially whether you’re passing schedules=[...] with slug=...) and whether anyone has also used the CLI/UI to add schedules, I can suggest the minimal change to make your redeploys idempotent and a one-time cleanup plan.Mrigank .
03/25/2026, 8:33 AMget_flow.from_source(
source=str(Path(__file__).parent),
entrypoint="<workflow_name>.py:get_flow",
).deploy(
name="<deployement_name>",
work_pool_name="<workpool_name>",
parameters={
"flow_params": {
}
},
cron="30 00 * * 2-6",
tags=["<tag1>", "<tag2>"],Marvin
03/25/2026, 8:33 AMMarvin
03/25/2026, 8:34 AMget_flow.from_source(...).deploy(...) is correct in Prefect 3.x.
- Make sure your parameters keys match your flow’s signature (e.g., your flow must accept a flow_params argument if you pass it here).
- The cron string "30 00 * * 2-6" will run at 00:30 on Tue–Sat. If you meant Mon–Fri, use "30 0 * * 1-5". Both 0 and 00 are acceptable for the hour.
- By default, schedules use the server/account timezone (often UTC). If you need a specific timezone, prefer using a Cron schedule with timezone instead of the cron= shorthand.
Here’s a clean, complete version:
from pathlib import Path
from prefect.schedules import Cron
# from your_module import get_flow # ensure this imports your @flow
get_flow.from_source(
source=str(Path(__file__).parent), # local source dir (ensure workers can access it)
entrypoint="<workflow_name>.py:get_flow",
).deploy(
name="<deployment_name>",
work_pool_name="<workpool_name>",
parameters={
"flow_params": {}
},
# Option A: simple cron (UTC or server/account timezone)
# cron="30 0 * * 1-5",
# Option B: explicit schedule with timezone
schedule=Cron("30 0 * * 1-5", timezone="America/New_York"),
tags=["<tag1>", "<tag2>"],
)
A couple of gotchas:
- Use only one of interval, cron, rrule, or schedule(s) in deploy.
- If your worker runs remotely, source=str(Path(__file__).parent) won’t be available there; use a remote source (e.g., .from_source('<https://github.com/org/repo>') or a remote filesystem) instead.
Is there something specific that isn’t working with this setup (e.g., schedule timing, timezone, parameters binding, or worker not picking up code)? Happy to help troubleshoot.Mrigank .
03/25/2026, 8:36 AMMarvin
03/25/2026, 8:36 AMMarvin
03/25/2026, 8:39 AMcron= (or schedule=) repeatedly. In Prefect 3.x, each deploy() call with cron= will add a new schedule — there’s no automatic dedup. The reliable fix is to define your schedule(s) with stable slugs so subsequent deploys update them instead of creating new ones.
What to change going forward
- Don’t use the cron= shortcut when you’re redeploying often.
- Use a Schedule (or Cron) with a fixed slug via schedule= or schedules=[...]. When you redeploy with the same slug, Prefect updates the existing schedule instead of adding another.
Example converting your snippet
from pathlib import Path
from prefect.schedules import Cron
get_flow.from_source(
source=str(Path(__file__).parent),
entrypoint="<workflow_name>.py:get_flow",
).deploy(
name="<deployment_name>",
work_pool_name="<workpool_name>",
parameters={"flow_params": {}},
# Use a slug to ensure redeploys update (not duplicate) the schedule
schedule=Cron("30 0 * * 1-5", timezone="UTC", slug="weekday-0030"),
tags=["<tag1>", "<tag2>"],
)
- If you need multiple schedules, pass schedules=[Cron(..., slug="a"), Cron(..., slug="b")] and keep those slugs stable across redeploys.
One-off cleanup for existing duplicates
You can disable extra schedules programmatically and keep just one active. Here’s an async helper you can run once (supply your deployment_id — you can copy it from the UI):
import asyncio
from prefect.client.orchestration import get_client
async def cleanup_duplicate_schedules(deployment_id):
async with get_client() as client:
schedules = await client.read_deployment_schedules(deployment_id)
# Group by a simple key (cron/interval + tz + params). Adjust if needed.
def key(s):
sched = s.schedule
kind = getattr(sched, "type", None) or sched.__class__.__name__
tz = getattr(sched, "timezone", None)
cron = getattr(sched, "cron", None)
interval = getattr(sched, "interval", None)
params = tuple(sorted((s.parameters or {}).items()))
return (kind, tz, cron, interval, params)
groups = {}
for s in schedules:
groups.setdefault(key(s), []).append(s)
# For each group with duplicates, keep only the first active and disable others
for _, dupes in groups.items():
if len(dupes) > 1:
# Keep the first; disable the rest
to_disable = dupes[1:]
for s in to_disable:
await client.update_deployment_schedule(
deployment_id=deployment_id,
schedule_id=s.id,
active=False,
)
print(f"Disabled duplicate schedule {s.id} (slug={s.slug})")
# asyncio.run(cleanup_duplicate_schedules(<UUID>))
Notes
- Toggling schedules off/on can mitigate symptoms but won’t stop new duplicates from future deploys unless you switch to slugged schedules.
- If your CI/CD might run the same deploy() more than once, using slugs is essential to avoid duplicates.
- If workers run remotely, ensure your source is accessible to them (e.g., use a Git URL via .from_source('<https://github.com/org/repo>') or a remote filesystem) so redeploys don’t drift.
Docs for reference
- Schedules in Prefect 3: https://docs.prefect.io/latest/concepts/schedules/
- Flow.deploy API reference: https://docs.prefect.io/latest/api-ref/prefect/flows/#prefect.flows.Flow.deploy
If you share how you’re triggering redeploys (CI/CD, manual, frequency), I can tailor a cleanup/prevention script for your setup (including scanning all deployments and disabling duplicates safely).