Empty workflow run output, or: how to get the task output of a workflow run?

[sanzoghenzo]

Hi! first of all, many thanks for this great project!

I'm evaluating it with a simple example that mimics most of my intended usage (for now): basically a REST API to start tasks and retrieve their status and result.

I can successfully start a task as instructed by the user guide, and I can retrieve the status of the job via hatchet.runs.aio_get_status, but when the task finishes the hatchet.runs.aio_get_result returns an empty payload, even though I see the correct result in Hatchet dashboard (in the "Output" tab of the run).
Looking at the network calls of the dashboard, I saw that indeed the output field of the /api/v1/stable/workflow-runs/{id} is empty, and the info I'm looking for is inside the first element of the tasks array, or more directly at the /api/v1/stable/tasks/{id} endpoint.

My question is: what is the cleanest option to get the result?

I still have to go through the task orchestration part, but I assume that the output because a workflow run can be composed by multiple tasks, so I suppose I cannot rely on the hatchet.runs.aio_get to then get the first (or last?) task result, or can I?

My worker is set up like this:

import asyncio

from hatchet_sdk import Context
from hatchet_sdk import Hatchet
from pydantic import BaseModel


class FirstTaskInput(BaseModel):
    message: str
    delay: int


hatchet = Hatchet(debug=True)


@hatchet.task(name="FirstTask", input_validator=FirstTaskInput)
async def first_task(input: FirstTaskInput, ctx: Context) -> dict[str, str]:
    await asyncio.sleep(input.delay)
    return {"message": input.message}


def main() -> None:
    worker = hatchet.worker("test-worker", workflows=[first_task])
    worker.start()


if __name__ == "__main__":
    main()

and my FastAPI server is like this:

from typing import Any

import uvicorn
from fastapi import FastAPI
from fastapi.responses import RedirectResponse
from hatchet_sdk import V1TaskStatus

from worker import FirstTaskInput
from worker import first_task
from worker import hatchet


app = FastAPI()


@app.post("/task")
async def start_task(input: FirstTaskInput):
    ref = await first_task.aio_run_no_wait(input=input)
    return RedirectResponse(f"/task/{ref.workflow_run_id}", status_code=302)


@app.get("/task/{task_id}")
async def get_task(task_id: str) -> dict[str, Any]:
    status = await hatchet.runs.aio_get_status(task_id)
    if status == V1TaskStatus.COMPLETED:
        return await hatchet.runs.aio_get_result(task_id)
    return {"status": status.value}


def main():
    """Run the application."""
    uvicorn.run(
        "my_project_api:app",
        host="0.0.0.0",
        port=8000,
        reload=True,
        log_level="debug",
    )


if __name__ == "__main__":
    main()

[sanzoghenzo]

For completeness: replacing the aio_get_result with the following works:

run = await hatchet.runs.aio_get(task_id)
return run.tasks[-1].output

While this will be fine for many of my use cases, I'm still dubious if this it the right thing to do...

[mrkaye97]

hey @sanzoghenzo - I think the best thing to do here would be to construct a WorkflowRunRef and use that to get the result of the task / workflow! The simplest way is probably to use hatchet.runs.get_run_ref and pass the workflow run id, and then call await ref.aio_result() in your other endpoint to retrieve the result.

Let me know if that helps!

[mrkaye97]

Also one other thing by the way, I'm not too sure what your objectives are with your API, but it might be preferable to call the run methods (instead of the run_no_wait ones) since they will run "fire and wait" style, meaning they'll wait for the result to come back

[sanzoghenzo]

Hi @mrkaye97 , thanks for the quick reply! I'll try your suggestion ASAP.

Long running task will timeout the request (usuallly client side) if I use the run method, won't they?

A use case I just implemented on one of my project was to generate a word report based on a grafana dashboard given some parameters passed by the user from a web UI. I don't want the web UI/http request to hang until the task is done, so I just redirect to the task status endpoint and poll it until it is done.
I thought this is a frequent use case, but I may be missing something

[mrkaye97]

yes, definitely - in that case, it definitely makes sense to return the ID and then poll for the result! What you're doing should work, this definitely looks like a bug on our end on the REST API. Fortunately, the gRPC API should help here and will be less prone to race conditions, which also pop up in the REST API. Here's an example based on what you have now

status = await hatchet.runs.aio_get_status(workflow_run_id)

if status == V1TaskStatus.COMPLETED:
    ref = hatchet.runs.get_run_ref(workflow_run_id)
    
    try:
        ## set this timeout to whatever you want
        async with asyncio.timeout(3): 
            return await ref.aio_result()
    except asyncio.TimeoutError:
        ## do something helpful here
        pass

the aio_result will hang until the result comes back, but you can have it time out and retry on whatever cadence works for you. Hopefully this helps!

[sanzoghenzo]

Thank you very much @mrkaye97 !

the result is wrapped inside a dictionary with the task name as the key ("firstask" in this example) so I have to extract it, but it is indeed cleaner this way!

Have a wonderful day 😉

[mrkaye97]

Great! I'll write a little helper to make it easier to do that so you don't need to extract the result from the dictionary

[mrkaye97]

This is live in 1.16.0! You should be able to use await your_task.aio_get_result(run_id) now to replace the ref = ... and then the await ref.aio_result()!