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.

Prefect Community

<@ULVA73B9P> how can I access unpersisted result of a flow in the on complete hook?

To access the unpersisted result of a flow in the `on_complete` hook in Prefect 3.x, you can utilize the `state` object passed to the hook. The `state` object contains information about the flow's execution, including the result if the flow completed successfully. Here's a concise way to do it:

1. *Define the `on_complete` Hook*: Create a function that will act as your `on_complete` hook. This function will receive the flow, flow run, and state as arguments.

2. *Access the Result*: Within the hook, you can check if the state is completed and access the result from the state.

Here's a simple example:

```python
from prefect import flow, task, State

@task
def my_task():
    return "Hello, World!"

def on_complete_hook(flow, flow_run, state: State):
    if state.is_completed():
        result = state.result()
        print(f"Flow completed with result: {result}")

@flow(on_completion=[on_complete_hook])
def my_flow():
    my_task()

my_flow()
```

In this example, the `on_complete_hook` accesses the result of the flow execution if it is completed, and prints it out.

For more details on managing results and states in Prefect, you can refer to the <https://docs-3.prefect.io/v3/develop/results|Prefect 3.x documentation on results> and <https://docs-3.prefect.io/v3/develop/manage-states|state management>. 

If you have any specific configurations or setups, feel free to share, and I can provide more tailored guidance.

<@ULVA73B9P> this doesn’t quite work when I use .submit to run the task

When using `.submit()` to run a task in Prefect, the task runner creates a `PrefectFuture` object. This object provides access to the state and result of the task. In the `on_complete` hook, you can access the result of a future using the `.result()` method of the `PrefectFuture`.

Here's a concise example of how you might handle this:

1. *Submit the Task*: Use `.submit()` to run your task, which returns a `PrefectFuture`.

2. *Access the Result in the Hook*: In the `on_complete` hook, use the `.result()` method on the `PrefectFuture` to access the task's result.

Example:

```python
from prefect import flow, task, State

@task
def my_task():
    return "Hello, Future!"

def on_complete_hook(flow, flow_run, state: State):
    if state.is_completed():
        # Assuming the state.result() returns a PrefectFuture
        future_result = state.result()
        result = future_result.result()  # Access the actual result
        print(f"Flow completed with result: {result}")

@flow(on_completion=[on_complete_hook])
def my_flow():
    future = my_task.submit()
    return future  # Return the future from the flow

my_flow()
```

In this example, the `on_complete_hook` accesses the result of the task using the `.result()` method on the `PrefectFuture` returned by `state.result()`.

For more details on using `PrefectFuture` and task runners, you can refer to the <https://docs-3.prefect.io/v3/develop/task-runners|Prefect documentation on task runners>.