How to load tasks from DB

[suresh-subramaniyan-s]

Hi ,

Need help on below 2 points

1. How to use rocketry to load the task from DB
2. whether rocketry able to run the task if task time falls in server scheduled maintenance ( misfire_grace_time functionality in APscheduler )

[Miksus]

Hi!

For the first point, Rocketry provides multiple ways to create tasks and those can be tailored as you wish.

One option is to use the app.task which is quite dynamic. The task does not necessarily need to be a decorator:

from rocketry import Rocketry

def get_tasks_from_database():
    # Query your database with whatever way
    ...
    yield {"name": "mytask", "path": "path/to/script.py", "func_name": "main"}

app = Rocketry()

for task_kwargs in get_tasks_from_database():
    app.task(**task_kwargs)

If you wish to have your tasks in sync with the database, you can create a meta task that runs with execution="main" and possibly runs perpetually. This task can poll other tasks from your database, just remove the previous.

For the second point, misfire is not yet built in but you can create your own condition. Of course, requires quite a bit of tinkering with the time mechanics. I think the upcoming event system could fulfil this hole partially but there are still things I'm not sure. I have also thought to implement a misfire supplementary condition but I remember that was quite non-trivial. Still on my list to implement at some point.