Scheduling dynamically generated flows/tasks

[mjarosie]

Hi,

I'm trying to implement a service that builds long-running flows/tasks dynamically and invokes them asynchronously. Here's a very simplified version of my service (based on "How to deploy a web application powered by background tasks"):

from fastapi import FastAPI
from pydantic import BaseModel
from uuid import UUID
from prefect import task
from prefect.client import get_client

@task
def say_hello(name: str):
    print(f"Hello, {name}!")
    return f"Greeted {name}"

def generate_task_dynamically(say_hello_twice: bool):
    @task(name=f"my_dynamic_task")
    def some_dynamic_task(name):
        if say_hello_twice:
            say_hello(name)
            say_hello(name)
        else:
            say_hello(name)

    return some_dynamic_task

app = FastAPI()

dynamic_task = None

class CreateRequest(BaseModel):
    say_hello_twice: bool

@app.post("/workflow/create")
async def create(request: CreateRequest):
    global dynamic_task
    dynamic_task = generate_task_dynamically(request.say_hello_twice)

class SubmitRequest(BaseModel):
    name: str

@app.post("/workflow/submit")
async def submit(request: SubmitRequest):
    future = dynamic_task.delay(request.name)
    return {"task_run_id": str(future.task_run_id)}


@app.get("/workflow/status/{task_run_id}")
async def status(task_run_id: UUID):
    async with get_client() as client:
        task_run = await client.read_task_run(task_run_id)
        return {
            "status": task_run.state.name,
        }

I'd like the flow/task to be run asynchronously (i.e. POST /submit to return almost immediately) and get a result eventually, possibly retrieving it via another, follow-up API call.

Based on my requirements and the docs linked above I figured I should use .delay(). But the issue is - for the task to be picked up by a worker, it needs to be aware of the implementation of the task, so that it can serve() it. That seems to be a blocker - I couldn't find a way to make it work. Is there an alternative approach to make it work?

[desertaxle]

Hey again @mjarosie! You could use FastAPI's background task functionality to run these dynamically generated tasks, but I think that would generally be less durable than using .delay with task workers.

I think it's worth asking why you need to generate tasks dynamically. Could you instead accept arguments to your tasks and handle the dynamic aspects via logic within your task? I'm not sure if that will work with your use case, but that approach could simplify your implementation.