I have a task that takes a single argument: a

date

(think a task that downloads data corresponding to a date). I would like to have this task run once a day for all days, starting from a date in the past, lets call it

start_time

. How do I achieve this with Prefect? The closes I've found is this: https://github.com/PrefectHQ/prefect/issues/1179

current_time = Parameter("timestamp", default=None)

@prefect.task
def do_something_time_specific(current_time):
    if current_time is None:
        current_time = prefect.context.get("scheduled_start_time")
    if isinstance(current_time, str):
        current_time = pendulum.parse(current_time)
    # does something dealing with time

In words: the task depends on a

date

but if a date isn't passed, it uses the

scheduled_start_time

. This seems to have a problem. Lets suppose it's now

2022-01-01

and Prefect has scheduled tasks for

2022-01-02

,

2022-01-03

, etc... Now my computer is off for a few days and I turn it on again on

2022-01-04

. What happened to the runs of those two dates which were jumped? Do they both run on

2022-01-04

but with

scheduled_start_time

in

2022-01-02

,

2022-01-03

? If that's the case, this solution does the right thing. Or do they both run on

2022-01-04

but the

scheduled_start_time

is

2022-01-04

for both? (in which case this solution doesn't work)

Hey @vawati, I think this should work. When you restart the agent, it will get the flows that have not run yet. The scheduled start time will remain as intended. Just note that this value is UTC.

Yes, flows, sorry. Ok, thank you. I'll try and check if this actually works (i.e. by turning my off computer for a couple of days 🤣

The only caveat here is that Prefect scheduler never creates flow runs for the past, but the future runs will work fine when you restart the agent, this page explains it: https://docs.prefect.io/orchestration/concepts/services.html#how-does-it-work