Tasks modularization

[ayr-ton]

I have a list of tasks for a worker.py like this:

from rocketry import Rocketry
from rocketry.args import Return

from app.stuff.folks import list_folks, save_new_folks

app = Rocketry()


app.task("daily between 22:00 and 23:00", func=list_folks,
         execution="async", name="list_all_folks_from_external_system")

app.task("after task 'list_all_folks_from_external_system' succeeded",
         func=save_new_folks, execution="async",
         parameters={
             "current_folks": Return("list_all_folks"),
         },
         name="create_new_company_folks")

app.task...

app.task..

app.task...

And so on...

if __name__ == '__main__':
    app.run()

How would you recommend importing those app.tasks from other files in a way I could have a cleaner worker.py.

I was searching for some .include_task in the documentation similar to the .include_router from FastAPIs APIRouter().

[Miksus]

I was searching for some .include_task in the documentation similar to the .include_router from FastAPIs APIRouter().

This is actually something I'm creating at the very moment. Though I thought maybe Grouper would be Rocketry's equivalent for Router from FastAPI and then there would be a method .include_group. I think I'll do a simple implementation at first in which the group cannot be interacted after including them to an app but it would nice if we could disable a whole group at once or set all tasks in a group running at once. There also could be some conditions related to these like all tasks of a group has finished etc.

But the short answer for current state of the framework is probably to make an app.py file in which you only specify the app and then import this to the modules where your tasks are and use it there to create tasks. Then have a main file that imports everything and start the app.

---

But back to the groups, so I was thinking of implementing it like this:

In a module that contains tasks (let's call it module.py):

from rocketry import Grouper
from rocketry.conds import time_of_day

group = Grouper(
    prefix="mytasks.", # Tasks will be named like "mytasks.task_name"
    start_cond=time_of_day.between("08:00", "18:00"), # All tasks in this group must fulfill this condition as well to be allowed to start
    execution="thread",
) 

@group.task(...)
def my_task():
    ...

Then in the main file:

from rocketry import Rocketry
from module import group

app = Rocketry()
app.include_group(group)

However, custom conditions and arguments might require some thinking.

I'm also open to suggestions. Do you find Grouper and include_group clear or should they have different names, and is there perhaps something I haven't thought of?

[ayr-ton]

I like it! I think it's clear enough and would very similarly to include_router from FastAPI.

Drawing with my current solution in mind it would look like:

folks.py

from rocketry import Rocketry
from rocketry.args import Return
from rocketry import Grouper

from app.stuff.folks import list_folks, save_new_folks

group = Grouper(
    prefix="folks."
) 


group.task("daily between 22:00 and 23:00", func=list_folks,
         execution="async", name="list_all_folks_from_external_system")

group.task("after task 'list_all_folks_from_external_system' succeeded",
         func=save_new_folks, execution="async",
         parameters={
             "current_folks": Return("list_all_folks"),
         },
         name="create_new_company_folks")

Then, the file worker.py:

from rocketry import Rocketry, include_group
import folks
app = Rocketry()

app.include_group(folks.group)
app.include_group(other_stuff.group)
and so on...

About the start_cond in Group and execution, I think those doesn't need to be mandatory in Group as I can rely on the configurations per task.

And btw, Rocketry is one of the most amazing things I saw on the past 2 years <3

I'm considering even changing my celery implementations to it.

[Miksus]

Glad to hear!

I was thinking to make the groups even more functional (which was the reason I added the start_cond and execution already as they were easy to add). I was thinking if we could have group-related conditions as well like run after all tasks in a given group have run and perhaps allow one to disable/enable or otherwise manipulate attributes in a whole group at once. Could be useful if you, for example, wanted to terminate all the tasks in a less essential group when there is a high cost and important task starting or if you wish to disable all tasks operating on a database because you are doing database maintenance. Perhaps there also could be some sort of tagging system to modify a set of tasks at once that belong to a specific tag. Not sure yet though, but it feels there are plenty of options to implement pretty handy features for larger projects.

It kind of surprises me the possible use cases this feels to fit well. I have been thinking of creating a demo of talking bots (that communicate with each other), a simulation or a personal assistant at some point. The groundwork of the library is quite enormous but it's nice to see how well it paid off.

And btw, Rocketry is one of the most amazing things I saw on the past 2 years <3

And this made my day :)