<@ULVA73B9P> how can we toggle the deployment sche...
# ask-marvin
m
@Marvin how can we toggle the deployment schedule using python SDK i.e. make the schedule inactive and active after say 5 seconds
m
thought for 140.3 seconds
Great question. In Prefect 3.x you can toggle a deployment’s schedule by updating the schedule’s active flag via the orchestration client. Deployments can have multiple schedules, so you toggle them per-schedule. Below are concise examples that deactivate a schedule, wait 5 seconds, then reactivate it. Async (recommended)
Copy code
import asyncio
from uuid import UUID
from prefect.client.orchestration import get_client

async def toggle_schedule_for_5s(deployment_id: str, schedule_id: str, seconds: int = 5):
    dep_id = UUID(deployment_id)
    sch_id = UUID(schedule_id)

    async with get_client() as client:
        # Deactivate
        await client.update_deployment_schedule(
            deployment_id=dep_id,
            schedule_id=sch_id,
            active=False,
        )
        await asyncio.sleep(seconds)
        # Reactivate
        await client.update_deployment_schedule(
            deployment_id=dep_id,
            schedule_id=sch_id,
            active=True,
        )

# asyncio.run(toggle_schedule_for_5s("YOUR-DEPLOYMENT-ID", "YOUR-SCHEDULE-ID", 5))
Sync
Copy code
import time
from uuid import UUID
from prefect.client.orchestration import get_client

def toggle_schedule_for_5s_sync(deployment_id: str, schedule_id: str, seconds: int = 5):
    dep_id = UUID(deployment_id)
    sch_id = UUID(schedule_id)

    with get_client(sync_client=True) as client:
        client.update_deployment_schedule(deployment_id=dep_id, schedule_id=sch_id, active=False)
        time.sleep(seconds)
        client.update_deployment_schedule(deployment_id=dep_id, schedule_id=sch_id, active=True)

# toggle_schedule_for_5s_sync("YOUR-DEPLOYMENT-ID", "YOUR-SCHEDULE-ID", 5)
Finding schedule IDs for a deployment
Copy code
from uuid import UUID
from prefect.client.orchestration import get_client
import asyncio

async def list_schedules(deployment_id: str):
    async with get_client() as client:
        schedules = await client.read_deployment_schedules(UUID(deployment_id))
        for s in schedules:
            print("id:", s.id, "| slug:", s.slug, "| active:", s.active, "| schedule:", s.schedule)

# asyncio.run(list_schedules("YOUR-DEPLOYMENT-ID"))
Notes - This toggles the specific schedule’s
active
flag; repeat for each schedule if your deployment has multiple. - Ensure your environment is authenticated to your Prefect API (e.g.,
PREFECT_API_URL
and
PREFECT_API_KEY
for Cloud). If you’d prefer to target a schedule by slug or by deployment name instead of IDs, let me know and I’ll tailor the snippet.
m
@Marvin I only want using the deployment IDs? Do we need schedule IDs?
m
thought for 20.2 seconds
Short answer: yes—you need schedule IDs to toggle a specific schedule. Prefect 3.x stores schedules per-deployment, and the SDK’s toggle endpoint is per-schedule. There isn’t a single “deployment-level” on/off switch. If you only have the deployment ID, you can: - Read all schedules for that deployment - Toggle them (one or all) by their IDs Example: toggle all schedules for a deployment for 5 seconds using only the deployment ID
Copy code
import asyncio
from uuid import UUID
from prefect.client.orchestration import get_client

async def pause_deployment_schedules_for(deployment_id: str, seconds: int = 5):
    dep_id = UUID(deployment_id)
    async with get_client() as client:
        schedules = await client.read_deployment_schedules(dep_id)
        if not schedules:
            print("No schedules found for this deployment.")
            return

        # Deactivate all schedules
        await asyncio.gather(*[
            client.update_deployment_schedule(
                deployment_id=dep_id,
                schedule_id=s.id,
                active=False
            ) for s in schedules
        ])

        await asyncio.sleep(seconds)

        # Reactivate all schedules
        await asyncio.gather(*[
            client.update_deployment_schedule(
                deployment_id=dep_id,
                schedule_id=s.id,
                active=True
            ) for s in schedules
        ])

# asyncio.run(pause_deployment_schedules_for("YOUR-DEPLOYMENT-ID", seconds=5))
If your deployment has exactly one schedule and you only want to toggle that one, you can enforce that:
Copy code
import asyncio
from uuid import UUID
from prefect.client.orchestration import get_client

async def pause_single_schedule_for(deployment_id: str, seconds: int = 5):
    dep_id = UUID(deployment_id)
    async with get_client() as client:
        schedules = await client.read_deployment_schedules(dep_id)
        if len(schedules) != 1:
            raise ValueError(f"Expected exactly 1 schedule, found {len(schedules)}")
        s = schedules[0]

        await client.update_deployment_schedule(deployment_id=dep_id, schedule_id=s.id, active=False)
        await asyncio.sleep(seconds)
        await client.update_deployment_schedule(deployment_id=dep_id, schedule_id=s.id, active=True)

# asyncio.run(pause_single_schedule_for("YOUR-DEPLOYMENT-ID", 5))
If you’d prefer sync versions, I can share those too. Also, let me know if you want to target a specific schedule by slug or other criteria instead of toggling all.
m
@Marvin suggest a sync version please
m
thought for 18.1 seconds
Absolutely—here are sync versions using only the deployment ID. These read the schedules for the deployment and toggle them by their IDs under the hood. Pause all schedules for N seconds (restores only those that were originally active)
Copy code
import time
from uuid import UUID
from prefect.client.orchestration import get_client

def pause_deployment_schedules_for_sync(deployment_id: str, seconds: int = 5):
    dep_id = UUID(deployment_id)
    with get_client(sync_client=True) as client:
        schedules = client.read_deployment_schedules(dep_id)
        if not schedules:
            print("No schedules found for this deployment.")
            return

        # Track which schedules were originally active
        originally_active = {s.id for s in schedules if s.active}

        # Deactivate only active schedules
        for s in schedules:
            if s.active:
                client.update_deployment_schedule(
                    deployment_id=dep_id,
                    schedule_id=s.id,
                    active=False
                )

        time.sleep(seconds)

        # Reactivate only those that were originally active
        for s in schedules:
            if s.id in originally_active:
                client.update_deployment_schedule(
                    deployment_id=dep_id,
                    schedule_id=s.id,
                    active=True
                )
If the deployment has exactly one schedule and you only want to toggle that one
Copy code
import time
from uuid import UUID
from prefect.client.orchestration import get_client

def pause_single_schedule_for_sync(deployment_id: str, seconds: int = 5):
    dep_id = UUID(deployment_id)
    with get_client(sync_client=True) as client:
        schedules = client.read_deployment_schedules(dep_id)
        if len(schedules) != 1:
            raise ValueError(f"Expected exactly 1 schedule, found {len(schedules)}")

        s = schedules[0]
        if s.active:
            client.update_deployment_schedule(deployment_id=dep_id, schedule_id=s.id, active=False)
            time.sleep(seconds)
            client.update_deployment_schedule(deployment_id=dep_id, schedule_id=s.id, active=True)
        else:
            # Already inactive; just wait without toggling
            time.sleep(seconds)
These use Prefect 3.x’s sync client with
get_client(sync_client=True)
. If you’d like a version that targets a specific schedule by slug (still starting from the deployment ID), I can share that too.
m
@Marvin I am getting AttributeError: 'SyncPrefectClient' object has no attribute 'update_deployment_schedule'
m
thought for 70.9 seconds
Thanks for flagging that. An AttributeError on SyncPrefectClient usually means either: - You’re on Prefect 2.x (the method doesn’t exist there), or - You’re on an older 3.x where the sync surface didn’t include that method yet. Could you share the output of
prefect version
and the exact import you’re using? That’ll let me tailor the fix. Workarounds you can use right now from sync code 1) Call the async client from sync (works across versions)
Copy code
import anyio
from uuid import UUID
from prefect.client.orchestration import get_client

def toggle_schedule_sync_via_async(deployment_id: str, schedule_id: str, active: bool):
    async def _run():
        async with get_client() as client:
            await client.update_deployment_schedule(
                deployment_id=UUID(deployment_id),
                schedule_id=UUID(schedule_id),
                active=active,
            )
    anyio.run(_run)
2) Direct REST call (no SDK surface required)
Copy code
import os
import time
import requests

def pause_all_schedules_for_sync(deployment_id: str, seconds: int = 5):
    api_url = os.environ["PREFECT_API_URL"]
    api_key = os.environ.get("PREFECT_API_KEY")
    headers = {"Authorization": f"Bearer {api_key}"} if api_key else {}

    # Read deployment to get its schedules
    dep = requests.get(f"{api_url}/deployments/{deployment_id}", headers=headers, timeout=30).json()
    schedules = dep.get("schedules", [])
    originally_active = [s for s in schedules if s.get("active")]

    # Deactivate only those that are active
    for s in originally_active:
        r = requests.patch(
            f"{api_url}/deployments/{deployment_id}/schedules/{s['id']}",
            json={"active": False},
            headers=headers,
            timeout=30,
        )
        r.raise_for_status()

    time.sleep(seconds)

    # Reactivate only those that were originally active
    for s in originally_active:
        r = requests.patch(
            f"{api_url}/deployments/{deployment_id}/schedules/{s['id']}",
            json={"active": True},
            headers=headers,
            timeout=30,
        )
        r.raise_for_status()
If you are on Prefect 3.x and still missing
update_deployment_schedule
on the sync client, upgrading to the latest 3.x should expose it:
Copy code
pip install -U prefect
Once we confirm your Prefect version, I can provide a version-specific snippet (or the exact fix) to use
SyncPrefectClient
directly.
m
@Marvin I am using prefect v3.1.8
m
thought for 135.9 seconds
Thanks for the version info. In Prefect 3.1.8, SyncPrefectClient does not expose update_deployment_schedule/read_deployment_schedules; those are only on the async client. That’s why you’re seeing the AttributeError. Here’s a pure sync workaround using the sync client’s underlying HTTP client to call the REST endpoints directly. It uses only the deployment ID: it fetches the schedules, deactivates the ones that are currently active, waits 5 seconds, then reactivates only those.
Copy code
import time
from uuid import UUID
from prefect.client.orchestration import get_client

def pause_deployment_schedules_for_sync(deployment_id: str, seconds: int = 5):
    dep_id = UUID(deployment_id)
    with get_client(sync_client=True) as client:
        # Read schedules for the deployment
        r = client._client.get(f"/deployments/{dep_id}/schedules")
        r.raise_for_status()
        schedules = r.json()  # list[dict] with keys like id, active, schedule, slug, etc.

        # Remember which were active
        originally_active = [s for s in schedules if s.get("active")]

        # Deactivate only those that were active
        for s in originally_active:
            rr = client._client.patch(
                f"/deployments/{dep_id}/schedules/{s['id']}",
                json={"active": False},
            )
            rr.raise_for_status()

        time.sleep(seconds)

        # Reactivate only those that were originally active
        for s in originally_active:
            rr = client._client.patch(
                f"/deployments/{dep_id}/schedules/{s['id']}",
                json={"active": True},
            )
            rr.raise_for_status()
If you prefer to keep everything strictly in the SDK without touching _client, another option is to call the async client from sync using anyio.run and use the async methods directly—happy to share that snippet too. If you can upgrade later, newer 3.x releases expose more orchestration helpers on the async client (and in some cases sync), but the above will work on 3.1.8 today.