longtasks/README.md

longtasks

Usage

Use with ahserver, you need to save a instance of LongTasks or its child class to ServerEnv, in child class, implements the process_task method for your business

save LongTasks instace to ServerEnv:

from ahserver.serverenv import ServerEnv
from longtasks.longtasks import Longtasks
from appPublic.worker import schedule_once

class MyTasks(LongTasks):
	# Child class
    async def process_task(self, payload):
        # use logic to execute task
		pass

def load_longtasks()
    longtasks = MyTasks('redis://127.0.0.1:6379', 'example')
    env = ServerEnv()
    env.longtasks = longtasks
    # run the backend job 
	schedule_once(0.1, longtasks.run)

submit a task in dspy

payload = {
	'prompt':'gagagag'
}
x = await longtasks.submit_task(payload)
# x is a dict with has a 'task_id' key
return x

query task status

taskid = 'mytaskid'
task_status = await longtasks.get_status(taskid)
return task_status