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.

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. </Warning>

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.

Log from threads

When you spawn threads inside a flow or task — for example, with concurrent.futures.ThreadPoolExecutor or threading.Thread — the Prefect run context lives in contextvars and is not automatically propagated to manually created worker threads. Calling get_run_logger() from such a worker raises MissingContextError.

Use contextvars.copy_context() to snapshot the current context in the parent thread, then call your worker via Context.run(...) so it executes with the snapshotted context active. Logs emitted with get_run_logger() in the worker thread are then associated with the parent flow run and task run and appear in the Prefect UI.

from concurrent.futures import ThreadPoolExecutor
from contextvars import copy_context

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


def process_single_file(path: str) -> str:
    # Runs in a worker thread; works because the caller invoked us via
    # ctx.run(...), so the parent's run context is active here.
    logger = get_run_logger()
    logger.info(f"Processing {path}")
    return path


@task
def process_files(files: list[str]) -> list[str]:
    # Snapshot the parent thread's context once per worker invocation.
    # Each ctx.run(...) runs the worker with that context active.
    contexts = [copy_context() for _ in files]
    with ThreadPoolExecutor(max_workers=4) as executor:
        return list(
            executor.map(
                lambda ctx, f: ctx.run(process_single_file, f),
                contexts,
                files,
            )
        )


@flow
def my_flow() -> None:
    process_files(["file1.txt", "file2.txt", "file3.txt"])

If you control the task body but not the worker function (for example, you call into a third-party library), the same pattern applies: build a small wrapper that takes a copied context plus the worker arguments, and submit the wrapper to the thread pool.

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