Hi, how can I get task result from api ? I only get a reference from api and I don't know how to ret...

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 :/

Copy code

"data": {
    "type": "reference",
    "serializer_type": "compressed/json",
    "storage_block_id": "e3311389-34e1-491c-aaef-cbcf6e6ea977",
    "storage_key": "e4940b1d2ada42a7a1ac77710cae4d18"
  },

Nikhil Joseph

02/23/2023, 2:42 PM

Copy code

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

flapili

02/23/2023, 2:50 PM

yes but that's not really convenient 😕

flapili

02/23/2023, 2:50 PM

Copy code

# 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())

flapili

02/23/2023, 2:50 PM

got

ModuleNotFoundError: No module named '__prefect_loader__'

Open in Slack

Previous Next

Prefect Community

Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.