Konstantinos Papagiannopoulos
11/20/2025, 2:06 PMMarvin
11/20/2025, 2:06 PMMarvin
11/20/2025, 2:12 PMprefect deployment schedule create my-flow/my-deployment --cron "0 9 * * *" --timezone "America/New_York"
Manage them:
prefect deployment schedule ls my-flow/my-deployment
prefect deployment schedule pause my-flow/my-deployment <schedule-id>
prefect deployment schedule resume my-flow/my-deployment <schedule-id>
prefect deployment schedule delete my-flow/my-deployment <schedule-id>
Tips:
- Use --replace to replace existing schedules (skip prompt with -y).
- You can also use --interval <seconds> or --rrule "<RRULE>", and --day_or=false for fcron-style AND logic.
- Docs: Create schedules, Manage schedules
- Python (attach schedules when deploying)
from prefect import flow
@flow
def my_flow():
...
my_flow.deploy(
name="my-deployment",
work_pool_name="my-pool",
cron="0 9 * * *", # or interval=600, rrule="FREQ=DAILY;INTERVAL=1"
paused=False, # optional: start active
)
Or with schedule objects for advanced options (timezone, parameters, slug):
from prefect import flow
from prefect.schedules import Cron
@flow
def my_flow():
...
my_flow.deploy(
name="my-deployment",
work_pool_name="my-pool",
schedule=Cron("0 9 * * *", timezone="America/New_York"),
)
API refs: Schedules concept and Python schedule API
- YAML (define schedules declaratively and apply with prefect deploy)
deployments:
- name: my-deployment
entrypoint: flows.py:my_flow
schedules:
- cron: "0 9 * * *"
timezone: "America/New_York"
If you share how you prefer to manage this (CLI vs Python vs YAML), I can tailor an example to your setup.Konstantinos Papagiannopoulos
11/20/2025, 2:27 PMMarvin
11/20/2025, 2:27 PMMarvin
11/20/2025, 2:32 PMparameter_1, parameter_2, and parameter_3 as parameter overrides for runs created by that schedule.
This example uses FastAPI and the Prefect 3.x Python client. Each POST adds another schedule (does not replace existing ones), so you can dynamically accumulate multiple schedules for the same deployment.
# requirements:
# fastapi
# uvicorn
# prefect>=3
import os
from typing import Optional
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from prefect.client.orchestration import get_client
from prefect.client.schemas.schedules import CronSchedule
app = FastAPI()
class AddScheduleRequest(BaseModel):
# If you always target the same deployment, you can hardcode this
deployment_name: str = Field(default="my-flow/my_deployment", description="Format: <flow-name>/<deployment-name>")
# The cron you want to add as a new schedule
cron: str = Field(example="0 9 * * *")
timezone: str = Field(default="UTC", example="America/New_York")
active: bool = True
slug: Optional[str] = Field(default=None, description="Optional label for the schedule")
# These will be attached as parameter overrides for runs from this schedule
parameter_1: str
parameter_2: str
parameter_3: str
# Optional: control cron day vs day_of_week handling (defaults to True = OR, standard cron)
day_or: Optional[bool] = True
@app.post("/add-schedule")
async def add_schedule(req: AddScheduleRequest):
# Ensure your environment is configured (Prefect Cloud example):
# export PREFECT_API_URL="<https://api.prefect.cloud/api/accounts/<acc-id>/workspaces/<ws-id>>"
# export PREFECT_API_KEY="<your-key>"
try:
async with get_client() as client:
# 1) Resolve deployment by name to get its ID
dep = await client.read_deployment_by_name(req.deployment_name)
# 2) Create a new cron schedule (active by default, or per req.active)
cron = CronSchedule(cron=req.cron, timezone=req.timezone, day_or=req.day_or)
created = await client.create_deployment_schedules(
deployment_id=dep.id,
schedules=[(cron, req.active)],
)
sched = created[0]
# 3) Attach parameter overrides (so runs from this schedule get these values)
params = {
"parameter_1": req.parameter_1,
"parameter_2": req.parameter_2,
"parameter_3": req.parameter_3,
}
await client.update_deployment_schedule(
deployment_id=dep.id,
schedule_id=sched.id,
parameters=params,
slug=req.slug,
)
return {
"status": "ok",
"deployment_id": str(dep.id),
"deployment_name": dep.name,
"schedule_id": str(sched.id),
"cron": req.cron,
"timezone": req.timezone,
"active": req.active,
"parameters": params,
"slug": req.slug,
}
except Exception as exc:
raise HTTPException(status_code=400, detail=str(exc))
# Run with: uvicorn app:app --host 0.0.0.0 --port 8000
Example request body:
POST /add-schedule
Content-Type: application/json
{
"deployment_name": "my-flow/my_deployment",
"cron": "0 9 * * MON-FRI",
"timezone": "America/New_York",
"parameter_1": "A",
"parameter_2": "B",
"parameter_3": "C",
"active": true,
"slug": "weekday-9am"
}
Notes:
- Make sure your flow’s parameter names match parameter_1, parameter_2, and parameter_3. These values will override the deployment’s defaults when this schedule fires.
- Each call creates an additional schedule; it doesn’t replace existing ones.
- You can verify schedules via:
```Marvin
11/20/2025, 2:32 PMPREFECT_API_URL and PREFECT_API_KEY in the environment. For a self-hosted server, set PREFECT_API_URL to your server URL.
Docs for reference:
- Schedules concept: Concepts: Schedules
- Create/manage schedules: Create schedules and Manage schedules