<@ULVA73B9P> Is there a way to redirect Python's l...
# ask-marvin
a
@Marvin Is there a way to redirect Python's logging to Prefect's logger within a task?
m
thought for 232.9 seconds
Yes. In Prefect 3.x you can funnel standard Python logging into the current task’s (or flow’s) run logger. Preferred: use LogEavesdropper (built-in) - LogEavesdropper attaches a temporary handler to any stdlib logger name and forwards records to the current run logger. - Great for capturing logs from your own module or third-party libs (e.g., urllib3, boto3). Example:
Copy code
import logging
from prefect import task
from prefect.logging import get_run_logger, LogEavesdropper

@task
def my_task():
    run_logger = get_run_logger()

    # your module logger
    app_logger = logging.getLogger("my_app")

    # third-party example
    http_logger = logging.getLogger("urllib3")

    <http://run_logger.info|run_logger.info>("This is a Prefect run log")

    # Forward logs from 'my_app'
    with LogEavesdropper("my_app", level=<http://logging.INFO|logging.INFO>):
        <http://app_logger.info|app_logger.info>("This stdlib log will appear in Prefect task logs")

    # Forward logs from 'urllib3'
    with LogEavesdropper("urllib3", level=logging.WARNING):
        http_logger.warning("Captured third-party warning")
Notes and caveats: - Use it inside a running task/flow; that’s where the run logger is available. - Target specific logger names to avoid noisy or duplicate logs. - If you see duplicates, ensure the source logger isn’t also writing to stdout/stderr or to handlers that Prefect already captures; you may need to adjust
propagate
or handlers on the source logger. Alternative: custom handler that forwards to get_run_logger If you need finer control, you can add a handler that forwards stdlib logs to Prefect’s run logger. Be careful to avoid recursion and remember to remove the handler after use.
Copy code
import logging
from prefect import task
from prefect.logging import get_run_logger

class PrefectRunHandler(logging.Handler):
    def emit(self, record):
        # avoid loops by skipping Prefect’s own logs
        if record.name.startswith("prefect"):
            return
        logger = get_run_logger()
        msg = self.format(record)
        logger.log(record.levelno, msg)

@task
def my_task():
    run_logger = get_run_logger()
    <http://run_logger.info|run_logger.info>("Starting")

    lib_logger = logging.getLogger("my_lib")
    handler = PrefectRunHandler()
    lib_logger.addHandler(handler)
    try:
        <http://lib_logger.info|lib_logger.info>("Forwarded to Prefect run log")
    finally:
        lib_logger.removeHandler(handler)
Bonus: redirect print to Prefect logs
Copy code
from prefect.logging.loggers import patch_print

@task
def my_task():
    patch_print()
    print("This will show in Prefect logs")
References - Logging guide: Prefect Logging guide - Source for logging helpers (get_run_logger, LogEavesdropper, patch_print): prefect/logging/loggers.py If you can share which logger(s) you want to capture (root vs. specific libraries), I can suggest the safest setup to avoid duplicates.