<@ULVA73B9P> How can I pass in the time a deployme...
# ask-marvin
s
@Marvin How can I pass in the time a deployment created a flow run from a schedule via flow parameters? I'd like to do it from Python. Will it still work if it starts late?
m
thought for 258.4 seconds
Short answer: don’t pass it as a parameter—read it from Prefect’s runtime context inside your flow. It represents the originally scheduled time and stays the same even if the run starts late or retries. How to access the scheduled time (Python): - Via runtime helper:
Copy code
from 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):
Copy code
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 schedules
s
@Marvin What would be the past way to accomplish the following in that sort of set up? 1. A deployment schedule runs daily excluding certain dates, e.g. RRule schedule with excluded dates 2. The flow wants to find the previous date from that schedule
m
thought for 257.0 seconds
Great question. In Prefect 3.x, here’s a clean approach that covers both parts: 1) Define a daily deployment schedule with exclusions - Use an RRule schedule and encode exclusions with EXDATEs (iCalendar syntax). Keep the timezone explicit so DST behaves as expected. - Example:
Copy code
from 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.
Copy code
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 info
s
@Marvin would it be possible to pass in the previous one from the deployment parameters itself? or in general having a dynamic parameter set by a function like "yesterday"?
m
thought for 146.1 seconds
Short answer: deployment schedule parameters are static. Prefect doesn’t evaluate a callable per occurrence, so you can’t have the schedule itself inject a “dynamic” parameter like “yesterday” or “previous occurrence” at run creation time. Recommended patterns: - Compute inside the flow (simple and robust) - Use the flow’s scheduled time as your anchor so late starts don’t change behavior. - For “yesterday”:
Copy code
from 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:
Copy code
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:
Copy code
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.