https://prefect.io logo
Channels
Powered by
n

Nikola Milushev

02/12/2021, 5:28 PM
Hi guys, I have a scheduled prefect flow (runs once a day) that I register once. The flow has several Parameters, two of them being some dates which I would like to have default values with respect to the flow runs, not the flow registration. Now, the default values are set correctly upon flow registration, but on the next day, the same default dates appear, even though I expect them to be one day later. I've set the dates like this:
Copy code
schedule = Schedule(clocks=[IntervalClock(timedelta(days=1))])
with Flow(name=name, schedule=schedule) as flow:
    start_date = Parameter(name="start_date",
                           required=False,
                           default=(datetime.now() - timedelta(days=2)).strftime("%Y-%m-%d"))
    end_date = Parameter(name="end_date",
                         required=False,
                         default=(datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d"))
👍 1
j

Jim Crist-Harif

02/12/2021, 5:31 PM
Hi Nikola, this is partially a consequence of "pickle-based storage" (https://docs.prefect.io/orchestration/flow_config/storage.html#pickle-based-storage). When you build your flow, 
datetime.now()
runs then, and the parameter defaults are hardcoded. Then when you run it, the same default values (that were computed at registration time) are used.
upvote 2
Parameter defaults have an extra caveat in that they are also stored in the backend (prefect cloud or server), so when you kick off flow runs the defaults stored in the backend may also be used (and again, those were computed once at flow registration time).
To support what you want, I suggest using a default of 
None
and handling the none case special in your code. Then at runtime you can compute the offset you desire.
Copy code
from prefect import task, Parameter

@task
def normalize(date, days):
    if date is None:
        return (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d"))
    return date


with Flow(name=name, schedule=schedule) as flow:

    start_date = Parameter(name="start_date", required=False, default=None)
    end_date = Parameter(name="end_date", required=False, default=None)

    start_date = normalize(start_date, 2)
    end_date = normalize(end_date, 1)
👀 1
n

Nikola Milushev

02/12/2021, 5:39 PM
Hi Jim, thanks for this super quick and detailed explanation. As in your example, since you've moved the date's initialization in a task, the same will be run everytime the flow is run therefore dates will get updated everytime as well, right?
j

Jim Crist-Harif

02/12/2021, 5:39 PM
yes, that's correct
n

Nikola Milushev

02/12/2021, 5:40 PM
Alright, I will try that right away, thanks again!
3 Views
Powered by