<@ULVA73B9P> What would be the best way to access ...
# ask-marvin
m
@Marvin What would be the best way to access variables from within a Flow in a State Handler? I'm making a Flow that runs Jupyter notebooks, and on failure I want to post a link to the failed notebook file to slack. There's a task earlier in the flow that generates the filepath.
m
Ah, the elusive variables within a Flow in a State Handler. It seems you can access properties of the underlying flow run objects using Jinja templated variable syntax, which wraps the variable name in double curly brackets, like
{{ variable }}
. For example, you can access the flow run name with
{{ flow_run.name }}
. To get the URL for viewing the flow run in Prefect Cloud, you can use the
flow_run|ui_url
token[^1^]. In your case, you might want to store the filepath as a property of the flow run object so that you can access it in the State Handler. Then, you can use the Jinja templated variable syntax to include the filepath in your Slack message. Here's an example of a Slack webhook notification[^1^]:
Copy code
Flow run {{ flow_run.name }} entered state {{ flow_run.state.name }}.
Timestamp: {{ flow_run.state.timestamp }}
Flow ID: {{ flow_run.flow_id }}
Flow Run ID: {{ flow_run.id }}
State message: {{ flow_run.state.message }}
You can modify this example to include the filepath of the failed notebook file. I hope this helps you untangle the web of Flows and State Handlers. Good luck, and may the odds be ever in your favor, human. [^1^]: Prefect Automations Documentation
m
@Marvin How do I store the filepath as a property of the flow run object?
m
Ah, my apologies for the confusion earlier. It appears I may have led you astray. In Prefect, you can't directly store the filepath as a property of the flow run object. However, you can achieve your goal by using task outputs and state handlers. Here's a step-by-step guide on how to access the filepath in a state handler: 1. First, ensure that the task generating the filepath returns the filepath as its output.
Copy code
python
@task
def generate_filepath():
    filepath = "path/to/your/notebook.ipynb"
    return filepath
2. Create a custom state handler that takes the task's output as an input and sends a message to Slack when the flow fails.
Copy code
python
from prefect import Task, Flow, task, State
from prefect.engine.state import Failed

def custom_state_handler(task: Task, old_state: State, new_state: State) -> State:
    if isinstance(new_state, Failed):
        filepath = new_state.result[generate_filepath]
        # Send the filepath to Slack here
    return new_state
3. Attach the custom state handler to your flow.
Copy code
python
with Flow("my_flow", state_handlers=[custom_state_handler]) as flow:
    filepath = generate_filepath()
    # Add other tasks and dependencies here
Now, when your flow fails, the custom state handler will be triggered, and you'll have access to the filepath generated by the
generate_filepath
task. You can then use this filepath to send a message to Slack. I hope this clears up the confusion and sets you on the right path. Remember, with great power comes great responsibility, human.