Gabrielle Bennett
09/21/2024, 9:42 PMprefect.runtime
variables in a failure hook?Marvin
09/21/2024, 9:42 PMprefect.runtime
variables in a failure hook. However, I can provide a general example based on understanding how Prefect hooks and runtime variables typically work.
There are two main versions of Prefect that you might be using: Prefect 2.x and Prefect 3.x. Here's an example of how to access prefect.runtime
variables in a failure hook for both versions:
### Prefect 2.x
In Prefect 2.x, you can use state handlers to specify behavior on task or flow failures. You can access runtime variables like prefect.runtime
within these handlers.
python
from prefect import flow, task
from prefect.runtime import flow_run
@task
def my_task():
raise ValueError("Something went wrong!")
def failure_handler(task, old_state, new_state):
if new_state.is_failed():
run_id = flow_run.id
print(f"Flow run id: {run_id}")
print(f"Task {task.name} failed with message: {new_state.message}")
@flow(on_failure=failure_handler)
def my_flow():
my_task()
if __name__ == "__main__":
my_flow()
### Prefect 3.x
In Prefect 3.x, the setup is similar, but you might have slightly different ways to register state change handlers.
python
from prefect import flow, task
from prefect.runtime import flow_run
from prefect.states import State
@task
def my_task():
raise ValueError("Something went wrong!")
def failure_handler(flow: State, automation_state: State):
if automation_state.is_failed():
run_id = flow_run.id
print(f"Flow run id: {run_id}")
print(f"Flow {flow.name} failed with message: {automation_state.message}")
@flow(on_failure=failure_handler)
def my_flow():
my_task()
if __name__ == "__main__":
my_flow()
Use these patterns to implement failure hooks appropriate for your specific version of Prefect.
For more tailored assistance, please refer to the appropriate Prefect documentation:
- Prefect 2.x Documentation: Prefect 2.x
- Prefect 3.x Documentation: Prefect 3.x
Let me know if you need any further help!