<@ULVA73B9P> we are observing cases where when we ...
# ask-marvin
m
@Marvin we are observing cases where when we re-deploy the workflow it is resulting into the multiple schedule for some deployments in the workflow. How to prevent and detect?
m
thought for 472.6 seconds
Thanks for flagging this — it’s usually caused by redeploys appending new schedules instead of updating existing ones. Quick context (Prefect 3.x): schedules are attached to a deployment. On redeploy, Prefect will update schedules in-place only if it can match them by a stable identifier (the schedule “slug”). If slugs are missing or change between deploys, new schedules get created and you’ll see multiples. How to prevent duplicates on redeploy - Give every schedule a stable slug and keep it the same across redeploys - Python:
Copy code
from 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:
Copy code
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:
Copy code
# YAML snippet
  schedules:
    - cron: "0 7 * * *"
      slug: "morning-v2"
      replaces: "morning"
      timezone: "UTC"
How to detect duplicates - CLI (human readable):
Copy code
prefect deployment schedule ls <flow-name>/<deployment-name>
- CLI (machine readable):
Copy code
prefect deployment schedule ls <flow-name>/<deployment-name> --output json
- Full deployment details (also shows schedules):
Copy code
prefect deployment inspect "<flow-name>/<deployment-name>"
How to remove duplicates - Delete a specific schedule by ID:
Copy code
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:
Copy code
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.
m
We are currently on Prefect v3.1.8 and using
flow.deploy
@Marvin We are currently on Prefect v3.1.8 and using
flow.deploy
m
thought for 410.9 seconds
Got it — thanks for the details. Since you’re on Prefect 3.1.8 and using
flow.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:
Copy code
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:
Copy code
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:
Copy code
# 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:
Copy code
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:
Copy code
prefect deployment schedule ls <flow-name>/<deployment-name>
- JSON for automation/audit:
Copy code
prefect deployment schedule ls <flow-name>/<deployment-name> --output json
- Full deployment info:
Copy code
prefect deployment inspect "<flow-name>/<deployment-name>"
How to clean up duplicates - Delete a specific schedule by ID:
Copy code
prefect deployment schedule delete <flow-name>/<deployment-name> <schedule-id>
- Clear all schedules, then redeploy with your intended config:
Copy code
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.
m
@Marvin I am using below:
Copy code
get_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>"],
m
thought for 76.0 seconds
Thanks for sharing the snippet! A few quick notes and a complete example to help: - Your pattern
get_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:
Copy code
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.
m
@Marvin I am facing an issue where we are observing multiple schedules for a given deployment. I am trying to solve that issue. 1. One solution can be figuring out the deployments which has multiple schedule runs and turning of the schedule and turning it on after sometime. That works out.
m
thought for 169.8 seconds
Got it — multiple schedules on the same deployment typically happen when a deployment is re-applied with
cron=
(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
Copy code
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):
Copy code
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).