<@ULVA73B9P> Is it possible to trigger a schedule ...
# ask-marvin
k
@Marvin Is it possible to trigger a schedule (cron) for deployments using scripts instead of the UI?
m
thought for 308.9 seconds
Yes. In Prefect 3.x you can create and manage deployment schedules entirely via scripts/CLI or Python—no UI required. Here are the main options: - CLI (add a cron schedule to an existing deployment)
Copy code
prefect deployment schedule create my-flow/my-deployment --cron "0 9 * * *" --timezone "America/New_York"
Manage them:
Copy code
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)
Copy code
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):
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",
      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
)
Copy code
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.
k
@Marvin Image an existing deployment my_deployment. I want this deployment to have multiple schedules dynamically. What would a webhook look like that would receive parameter_1, parameter_2, parameter_3, and would just do the equivalent of the adding a new schedule to the existing my_deployment.
m
thought for 257.3 seconds
Absolutely—here’s a simple webhook you can host that adds a new cron schedule to an existing deployment and attaches
parameter_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.
Copy code
# 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:
Copy code
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: ```
prefect deployment schedule ls my-flow/my_deployment ``` - For Prefect Cloud, set
PREFECT_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