@Marvin What is the recommended way to save or store information about why a flow failed? My use case involves flows that perform external calls and validation. These flows may fail due to internal criteria, such as insufficient data from the given ID or poorly prepared inputs. When a flow raises an exception and fails, I want to save the message content somewhere for easy retrieval later from the REST API. I'm not sure if I should use some state or artifact to store this information. What is the recommended way to handle this situation? and how can I do it?

To store information about why a flow failed, Prefect recommends using its mechanisms for logging, state handling, and artifacts. Here's what you can do: - Logging: Prefect's logging system allows you to log error messages when exceptions occur. You can use this system to record the reasons for failure. Logs can be examined via the Prefect UI or programmatically retrieved via the REST API. - State Handlers: Implement state handlers for your tasks and flows. These are functions that get called whenever a Task changes state. Your state handler function can capture exceptions and their messages, which you can then log or persist elsewhere. - Artifacts: You can create artifacts during a flow run to record more structured or detailed information about errors. Artifacts allow you to publish data from task and flow runs, and can be links, Markdown or tables. Artifacts are visible in the Prefect UI, which makes it easy to access and share information about errors and their context. For more information about handling errors and storing information related to flow failures, you can check Prefect's documentation on writing logs, creating and using artifacts, and parsing error summaries powered by Marvin AI, which can help you understand the root cause of flow run failures. Unfortunately, direct information about implementing state handlers could not be found, but you can typically locate examples and best practices for state handlers in the Prefect documentation. By combining these approaches, you can effectively record, store, and retrieve the reasons behind flow failures, and also provide easy access to this information for later analysis or remediation.

@Marvin show me some code examples on how to implement this using logging. An example of writing the log and how to retrieve this specific log (not all others) from the REST API

To log error messages in Prefect, you can use Python’s built-in

print

statements or the

prefect.logging

module as shown in the following examples. By setting

log_prints=True

in your flow decorator, any

print

statements will be logged as INFO logs:

python
from prefect import flow

@flow(log_prints=True)
def my_workflow():
    # This will be logged as an INFO message
    print("This is a log message.")

if __name__ == "__main__":
    my_workflow()  # call it like a normal python function

In the case of exceptions, you can log them as follows:

python
import prefect
from prefect import task, flow

@task
def task_that_may_fail(some_param):
    if some_param:
        raise ValueError("This is an error message that will be logged.")

@flow(log_prints=True)
def my_workflow_with_exception():
    try:
        task_that_may_fail(False)
    except ValueError as e:
        # This will log the exception message as an ERROR
        prefect.logger.error(f"An error occurred: {e}")

if __name__ == "__main__":
    my_workflow_with_exception()

Regarding retrieving specific logs, Prefect does not have code examples available in their repository for fetching logs by a specific criteria (such as error messages only) via the REST API. However, you can do so by querying the REST API for logs associated with a particular flow run ID and filtering them to find error-level logs or logs containing specific error messages. You can use an endpoint of Prefect's REST API to fetch logs, like this hypothetical example:

python
import requests

flow_run_id = "your-flow-run-id"
prefect_api_url = "<http://your-prefect-api-url>"
response = requests.get(
    f"{prefect_api_url}/flow_run/{flow_run_id}/logs"
)

logs = response.json()
error_logs = [log for log in logs if "ERROR" in log["level"]]

for log in error_logs:
    print(log["message"])

For the actual API endpoint and parameters, you would need to refer to the Prefect REST API documentation, which is version-specific and hence necessitates checking the appropriate endpoint in the version of Prefect you work with. Keep in mind that these are simple examples and you might need to handle authentication, pagination, and error handling when dealing with real API requests.

yeah I think i would reach for artifacts here e.g.

create_markdown_artifact(key=..., markdown=...)