Dynamically adding tasks via plug-ins?

[RhetTbull]

Hi! First, thanks for Rocketry! I'm thinking about using Rocketry in a project that will need to run an event loop with that responds both to events and runs schedules tasks. Rocketry seems like it might be perfect for the latter. However, the project will also need to dynamically load plug-ins at run time. Looking through the docs & examples, I'm not sure if this is something Rocketry can do out of the box.

For example, a plug in author creates a plugin1.py as:

from rocketry.conds import daily, after_success
from rocketry.args import Return

@app.task(daily.after("07:00"))
def do_first():
    ...
    return 'Hello World'

@app.task(after_success(do_first))
def do_second(arg=Return('do_first')):
    # arg contains the value of the task do_first's return
    ...
    return 'Hello Python'

and another plug-in author creates plugin2.py:

from rocketry.conds import daily
from rocketry.args import Return

@app.task(daily.after("15:00"))
def do_task():
    ...
    return 'Hello World'

I'd like to be able to load the plug-ins at run-time (and after the session is already running). I could have a task that continually monitors the plug-in folder for new plug-ins and uses create_task but does that work with the decorators? (e.g. could plug-in author use @app.task(daily...) even though they don't have access to the app object that's already running?

Just brainstorming at this point--open to any ideas about implementing such a system and whether using or extending Rocketry might be a good approach.

[Miksus]

Hi! And thanks! It's always nice to hear people find it useful!

Rocketry is designed to create tasks on runtime. The create_task is slightly lower level (than the app instance) so it assumes keyword arguments (and does not accept the start_cond as first argument). The task decorator is actually implemented on the FuncTask level thus the create_task can do that but usually it's less practical. Instead, you can pass the func as an argument:

from rocketry.conds import daily

def do_things():
    ...

@app.task(daily, execution="main")
def create_do_things():
    app.session.create_task(func=do_things, start_cond=daily)

I used main here and created only one task per success but you could do an infinite loop, use thread or async and set the task as permanent (so there is no default timeout):

from rocketry.conds import daily
from rocketry.args import TerminationFlag
from rocketry.exc import TaskTerminationException

def do_things():
    ...

def has_new_task():
    ...

@app.task(daily, execution="thread", permanent_task=True)
def create_do_things(flag=TerminationFlag()):
    while True:
        if has_new_task():
            app.session.create_task(func=do_things, start_cond=daily)
        if flag.is_set():
            raise TaskTerminationException("Task was terminated")

Note that we used TerminationFlag and TaskTerminationException to make sure the task will end when the scheduler ask it to end and that the scheduler knows that the task was terminated (instead of succeeded). You don't need those with async but you need to use the await and release the execution at some point.

Also note that if you use the same function for multiple tasks, pass the name argument so that there are no name conflicts between the tasks.

Is this the direction you are looking for?

[Miksus]

EDIT: Forgot to say that set the permanent_task as True to indicate the task should not timeout and that the shut down won't wait till the finish of the task (which will never happen without termination).

[RhetTbull]

Thanks -- I'll play around with this to see if it fits my use case!

[Miksus]

Also, as of 2.4.0, you could do a task queue and event-based system with Rocketry as well. However, there are not much tools for those (yet) but there is the run method in tasks that sets the task to run:

from rocketry.conds import daily
from rocketry.args import TerminationFlag
from rocketry.exc import TaskTerminationException
import asyncio

@app.task()
def do_things(arg):
    ...

async def get_next_task():
    "Infinite task queue (that waits when there is no tasks to run)"
    while True:
        await asyncio.sleep(2)
        yield 'do_things', {"arg": "run specific arg"}

@app.task(daily, execution="async")
async def task_queue():
    for task_name, params in get_next_task():
        task = app.session[task_name]
        task.run(**params)

But of course there are no premade tools for such queues and events (yet).