<@ULVA73B9P> Hotw to catch logs from dependencies ...
# ask-marvin
r
@Marvin Hotw to catch logs from dependencies during flow execution in process worker ?
m
thought for 231.0 seconds
Great question. In Prefect 3.x, there are three knobs to capture logs from dependencies when running with the Process worker: 1) Capture print/stdout/stderr from your flow/tasks - Enable print capture so anything that writes to stdout/stderr (including libs that print) is sent to Prefect logs: - Per-flow: use
@flow(log_prints=True)
- Globally: set env var
PREFECT_LOGGING_LOG_PRINTS=True
2) Capture Python logging from third‑party libraries - Tell Prefect to attach its API handler to specific library loggers via `PREFECT_LOGGING_EXTRA_LOGGERS`: - Example:
PREFECT_LOGGING_EXTRA_LOGGERS=requests,urllib3,boto3,sqlalchemy
- Make sure the libraries’ logger levels are not higher than your desired level. Optionally set
PREFECT_LOGGING_LEVEL=INFO
(or DEBUG) or set each library’s level in code. 3) Understand Process worker output vs Prefect logs - Process work pool/worker has
stream_output
which streams a flow’s stdout/stderr to the worker console. This is helpful for live debugging, but it does not by itself send those lines to the Prefect UI. To see them in the UI, use
log_prints=True
(or log explicitly with
get_run_logger()
). Minimal setup that works well for most cases: - In your Process work pool or worker environment, set:
Copy code
PREFECT_LOGGING_LOG_PRINTS=True
PREFECT_LOGGING_EXTRA_LOGGERS=requests,urllib3,boto3,sqlalchemy
PREFECT_LOGGING_LEVEL=INFO
- In code:
Copy code
from prefect import flow
from prefect.logging import get_run_logger
import logging, requests

@flow(log_prints=True)
def my_flow():
    # Optional: ensure third-party logger levels are not too high
    logging.getLogger("requests").setLevel(<http://logging.INFO|logging.INFO>)

    logger = get_run_logger()
    print("stdout via print will appear in Prefect logs")
    <http://logger.info|logger.info>("Structured log via Prefect run logger")

    # Library logs (if included in PREFECT_LOGGING_EXTRA_LOGGERS) will appear
    requests.get("<https://example.com>")

if __name__ == "__main__":
    my_flow()
If you spawn subprocesses or use multiprocessing: - For shell/subprocess output: capture and log it explicitly, or rely on
log_prints=True
if you route it to stdout.
Copy code
import subprocess
from prefect import flow
from prefect.logging import get_run_logger

@flow(log_prints=True)
def run_cmd():
    logger = get_run_logger()
    r = subprocess.run(["python", "-V"], capture_output=True, text=True)
    if r.stdout:
        <http://logger.info|logger.info>(r.stdout.strip())
    if r.stderr:
        logger.error(r.stderr.strip())
- For multiprocessing where child processes need to log to the flow run, wrap functions with Prefect context (
prefect.context.with_context
) so
get_run_logger()
works in child processes. Docs for more detail: - Add logging to flows and tasks: How to add logging - Customize logging (extra loggers, config): Logging customization Quick follow-up: which dependencies are you trying to capture, and are they using Python’s
logging
module or printing to stdout/stderr (or running shell commands)? I can tailor the snippet for your exact case.