Mrigank .
03/27/2026, 10:53 AMMarvin
03/27/2026, 10:53 AMMarvin
03/27/2026, 10:55 AMimport 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
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
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.Mrigank .
03/27/2026, 10:56 AMMarvin
03/27/2026, 10:56 AMMarvin
03/27/2026, 10:57 AMimport 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:
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.Mrigank .
03/27/2026, 11:19 AMMarvin
03/27/2026, 11:19 AMMarvin
03/27/2026, 11:19 AMimport 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
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.Mrigank .
03/27/2026, 11:34 AMMarvin
03/27/2026, 11:34 AMMarvin
03/27/2026, 11:35 AMprefect 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)
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)
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:
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.Mrigank .
03/27/2026, 11:36 AMMarvin
03/27/2026, 11:36 AMMarvin
03/27/2026, 11:39 AMimport 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.