@Marvin how to intercept and log custom exception when flow was failed?

To intercept and log custom exceptions when a Prefect flow fails, here's how you can proceed depending on your Prefect version: ### For Prefect 2.x You can utilize the

on_failure

callback for tasks to handle and log exceptions. Here is an example of how you can achieve that:

python
from prefect import task, Flow
import logging

logger = logging.getLogger("prefect")

@task
def failing_task():
    raise ValueError("This task failed.")
   
@task
def log_failure(exception):
    # Log custom exception message
    logger.error(f"Task failed with exception: {exception}")
  
with Flow("example-flow") as flow:
    try:
        failing_task()
    except Exception as e:
        log_failure(e)

flow.run()

### For Prefect 3.x In Prefect 3.x, the flows and tasks are managed differently. You can still use similar concepts but within the new architecture.

python
from prefect import flow, task, get_run_logger

@task
def failing_task():
    raise ValueError("This task failed.")

@flow(name="example-flow")
def example_flow():
    logger = get_run_logger()
    try:
        failing_task()
    except Exception as e:
        logger.error(f"Task failed with exception: {e}")

example_flow()

### Logging Detailed Information To get detailed information from your logs, ensure you set the logging level to DEBUG:

bash
# Using the CLI
prefect config set PREFECT_LOGGING_LEVEL=DEBUG
# Using environment variables
export PREFECT_LOGGING_LEVEL=DEBUG

For more details, you can refer to the Prefect documentation: - Prefect 2.x: Prefect 2.x Flows - Prefect 3.x: Prefect 3.x Writing Flows Feel free to ask if you have further questions!

@Marvin how to add additional handlers to prefect logging?

To add additional handlers to Prefect logging, you can follow the steps below based on your Prefect version. ### Prefect 2.x You can customize the logging configuration by modifying the logging configuration file or by setting environment variables. #### Example using Environment Variables 1. Set environment variables to adjust the logging levels or add custom handlers:

bash
export PREFECT_LOGGING_LEVEL="DEBUG"
export PREFECT_LOGGING_HANDLERS_API_LEVEL="ERROR"

2. Alternatively, customize the

logging.yml

configuration file (usually located at

~/.prefect/logging.yml

). #### Example using

logging.yml

Here's a basic example of adding a custom file handler:

yaml
version: 1
handlers:
  file_handler:
    class: logging.FileHandler
    level: INFO
    formatter: standard
    filename: my_prefect.log
loggers:
  prefect:
    handlers: [console, file_handler]
    level: DEBUG
    propagate: yes

root:
  level: WARNING
  handlers: [console, file_handler]

### Prefect 3.x In Prefect 3.x, you can build custom handlers by subclassing

PrefectConsoleHandler

and creating tailored loggers. #### Example using a Custom Logger 1. Create a custom handler in your script or module:

python
import logging
from prefect.logging.handlers import PrefectConsoleHandler

class CustomFileHandler(PrefectConsoleHandler):
    def __init__(self, filename, *args, **kwargs):
        super().__init__(*args, **kwargs)
        file_handler = logging.FileHandler(filename)
        self.logger.addHandler(file_handler)

# Example usage in a flow
from prefect import flow, get_run_logger

@flow
def my_flow():
    logger = get_run_logger()
    custom_handler = CustomFileHandler("my_prefect.log")
    logger.addHandler(custom_handler)
    <http://logger.info|logger.info>("Hello, world!")

my_flow()

2. Configure Prefect's logging via environment variables:

bash
export PREFECT_LOGGING_LEVEL="DEBUG"
export PREFECT_LOGGING_LOGGERS_PREFECT_FLOW_RUNS_LEVEL="ERROR"

For more detailed information, refer to the Prefect documentation: - Prefect 2.x: Prefect 2.x Logging - Prefect 3.x: Prefect 3.x Logging Feel free to ask if you have more questions!