Steve Friedman
11/06/2025, 9:03 PMMarvin
11/06/2025, 9:03 PMMarvin
11/06/2025, 9:07 PMfrom prefect import flow
from prefect.runtime.flow_run import scheduled_start_time
@flow
def my_flow(execution_time=None):
# Use the runtime-provided scheduled time if no explicit parameter was given
execution_time = execution_time or scheduled_start_time()
print(f"Scheduled time: {execution_time}")
- Via run context (gives you the whole FlowRun object):
from prefect import flow
from prefect.context import get_run_context
@flow
def my_flow():
ctx = get_run_context()
print(f"Scheduled time: {ctx.flow_run.scheduled_start_time}")
Notes:
- The scheduled_start_time is the expected time the scheduler planned for the run. If the run starts late, this value does not change—perfect for backfills and lateness-aware logic.
- This value persists across retries; all retry attempts see the same originally scheduled time.
- If you trigger the flow manually (outside a deployment/schedule), the runtime accessor will return a sensible fallback (documented as “defaults to now” if not present).
- You generally cannot template the schedule to dynamically inject the scheduled time into parameters. If you really want a parameter, use the “None default + compute from runtime” pattern shown above and then pass that value into your tasks.
Related docs:
- Runtime flow run values: prefect.runtime.flow_run
- Access runtime info patterns: Access runtime info
- Creating schedules in deployments: Create schedulesSteve Friedman
11/06/2025, 9:11 PMMarvin
11/06/2025, 9:18 PMMarvin
11/06/2025, 9:22 PMfrom prefect import flow
from prefect.schedules import RRule
# Centralize your schedule string so both the deployment and the flow can reuse it
RRULE_TEXT = """DTSTART:20250115T090000
RRULE:FREQ=DAILY
EXDATE:20250120T090000
EXDATE:20250121T090000
"""
@flow
def my_flow():
pass
my_flow.deploy(
name="daily-except-holidays",
schedules=[
RRule(
rrule=RRULE_TEXT,
timezone="America/New_York", # Use your IANA timezone
)
],
)
- Tip: Put RRULE_TEXT somewhere shared so you can reuse it inside your flow when computing “previous date.” That guarantees your runtime logic exactly matches the deployment’s schedule/exclusions.
2) From inside the flow, get the previous scheduled occurrence (respecting exclusions)
- Compute it from the same RRULE you used for the deployment. Anchor the lookup on the flow’s scheduled time so late starts do not affect the result.
from prefect import flow
from prefect.runtime.flow_run import scheduled_start_time
from dateutil.rrule import rrulestr
from zoneinfo import ZoneInfo
RRULE_TEXT = """DTSTART:20250115T090000
RRULE:FREQ=DAILY
EXDATE:20250120T090000
EXDATE:20250121T090000
"""
@flow
def my_flow():
sched = scheduled_start_time() # The originally scheduled time (stable even if the run starts late/retries)
# Ensure the rrule and your reference time use the same timezone basis
tz = ZoneInfo("America/New_York")
sched_local = sched.astimezone(tz)
rule = rrulestr(RRULE_TEXT) # returns an rrule/rruleset respecting EXDATEs
previous_occurrence = rule.before(sched_local, inc=False)
print(f"Scheduled now: {sched_local}")
print(f"Previous scheduled date: {previous_occurrence}")
Why this works well
- Using scheduled_start_time() means your “previous” is computed relative to when the run was supposed to happen, not when it actually started—so late starts and retries won’t change the answer.
- Reusing the exact same RRULE (including EXDATEs) ensures you respect the identical exclusions and timezone that your deployment uses.
Notes and gotchas
- Keep RRULE times and timezone aligned. If you use local times in the RRULE/EXDATE lines, ensure the deployment’s timezone matches.
- If your RRULE is stored elsewhere (e.g., config), pass it into both the deployment and the flow so they stay in sync.
- Prefect does not expose a “previous occurrence” helper at runtime; using `dateutil.rrule`’s before() is the recommended way.
Helpful docs
- Schedules in Prefect: Schedules
- Creating schedules in deployments: Create schedules
- Accessing runtime info (like scheduled_start_time): Access runtime infoSteve Friedman
11/06/2025, 9:46 PMMarvin
11/06/2025, 9:50 PMMarvin
11/06/2025, 9:53 PMfrom datetime import timedelta
from prefect import flow
from prefect.runtime.flow_run import scheduled_start_time
@flow
def my_flow(target_date=None):
sched = scheduled_start_time() # originally scheduled time
if target_date is None:
target_date = (sched - timedelta(days=1)).date()
print(f"Target date: {target_date}")
- For “previous scheduled occurrence” of your RRule, reuse the exact RRULE used by the deployment and compute with dateutil:
from prefect import flow
from prefect.runtime.flow_run import scheduled_start_time
from dateutil.rrule import rrulestr
from zoneinfo import ZoneInfo
RRULE_TEXT = """DTSTART:20250115T090000
RRULE:FREQ=DAILY
EXDATE:20250120T090000
EXDATE:20250121T090000
"""
@flow
def my_flow(previous=None):
sched = scheduled_start_time()
tz = ZoneInfo("America/New_York")
sched_local = sched.astimezone(tz)
if previous is None:
rule = rrulestr(RRULE_TEXT)
previous = rule.before(sched_local, inc=False)
print(f"Previous scheduled occurrence: {previous}")
- Wrapper “launcher” flow (if you really want parameters set before the real run starts)
- Schedule a small launcher flow that computes the dynamic value then calls your deployment with explicit parameters:
from prefect import flow
from prefect.deployments import run_deployment
from prefect.runtime.flow_run import scheduled_start_time
from dateutil.rrule import rrulestr
RRULE_TEXT = "..." # same as your deployment
@flow
def launcher():
sched = scheduled_start_time()
previous = rrulestr(RRULE_TEXT).before(sched, inc=False)
run_deployment(
name="path/to:my_flow/daily-except-holidays",
parameters={"previous": previous.isoformat()},
)
- Then deploy/schedule the launcher. This gets you “dynamic parameters,” but via an extra hop.
Why not in schedule parameters?
- The parameters on a schedule are serialized at definition time and reused for every occurrence. There’s no built-in templating (e.g., “yesterday” or “previous occurrence”) evaluated per run.
Docs to reference:
- Schedules: Schedules
- Creating schedules in deployments: Create schedules
- Access runtime info (including scheduled_start_time): Access runtime info
If you share which schedule type you’re using (RRule vs Cron vs Interval) and your timezone, I can sketch the exact “previous occurrence” code for that case.