What is the recipe for configuring a scheduled Flo...
# prefect-community
s
What is the recipe for configuring a scheduled Flow's parameters based on the "scheduled time"? I found
parameter_defaults
for Clocks, but they seem to be static w.r.t each clock. For example, I want the parameter be "day of the week",i.e. Mon, Tue....
a
here is one example of using
parameter_defaults
with an `IntervalSchedule`:
Copy code
import datetime
import prefect
from prefect import task, Flow, Parameter
from prefect.schedules import clocks, Schedule

now = datetime.datetime.utcnow()

clock1 = clocks.IntervalClock(start_date=now, 
                                interval=datetime.timedelta(minutes=1), 
                                parameter_defaults={"p": "CLOCK 1"})
clock2 = clocks.IntervalClock(start_date=now + datetime.timedelta(seconds=30), 
                                interval=datetime.timedelta(minutes=1), 
                                parameter_defaults={"p": "CLOCK 2"})

@task
def log_param(p):
    logger = prefect.context['logger']
    <http://logger.info|logger.info>("Received parameter value {}".format(p))
    
    
with Flow("Varying Parameters", schedule = Schedule(clocks=[clock1, clock2])) as flow:
	p = Parameter("p", default=None, required=False)
    log_param(p)

flow.run()
to get a dynamic date within your task, you should use neither
parameter_defaults
on the clock nor the
default
value on the
Parameter
task because those values are frozen at registration time. Your best bet is doing something like:
Copy code
@task
def get_today():
    return datetime.today()
and then pass it as data dependency to your downstream task that needs this dynamic date. This blog post discusses this problem in much more detail: https://www.prefect.io/blog/how-to-make-your-data-pipelines-more-dynamic-using-parameters-in-prefect/
k
I think you can just
Copy code
@task
def get_param():
    scheduled_time = prefect.context.get("scheduled_start_time")
    return some_other_value_based_on_scheduled_time
and then use this at the start of your flow
upvote 2
a
good point Kevin, there is even today in context:
Copy code
prefect.context.get("today")