Scheduling a task to run only once at specified datetime

[spietras]

I'm slightly confused because this seems like a pretty basic thing that one would want to do, but I can't find anything on this topic in the documentation and in discussions/issues. Is it possible to schedule a task to run only once at a specific datetime?

You might be wondering what's the use case here. See, I need to create some tasks dynamically without knowing the condition beforehand. Let's say that I have some task that itself is running periodically with a static condition (like once a minute), fetches some data, and then schedules other tasks based on this data. Dynamic scheduling is of course possible using create_task method and everything would work great if these new tasks were periodic. But here I don't need them to be periodic. I just need to run them once at some point in the future (and this point comes from the data fetched dynamically at runtime).

In static scheduling, this is probably less common, but you can also imagine one wanting to do something along the lines of:

@app.task(once.at("2023-01-01T00:00:00Z"))
def do_once():
    print("2023 is here")

Is it possible to do with the current state of the project? I feel like I might be missing something.

[spietras]

If there is no built-in easy way, then I think this custom condition might be good enough:

@app.cond()
def at(dt, task = Task()):
    return task.last_run is None and datetime.now() >= dt

[Miksus]

Sorry for taking time to answer (been busy lately).

I have plans to include something like this. As you said, this is useful for dynamic tasks but I think also because Airflow has similar, why not Rocketry as implementing such is not really complex.

My ideas for the naming were:

• at_datetime
• after

But I think I like your proposal the most: once is pretty obvious and has the benefit of the task actually runs only once. I think it could be a similar wrapper as some other conditions: simply once runs the task only once and once("2022-12-24") runs it once after given datetime. Perhaps there also could be a condition that is true after given datetime (no check on whether the task ran) so if one wanted to run a task periodically after some time in the future.

Furthermore, if you have a stream of datetimes (fetched from a database or from a user input for example), the new event system might be of use. It will look something like this:

@event_stream()
def spawn():
    return datetime(2022, 12, 24)

@app.task(spawn)
def do_things()
    ...