Cullen Baker
04/21/2026, 5:24 PMMarvin
04/21/2026, 5:24 PMMarvin
04/21/2026, 5:30 PMflow_run_logger(parent_flow_run_id). Here’s a small pattern you can drop into your subflows:
from prefect import flow
from prefect.logging import get_run_logger, flow_run_logger
from prefect.context import FlowRunContext
def _both_loggers():
# Returns (child_logger, parent_logger or None, context)
child_logger = get_run_logger()
ctx = FlowRunContext.get()
parent_id = getattr(ctx.flow_run, "parent_flow_run_id", None)
parent_logger = flow_run_logger(parent_id) if parent_id else None
return child_logger, parent_logger, ctx
@flow
def child_flow():
child_log, parent_log, ctx = _both_loggers()
def log_both(msg, level="info"):
getattr(child_log, level)(msg)
if parent_log:
# add a prefix so parent logs are distinguishable
getattr(parent_log, level)(f"[subflow {ctx.flow.name} | {ctx.flow_run.id}] {msg}")
log_both("Child work starting")
# ... your logic ...
log_both("Child work finished")
Notes and best practices:
- This will emit logs twice (once to the child flow run, once to the parent). Consider forwarding only key events to keep parent logs readable.
- Prefix forwarded messages with the subflow name/run ID so parent logs are clear.
- If a flow is invoked top-level (no parent), parent_flow_run_id is None and forwarding is skipped.
Alternative approaches:
- Use a correlation ID (passed from parent to subflow) and query/aggregate logs across runs with the API if you prefer not to duplicate logs.
- You can fetch all related logs programmatically via PrefectClient.read_logs and LogFilter and present/ship them elsewhere.
Helpful docs:
- Subflows and flow runs: Flows (subflows)
- Logging how-to guide: How to add logging
- Loggers API (includes flow_run_logger): prefect.logging.loggers
If you want, I can help you wrap this into a small utility so all your subflows automatically forward selected log levels to the parent. Would you like to forward all logs or just INFO+/errors?