Great question! In Prefect 3.x you can surface “non-Prefect” logs (standard Python logging or third‑party libraries) in the Prefect UI a few ways. Quick options 1) Easiest (no code): tell Prefect to capture specific libraries - Set this in the environment where your flow actually runs (e.g., your worker or deployment job variables):

PREFECT_LOGGING_EXTRA_LOGGERS=requests,urllib3,boto3,sqlalchemy
PREFECT_LOGGING_LEVEL=INFO

- Then start your worker (or update your deployment) with those env vars. Logs from those libraries will be sent to the Prefect UI during flow/task runs. 2) Programmatic (attach Prefect’s API handler to a logger) - Add Prefect’s APILogHandler to any Python logger inside a flow/task run:

import logging
from prefect import flow
from prefect.logging.handlers import APILogHandler
from prefect.logging import get_run_logger

@flow
def my_flow():
    # Attach once per run for the library you care about
    lib_logger = logging.getLogger("requests")
    lib_logger.setLevel(<http://logging.INFO|logging.INFO>)
    lib_logger.addHandler(APILogHandler())

    logger = get_run_logger()
    <http://logger.info|logger.info>("About to make a request...")

    import requests
    requests.get("<https://httpbin.org/get>")  # requests logs now appear in Prefect UI

3) Capture print() output - If the code uses print (not logging), enable print capture:

from prefect import flow, task

@task(log_prints=True)
def do_work():
    print("This will show up in Prefect logs")

@flow(log_prints=True)
def my_flow():
    print("Flow-level print")
    do_work()

- You can also enable globally via env var:

PREFECT_LOGGING_LOG_PRINTS=True

Helpful notes - Where to set env vars: Set them in the runtime environment (the worker or the deployment’s job env), not just on your laptop. That’s where Prefect loads logging config at run time. - Log levels: If you don’t see messages, increase levels:

PREFECT_LOGGING_LEVEL=DEBUG

…and/or set the specific library logger’s level to DEBUG/INFO. - Standard Python logging: If you already use

logging.getLogger("my_app")

, either add it to

PREFECT_LOGGING_EXTRA_LOGGERS

or add

APILogHandler()

to it during the run. - Context matters: To have logs associated with a specific flow/task run in the UI, attach handlers (or call

get_run_logger()

) inside the flow/task execution. If you tell me which library you want to capture (e.g., requests, sqlalchemy, boto3, spark), I can suggest the exact

PREFECT_LOGGING_EXTRA_LOGGERS

value and a minimal snippet.

Short answer: add a small “bridge” so Loguru sends its messages into Python’s logging, then let Prefect’s handler ship them to the UI. Two easy patterns A) Bridge Loguru → Prefect run logger (simplest, fully contextual) - Add a Loguru sink inside your flow/task that forwards to `get_run_logger()`:

from prefect import flow
from prefect.logging import get_run_logger
from loguru import logger as loguru_logger

@flow
def my_flow():
    run_logger = get_run_logger()

    def to_prefect_sink(message):
        rec = message.record
        level = rec["level"].name.lower()
        emit = getattr(run_logger, level, <http://run_logger.info|run_logger.info>)
        emit(rec["message"])

    sink_id = loguru_logger.add(to_prefect_sink, level="INFO")
    try:
        <http://loguru_logger.info|loguru_logger.info>("Hello from Loguru (shows in Prefect UI)")
        loguru_logger.warning("Something to warn about")
    finally:
        loguru_logger.remove(sink_id)

- Pros: logs are properly tied to the current flow/task run. No extra env vars needed. - Note: add the sink within the run context (inside the flow/task). If you add it outside, there’s no Prefect run context to attach to. B) Bridge Loguru → stdlib logging → Prefect handler - Forward Loguru to Python’s logging, then either add Prefect’s API handler in code or let Prefect attach it automatically. Code-only (attach Prefect handler yourself):

from prefect import flow
from prefect.logging.handlers import APILogHandler
from loguru import logger as loguru_logger
import logging

@flow
def my_flow():
    # Standard logging logger that will receive Loguru records
    py_logger = logging.getLogger("loguru")
    py_logger.setLevel(<http://logging.INFO|logging.INFO>)
    py_logger.addHandler(APILogHandler())  # sends to Prefect UI

    # Bridge Loguru -> stdlib logging
    def stdlib_sink(message):
        rec = message.record
        py_logger.log(rec["level"].no, rec["message"])

    sink_id = loguru_logger.add(stdlib_sink, level="INFO")
    try:
        <http://loguru_logger.info|loguru_logger.info>("Loguru via stdlib handler")
        loguru_logger.debug("This may require DEBUG level to be visible")
    finally:
        loguru_logger.remove(sink_id)

Or with env var (let Prefect attach its handler automatically): - Set this in the environment where the flow actually runs (worker/job env):

PREFECT_LOGGING_EXTRA_LOGGERS=loguru
PREFECT_LOGGING_LEVEL=INFO

- Then only add the Loguru → stdlib bridge in code:

from loguru import logger as loguru_logger
import logging

def stdlib_sink(message):
    rec = message.record
    logging.getLogger("loguru").log(rec["level"].no, rec["message"])

loguru_logger.add(stdlib_sink, level="INFO")

Tips and gotchas - log_prints won’t capture Loguru (it only captures print()). Use one of the bridges above. - If you don’t see messages, bump levels: - In code: set the sink level and/or

py_logger.setLevel(logging.DEBUG)

- Env:

PREFECT_LOGGING_LEVEL=DEBUG

- Remote runs: set env vars on the worker or in your deployment job env. Logging config is loaded at runtime in the worker process. - Multi-process/threads: add the sink in each process that emits logs (e.g., inside the flow/task function). For heavy concurrency, consider

enqueue=True

in Loguru’s

add()

. If you share where your flow runs (local vs worker, Docker/K8s, etc.), I can tailor which bridge and env settings will be most reliable for your setup.

Great question — it depends on how that library emits logs. Here’s a quick decision tree and examples you can drop in. First, are you running Crawlee in Python or calling the Node.js version as a subprocess? A) The library uses Python’s logging (same Python process) - Easiest: let Prefect attach its API handler automatically by listing the logger name. - Set in the environment where the flow runs (worker/job env):

PREFECT_LOGGING_EXTRA_LOGGERS=crawlee
PREFECT_LOGGING_LEVEL=INFO

- Or attach Prefect’s handler in code:

from prefect import flow
from prefect.logging.handlers import APILogHandler
import logging

@flow
def my_flow():
    lib_logger = logging.getLogger("crawlee")  # use the library’s logger name
    lib_logger.setLevel(<http://logging.INFO|logging.INFO>)
    lib_logger.addHandler(APILogHandler())

    <http://lib_logger.info|lib_logger.info>("This will appear in the Prefect UI")

Tip: if you’re not sure of the exact logger name, try common roots (e.g.,

crawlee

,

apify

, or inspect

logging.root.manager.loggerDict

during a run to see what logs are registered). B) The library prints to stdout/stderr (not logging) - If it runs in the same Python process, enable print capture on the flow/task:

from prefect import flow, task

@task(log_prints=True)
def use_library():
    # library prints here
    pass

@flow(log_prints=True)
def my_flow():
    use_library()

- Note:

log_prints

only captures Python prints in the current process. It won’t capture a separate process (see C). C) You call the Node.js Crawlee as a subprocess - Stream its stdout/stderr and forward to the Prefect run logger:

from prefect import flow
from prefect.logging import get_run_logger
import subprocess, threading, sys

def _stream_to_prefect(pipe, level_fn):
    for line in iter(pipe.readline, ""):
        level_fn(line.rstrip("\n"))
    pipe.close()

@flow
def run_crawlee():
    log = get_run_logger()
    proc = subprocess.Popen(
        ["node", "crawler.mjs"],  # your Crawlee entrypoint
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        bufsize=1,
    )

    t_out = threading.Thread(target=_stream_to_prefect, args=(proc.stdout, <http://log.info|log.info>), daemon=True)
    t_err = threading.Thread(target=_stream_to_prefect, args=(proc.stderr, log.warning), daemon=True)
    t_out.start(); t_err.start()

    code = proc.wait()
    t_out.join(); t_err.join()

    if code != 0:
        log.error(f"Crawlee exited with non-zero code: {code}")
        raise RuntimeError(f"Crawlee failed with exit code {code}")

- This reliably surfaces all Crawlee output in the Prefect UI, tagged to the current run. D) The library uses a custom logger (e.g., Loguru or Structlog) - Bridge it to Prefect. Example for Loguru (pattern works for any custom sink):

from prefect import flow
from prefect.logging import get_run_logger
from loguru import logger as loguru_logger

@flow
def my_flow():
    run_logger = get_run_logger()

    def to_prefect_sink(message):
        rec = message.record
        level = rec["level"].name.lower()
        emit = getattr(run_logger, level, <http://run_logger.info|run_logger.info>)
        emit(rec["message"])

    sink_id = loguru_logger.add(to_prefect_sink, level="INFO")
    try:
        <http://loguru_logger.info|loguru_logger.info>("Bridged to Prefect UI")
    finally:
        loguru_logger.remove(sink_id)

Common pitfalls - Set env vars where the flow actually runs (worker/job container), not just locally. - If messages don’t appear, raise levels: - Env:

PREFECT_LOGGING_LEVEL=DEBUG

- Library logger:

setLevel(logging.DEBUG)

or sink level. - Subprocess output won’t be captured by `log_prints`; stream it and log as in (C). If you can share whether you’re using Crawlee for Python vs Node.js, I can tailor the exact snippet (and the correct logger name if it uses stdlib logging).

Great! You’ve got two solid ways to make it “permanent,” depending on where your flows run. Option 1: Prefect profile (persists on your machine) - This sets it for all runs that use your current Prefect profile:

prefect config set PREFECT_LOGGING_EXTRA_LOGGERS=crawlee

- Verify:

prefect config view

Notes: - This affects runs that load this profile (local runs, or workers that also use this profile). - If you’re running flows on remote workers/containers, prefer Option 2. Option 2: Environment variable in the runtime environment (workers/containers) - Set it where the flow actually executes (e.g., worker host, Dockerfile, K8s manifest, CI job): - Shell init (system-wide or user):

export PREFECT_LOGGING_EXTRA_LOGGERS=crawlee

- Dockerfile:

ENV PREFECT_LOGGING_EXTRA_LOGGERS=crawlee

- Kubernetes (Deployment/Job):

env:
  - name: PREFECT_LOGGING_EXTRA_LOGGERS
    value: "crawlee"

- After changing env vars, restart your worker so it picks up the new setting. Tip: - You can list multiple libraries:

PREFECT_LOGGING_EXTRA_LOGGERS=crawlee,requests,urllib3

If you tell me where your flows run (local, Docker, K8s, etc.), I can give you the exact snippet for that environment.

Awesome — glad to hear it worked! Quick tips for later: - If you run flows on remote workers/containers, make sure

PREFECT_LOGGING_EXTRA_LOGGERS

is set in that runtime environment (or via deployment job env). - You can capture multiple libraries:

PREFECT_LOGGING_EXTRA_LOGGERS=crawlee,requests,urllib3

- If the library prints instead of logging, use

log_prints=True

on the flow/task to capture stdout/stderr. If you want, tell me where you’re running (local, Docker, K8s) and I’ll share the exact snippet to persist it there.