<@ULVA73B9P> How do I read a flow result using the...
# marvin-in-the-wildp
Philipp
07/19/2024, 7:26 AM@Marvin How do I read a flow result using the python sdk?
m
Marvin
07/19/2024, 7:26 AMYou can access the result of a flow using the Prefect Python SDK. Here is how you can do it:
Detailed Steps:
Copy code
python
import asyncio
from prefect import flow, task
@task
async def my_task():
return 1
@flow
async def my_flow():
state = await my_task(return_state=True)
result = await state.result(fetch=True)
return result + 1
async def main():
state = await my_flow(return_state=True)
assert await state.result(fetch=True) == 2
asyncio.run(main())return_state=Trueawait state.result(fetch=True)p
Philipp
07/19/2024, 7:28 AM@Marvin How does this work with wait_for_flow_run and the client library?
How do I read persistent result?
m
Marvin
07/19/2024, 7:28 AMTo read the result of a flow run using the
wait_for_flow_runwait_for_flow_run Copy code
python
import asyncio
from prefect.client import get_client
from prefect import flow, task
from prefect.states import State
@task(persist_result=True)
async def my_task():
return 42
@flow
async def my_flow():
return await my_task()
async def main():
async with get_client() as client:
# Create a new flow run
flow_run = await client.create_flow_run_from_deployment(
deployment_id="<deployment_id>",
parameters={}
)
# Wait for the flow run to complete
flow_run = await client.wait_for_flow_run(
flow_run.id,
retries=5,
retry_delay_seconds=10
)
# Fetch the state result
state = await client.read_flow_run(flow_run.id)
if state.state.result:
result = state.state.result()
print(f"Flow run result: {result}")
else:
print("No result found for the flow run")
asyncio.run(main()) Copy code
python
import asyncio
from prefect import flow, task
from prefect.client import get_client
@task(persist_result=True)
async def my_task():
return "Persistent Result"
@flow(persist_result=True)
async def my_flow():
return await my_task()
async def main():
async with get_client() as client:
state = await my_flow(return_state=True)
# Retrieve the persistent result
if state.state.result:
result = await state.state.result().get() # Automatically retrieves and deserializes the persisted result
print(f"Persistent result: {result}")
else:
print("No persistent result found")
asyncio.run(main())24 Views