How to add logging to a workflow

Emit custom logs

To emit custom logs, use get_run_logger from within a flow or task.

from prefect import flow, task
from prefect.logging import get_run_logger


@task(name="log-example-task")
def logger_task():
    # this logger instance will emit logs 
    # associated with both the flow run *and* the individual task run
    logger = get_run_logger()
    logger.info("INFO level log message from a task.")
    logger.debug("DEBUG level log message from a task.")


@flow(name="log-example-flow")
def logger_flow():
    # this logger instance will emit logs
    # associated with the flow run only
    logger = get_run_logger()
    logger.info("INFO level log message.")
    logger.debug("DEBUG level log message.")
    logger.error("ERROR level log message.")
    logger.critical("CRITICAL level log message.")

    logger_task()

The logger returned by get_run_logger support the standard Python logging methods. Any logs emitted by the logger will be associated with the flow run or task run they are emitted from and sent to the Prefect backend. Logs sent to the Prefect backend are visible in the Prefect UI.

get_run_logger() can only be used in the context of a flow or task run. Calling it outside that context — for example, inside a state change hook — raises a MissingContextError.

To log from a state change hook and have logs appear in the Prefect UI, use flow_run_logger or task_run_logger from prefect.logging.loggers. See state change hook for more details.

To use a normal Python logger anywhere with your same configuration, use get_logger() from prefect.logging. The logger retrieved with get_logger() will not send log records to the Prefect API.

Log with print statements

To send print statements to the Prefect backend as logs, set the log_prints kwarg to True on the flow or task.

from prefect import task, flow

@task
def my_task():
    print("we're logging print statements from a task")

@flow(log_prints=True)
def my_flow():
    print("we're logging print statements from a flow")
    my_task()

The log_prints kwarg is inherited by default by nested flow runs and tasks. To opt out of logging print statements for a specific task or flow, set log_prints=False on the child flow or task.

from prefect import task, flow

@task(log_prints=False)
def my_task():
    print("not logging print statements in this task")

@flow(log_prints=True)
def my_flow():
    print("we're logging print statements from a flow")
    my_task()

You can configure the default log_prints setting for all Prefect flow and task runs through the PREFECT_LOGGING_LOG_PRINTS setting:

prefect config set PREFECT_LOGGING_LOG_PRINTS=True

Log from subprocesses

When you spawn subprocesses inside a flow or task — for example, with multiprocessing.Pool or concurrent.futures.ProcessPoolExecutor — the Prefect run context is not automatically available in the child process. This means get_run_logger() raises a MissingContextError.

Use with_context from prefect.context to propagate the current run context into subprocess workers. Logs emitted with get_run_logger() in the child process are associated with the parent flow run and task run and appear in the Prefect UI.

import multiprocessing
from prefect import flow, task
from prefect.context import with_context
from prefect.logging import get_run_logger


def process_item(item):
    logger = get_run_logger()
    logger.info(f"Processing {item}")
    return item * 2


@task
def parallel_task(items):
    with multiprocessing.Pool() as pool:
        return pool.map(with_context(process_item), items)


@flow
def my_flow():
    results = parallel_task([1, 2, 3, 4])

with_context also works with concurrent.futures.ProcessPoolExecutor and multiprocessing.Process.

Access logs from the command line

You can retrieve logs for a specific flow run ID using Prefect’s CLI:

prefect flow-run logs MY-FLOW-RUN-ID

This can be particularly helpful if you want to access the logs as a local file:

prefect flow-run logs  MY-FLOW-RUN-ID > flow.log