https://prefect.io logo
Title
f

flapili

02/23/2023, 2:32 PM
Hi, how can I get task result from api ? I only get a reference from api and I don't know how to retreve the state from pure python code :/
"data": {
    "type": "reference",
    "serializer_type": "compressed/json",
    "storage_block_id": "e3311389-34e1-491c-aaef-cbcf6e6ea977",
    "storage_key": "e4940b1d2ada42a7a1ac77710cae4d18"
  },
n

Nikhil Joseph

02/23/2023, 2:42 PM
storage_block = Block.load('storage-block')
result = storage_block.read_path(storage_key)
try this. this should give u the raw result data. might be unpickle depending on how you are storing it
f

flapili

02/23/2023, 2:50 PM
yes but that's not really convenient 😕
# coding: utf-8
from uuid import UUID

from fastapi import FastAPI
from prefect import get_client
from prefect.states import StateType

app = FastAPI(title="Tasks results")


@app.get("/flow_run_state/{state_id}/result")
async def get_task_result(state_id: UUID):
    async with get_client() as client:
        states = await client.read_flow_run_states(flow_run_id=state_id)
        for state in states:
            if state.type == StateType.COMPLETED:
                result = state.result()
                print(await result.get())
got
ModuleNotFoundError: No module named '__prefect_loader__'