@Marvin how do I get prefect to show logs with loguru

To integrate

loguru

with Prefect and show logs, you could try the following methods: 1. Configuring Log Levels: You might need to configure the log levels for Prefect and other libraries to ensure their logs are captured and shown. Prefect captures logs by default, but you might need to adjust the log levels used by loggers from other libraries. You can override logging configuration by setting environment variables or Prefect Profile settings using the syntax

PREFECT_LOGGING_[PATH]_[TO]_[KEY]

. 2. Using Prefect's Logging: Prefect already captures logs for your flows and tasks, which you can view and filter in the Prefect UI or via the API. To increase the log verbosity, you might set the appropriate environment variable to change the default logging levels for Prefect to

DEBUG

or another level that might capture

loguru

logs. The documentation on logging within Prefect can be found here. However, the specific integration with

loguru

isn't explicitly detailed in the provided excerpts, and it's possible that further custom configuration would be necessary to integrate

loguru

-generated logs with Prefect's system. For more detailed help on integrating

loguru

, you might consider the Prefect Community Slack or GitHub to see if others have solved this integration issue. If you have specific issues or errors while trying to set this up, searching the GitHub issues or Prefect documentation with details of those issues would be the next step.

I use this:

import functools
from typing import Any, Callable, TypeVar

import prefect
from loguru import logger
from prefect import flow, get_run_logger
from prefect.blocks.core import Block
from prefect.blocks.system import Secret
from prefect.exceptions import MissingContextError
from prefect.runner.storage import GitRepository

T = TypeVar("T", bound=Callable[..., Any])


def forward_loguru_logs(func: T) -> T:
    """
    Make sure you put the @flow or @task decorator at the top.

    Usage:
        @flow
        @forward_loguru_logs
        def my_flow():
            ...

        @task
        @forward_loguru_logs
        def my_task():
            ...
    """

    @functools.wraps(func)
    def wrapper(*args, **kwargs) -> Any:
        enable_loguru_support()
        return func(*args, **kwargs)

    return wrapper


def enable_loguru_support() -> None:
    """
    Redirect loguru logging messages to the prefect run logger.
    This function should be called from within a Prefect task or flow before calling any module that uses loguru.
    """
    from loguru import logger

    def add_handler(loguru_level: str, prefect_level):
        """Helper function to add a handler for a specific log level."""
        logger.add(
            prefect_level,
            filter=lambda record: record["level"].name == loguru_level,
            level="DEBUG",  # minimum level of log messages that this handler will capture and redirect.
            format="{name}:{function}:{line} - {message}",
        )

    try:
        if prefect.runtime.flow_run.parent_flow_run_id is not None:
            # Don't set up loguru support for subflows, as it should already be set up by the parent flow,
            # and we don't want to duplicate logs
            return

        run_logger = get_run_logger()
        logger.remove()  # Clear existing loguru handlers

        # Define log levels and their corresponding Prefect logger methods
        log_levels = {
            "DEBUG": run_logger.debug,
            "WARNING": run_logger.warning,
            "ERROR": run_logger.error,
            "CRITICAL": run_logger.critical,
        }

        # Add handlers for each log level
        for loguru_level, prefect_level in log_levels.items():
            add_handler(loguru_level, prefect_level)

        # Special handling for INFO level to avoid duplication with other levels
        logger.add(
            run_logger.info,
            filter=lambda record: record["level"].name
            not in ["DEBUG", "WARNING", "ERROR", "CRITICAL"],
            level="DEBUG",  # minimum level of log messages that this handler will capture and redirect.
            format="{name}:{function}:{line} - {message}",
        )
    except MissingContextError:
        # we can't test tasks that use get_run_logger as it doesn't have the context, so just default to using the loguru logger
        logger.warning(
            "Prefect context not found, using loguru logger instead of prefect run logger."
        )
        return logger

Then use these decorators to decorate the flow only (not tasks as well):

@flow
@forward_loguru_logs

nice!

No worries