Sorry, I encountered an error while processing your request. Please try again.

Assuming Prefect 3.x: use the orchestration client to look up the deployment by its fully-qualified name

"<flow_name>/<deployment_name>"

, then delete it by ID. Async (recommended):

from prefect.client.orchestration import get_client
from prefect.exceptions import ObjectNotFound

async def delete_deployment_by_name(flow_name: str, deployment_name: str):
    fq_name = f"{flow_name}/{deployment_name}"
    async with get_client() as client:
        try:
            dep = await client.read_deployment_by_name(fq_name)
            await client.delete_deployment(dep.id)
            print(f"Deleted deployment: {dep.name} ({dep.id})")
        except ObjectNotFound:
            print(f"Deployment not found: {fq_name}")

Synchronous:

from prefect.client.orchestration import get_client
from prefect.exceptions import ObjectNotFound

def delete_deployment_by_name_sync(flow_name: str, deployment_name: str):
    fq_name = f"{flow_name}/{deployment_name}"
    with get_client(sync_client=True) as client:
        try:
            dep = client.read_deployment_by_name(fq_name)
            client.delete_deployment(dep.id)
            print(f"Deleted deployment: {dep.name} ({dep.id})")
        except ObjectNotFound:
            print(f"Deployment not found: {fq_name}")

Notes: - The lookup format must be

"<flow_name>/<deployment_name>"

. - Deletion removes the deployment (and its schedules) but does not cancel already-created flow runs. If you prefer filtering (e.g., fuzzy matches or batch deletes), you can use

read_deployments(...)

with

DeploymentFilter

and

FlowFilter

, then call

delete_deployment

for each result.

Assuming Prefect 3.x: How to get schedule IDs for a deployment - Look up the deployment by fully-qualified name

<flow_name>/<deployment_name>

, then list its schedules. Async:

from prefect.client.orchestration import get_client
from prefect.exceptions import ObjectNotFound

async def list_schedule_ids(flow_name: str, deployment_name: str):
    fq_name = f"{flow_name}/{deployment_name}"
    async with get_client() as client:
        try:
            dep = await client.read_deployment_by_name(fq_name)
            # Option 1: via a dedicated API
            schedules = await client.read_deployment_schedules(dep.id)
            for s in schedules:
                print(f"schedule_id={s.id} slug={s.slug} active={s.active}")
            # Option 2: from the deployment response, if present
            # for s in dep.schedules:
            #     print(s.id, s.slug, s.active)
        except ObjectNotFound:
            print(f"Deployment not found: {fq_name}")

How to stop all schedules of a deployment You have two main options: Option A (simplest): Pause the entire deployment (pauses all schedules)

async def pause_all_schedules(flow_name: str, deployment_name: str):
    fq_name = f"{flow_name}/{deployment_name}"
    async with get_client() as client:
        dep = await client.read_deployment_by_name(fq_name)
        await client.pause_deployment(dep.id)
        print(f"Paused deployment {dep.name} and all schedules")

Option B (granular): Disable each schedule (keep deployment active)

async def disable_all_schedules(flow_name: str, deployment_name: str):
    fq_name = f"{flow_name}/{deployment_name}"
    async with get_client() as client:
        dep = await client.read_deployment_by_name(fq_name)
        schedules = await client.read_deployment_schedules(dep.id)
        for s in schedules:
            await client.update_deployment_schedule(
                deployment_id=dep.id,
                schedule_id=s.id,
                active=False
            )
        print(f"Disabled {len(schedules)} schedules for {dep.name}")

Bonus: Delete all schedules (removes them entirely)

async def delete_all_schedules(flow_name: str, deployment_name: str):
    fq_name = f"{flow_name}/{deployment_name}"
    async with get_client() as client:
        dep = await client.read_deployment_by_name(fq_name)
        schedules = await client.read_deployment_schedules(dep.id)
        for s in schedules:
            await client.delete_deployment_schedule(dep.id, s.id)
        print(f"Deleted {len(schedules)} schedules for {dep.name}")

Synchronous equivalents Add

sync_client=True

and call methods without `await`:

from prefect.client.orchestration import get_client

def pause_all_schedules_sync(flow_name: str, deployment_name: str):
    fq_name = f"{flow_name}/{deployment_name}"
    with get_client(sync_client=True) as client:
        dep = client.read_deployment_by_name(fq_name)
        client.pause_deployment(dep.id)

Notes - Format for lookup is

<flow_name>/<deployment_name>

. - Pausing a deployment does not delete schedules;

resume_deployment

will re-enable them. - Disabling schedules (

active=False

) keeps the deployment active but prevents new scheduled runs.

In Prefect 3.x, use Prefect’s built-in helper to list all supported IANA timezones:

from prefect.types._datetime import get_timezones

timezones = get_timezones()  # tuple[str, ...]
print(len(timezones), "timezones available")
print(timezones[:10])  # preview first few

Example: picking a timezone for a schedule

from prefect.schedules import Cron

Cron("0 9 * * *", timezone="America/New_York")

Tips - To validate a timezone string, check membership:

tz in get_timezones()

- If you prefer stdlib, you can also use

zoneinfo.available_timezones()

, but

get_timezones()

is the Prefect-approved list.

Great question. In Prefect 3.x, cron strings are validated using Prefect’s vendored croniter plus an extra validator that blocks non-deterministic patterns. How to validate a cron string (recommended)

from prefect._internal.schemas.validators import validate_cron_string

try:
    validate_cron_string("0 9 * * *")
    print("Valid")
except ValueError as e:
    print(f"Invalid: {e}")

You can also validate implicitly by constructing a schedule:

from prefect.schedules import Cron

Cron("0 9 * * *", timezone="UTC")  # raises ValueError if invalid

What formats are allowed - Fields: 5-field (minute hour day_of_month month day_of_week) and 6-field with seconds at the front (second minute hour day_of_month month day_of_week) - Supported syntax: - Wildcards:

*

- Lists:

1,5,10

- Ranges:

1-5

- Steps:

*/5

,

1-10/2

- Named months/days:

jan-mar

,

mon-fri

(case-insensitive) - Last day of month:

l

or

L

(e.g.,

0 0 l * *

) - Nth weekday of month:

2#3

(e.g.,

0 0 * * 2#3

is third Tuesday) - Aliases:

@hourly

,

@daily

,

@weekly

,

@monthly

, `@yearly`/`@annually` Not allowed - 7-field cron with year (e.g.,

0 0 0 * * * 2024

) is not supported for schedules - Randomized/hashed patterns are explicitly rejected: -

R

,

H

, and

h(...)

(e.g.,

0 0 * * R

) - Out-of-range values (e.g.,

60 * * * *

,

* 24 * * *

) Examples - Valid: -

0 0 * * *

(daily at midnight) -

*/15 * * * *

(every 15 minutes) -

0 9 * * 1-5

(weekdays at 9 AM) -

0 0 l * *

(last day of month) -

0 0 * * 2#3

(third Tuesday) -

0 0 0 * * *

(daily at midnight with seconds) -

@daily

- Invalid: -

0 0 * * R

(random) -

60 * * * *

(minute out of range) -

0 0 0 * * * 2024

(7 fields) If you want a simple boolean check instead of raising, you can use croniter:

from prefect._vendor.croniter import croniter

is_valid = croniter.is_valid("0 9 * * *")

简短结论： - 在 Prefect 3.x 中，调度（schedule）的

slug

不会自动生成，也不会根据 cron 自动转成“every minute every day”这类人类可读字符串。 -

slug

是可选字段，默认是 `None`。如果你没传，数据库就保存为 `NULL`，Prefect 不会替你生成。 - 如果你需要一个可读的标识，请在创建调度时自己设置 `slug`。 关键证据（基于源码核实）： -

prefect.schedules.Cron(..., slug: str | None = None)

中

slug

可选且默认 None。 - 客户端

DeploymentScheduleCreate(slug=None)

直接透传；服务端创建逻辑不生成默认 slug。 - 数据库有唯一索引 `(deployment_id, slug)`，但只在

slug IS NOT NULL

时约束；允许多个 NULL slug 并存。 建议实践： - 在创建调度时明确给一个稳定、可读的 slug，便于后续按 slug 精准更新或替换。例如：

from prefect.schedules import Cron

# 自定义一个可读 slug（自行约定格式）
daily_minutely_slug = "every-minute-every-day"

schedule = Cron(
    "*/1 * * * *",          # every minute
    timezone="UTC",
    slug=daily_minutely_slug
)

如果你想要“根据 cron 自动生成 slug”的效果，可以自定义一段小工具，把常见 cron 模式映射为你喜欢的文案/slug（例如

*/1 * * * *

-> `every-minute-every-day`，`0 9 * * 1-5` ->

weekdays-9am

等），然后把生成的结果传给 `slug`。需要的话我可以帮你写一个小的映射函数。

Great observation—the Prefect UI shows human-readable cron like “every minute every day.” That text is generated in the UI using the JavaScript library cronstrue. The Python SDK doesn’t include a built-in “cron to human text” helper, but you can replicate the UI behavior in Python with a small dependency. Recommended approach (closest to UI output): - Install cronstrue for Python and use it to render your cron string. Example:

# pip install cronstrue

from cronstrue import cronstrue

print(cronstrue("* * * * *"))           # Every minute
print(cronstrue("0 0 * * *"))           # At 12:00 AM
print(cronstrue("0 9 * * 1-5"))         # At 09:00 AM, Monday through Friday
print(cronstrue("*/15 * * * *"))        # Every 15 minutes
print(cronstrue("0 0 1 * *"))           # At 12:00 AM, on day 1 of the month

If you want a helper that accepts a Prefect Schedule and returns a description (Cron, Interval, RRule), here’s a compact utility:

from datetime import timedelta
from prefect.schedules import Cron, Interval, RRule

try:
    from cronstrue import cronstrue
    HAS_CRONSTRUE = True
except ImportError:
    HAS_CRONSTRUE = False

def human_duration(seconds: int) -> str:
    units = [("day", 86400), ("hour", 3600), ("minute", 60), ("second", 1)]
    parts = []
    for name, value in units:
        if seconds >= value:
            count, seconds = divmod(seconds, value)
            parts.append(f"{count} {name}{'s' if count != 1 else ''}")
    return " ".join(parts) if parts else "0 seconds"

def describe_schedule(schedule):
    # Interval schedule
    if getattr(schedule, "interval", None) is not None:
        secs = int(schedule.interval.total_seconds()) if isinstance(schedule.interval, timedelta) else int(schedule.interval)
        return f"Every {human_duration(secs)}"
    # Cron schedule
    if getattr(schedule, "cron", None) is not None:
        if HAS_CRONSTRUE:
            return cronstrue(schedule.cron)
        return f"Cron: {schedule.cron}"
    # RRule schedule
    if getattr(schedule, "rrule", None) is not None:
        try:
            from rrule import rrulestr  # pip install rrule
            return rrulestr(schedule.rrule).toText()
        except Exception:
            return f"RRule: {schedule.rrule}"
    return "Unknown schedule"

# Examples
print(describe_schedule(Cron("* * * * *")))                 # Every minute
print(describe_schedule(Interval(timedelta(minutes=5))))    # Every 5 minutes
print(describe_schedule(RRule("RRULE:FREQ=DAILY;INTERVAL=1")))

Notes: - The UI uses cronstrue (JS). The Python package cronstrue aims to match that output closely. - For validation before describing, Prefect uses croniter under the hood; you can check validity with:

from prefect._vendor.croniter import croniter
valid = croniter.is_valid("*/5 * * * *")

- For RRule descriptions, use the rrule package (

pip install rrule

) and call

rrulestr(...).toText()

.