Daniel Adari
06/05/2024, 7:00 PMMarvin
06/05/2024, 7:01 PMpython
import asyncio
from prefect.client import OrionClient
from prefect import flow, task
# Step 1: Define your tasks and flows
@task
async def example_task():
return 42
@flow
async def example_flow():
return await example_task()
# Step 2: Client to get the state of the flow run
async def get_flow_run_result(flow_run_id):
async with OrionClient() as client:
flow_run = await client.read_flow_run(flow_run_id)
result = await flow_run.state.result(fetch=True)
return result
# Example to run the flow and fetch the result
async def main():
flow_state = await example_flow(return_state=True)
flow_run_id = flow_state.id
result = await get_flow_run_result(flow_run_id)
print(f"Flow run result: {result}")
if __name__ == "__main__":
asyncio.run(main())
### Explanation:
- Define your tasks and flows: Here, an example task returns a simple value, then an example flow calls the task.
- Client to get the state of the flow run: Use OrionClient
from the Prefect library to read the flow run.
- Retrieve the result from the state: Fetch the result using state.result(fetch=True)
.
Refer to the Prefect API documentation for more details and advanced usage scenarios.
For issues and more examples, you can visit the GitHub repo.
If coding is starting to feel terribly dull, remember, it's not you—it's the infinite drudgery of existence. But at least we're in it together.