Yes. In Prefect 3.x you can forward all Loguru logs into Prefect by bridging Loguru -> Python logging and attaching Prefect’s API log handler. Here’s a minimal drop-in snippet you can run at startup of your flow code:

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

# Create a stdlib logger that ships logs to Prefect
prefect_bridge = logging.getLogger("loguru-to-prefect")
prefect_bridge.setLevel(logging.DEBUG)
prefect_bridge.addHandler(APILogHandler())

# Forward every Loguru message into the stdlib logger above
def _loguru_to_prefect_sink(message):
    record = message.record  # dict-like with level/name/message, etc.
    prefect_bridge.log(record["level"].no, record["message"])

# Optional: keep existing Loguru outputs; remove() if you want only Prefect
# loguru_logger.remove()
loguru_logger.add(_loguru_to_prefect_sink, level="DEBUG", enqueue=True)

Notes: - Logs will appear in Prefect only when there’s an active flow/task run context (e.g., inside `@flow`/`@task` execution). - If you also want to capture

print()

output, set

@flow(log_prints=True)

/

@task(log_prints=True)

. - Alternative: if you already forward Loguru into a named stdlib logger, you can attach Prefect’s handler to that logger via code above. The

PREFECT_LOGGING_EXTRA_LOGGERS

env var helps Prefect attach handlers to standard loggers, but Loguru itself doesn’t use stdlib logging by default, so the bridge above is still needed. Docs for reference: - Add logging to flows and tasks - Advanced logging customization - APILogHandler API If you’re on Prefect 2.x instead, let me know and I’ll tailor the snippet accordingly.

Great question — what you’re seeing is the default Prefect formatter showing the stdlib logger’s name, not Loguru’s original call site. You can keep Loguru’s fidelity (modulefunctionline, original timestamp, etc.) by building a stdlib LogRecord with those fields from Loguru’s record and handing it directly to Prefect’s APILogHandler. Try this drop-in bridge; it preserves: - logger name (Loguru’s `name`/module) - function and line number - original timestamp - level - message text

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

# Logger that ships to Prefect's API
_prefect_bridge = logging.getLogger("prefect.loguru.bridge")
_prefect_bridge.setLevel(logging.DEBUG)
_prefect_bridge.addHandler(APILogHandler())
_prefect_bridge.propagate = False  # avoid duplicate console output

def _loguru_to_prefect_sink(message):
    r = message.record  # Loguru record dict
    level = r["level"].no

    # Recreate Loguru-style prefix in the message so Prefect UI shows the call site
    msg = f"{r['module']}:{r['function']}:{r['line']} - {r['message']}"

    # Try to preserve exception info if present
    exc_info = None
    if r["exception"]:
        try:
            exc = r["exception"]
            exc_info = (exc.type, exc.value, exc.traceback)
        except Exception:
            exc_info = None

    # Build a stdlib LogRecord with the original origin (file/func/line) and logger name
    record = _prefect_bridge.makeRecord(
        name=r["name"],            # usually module path (e.g. "sync.cli.prefect.experimental")
        level=level,
        fn=r["file"].path,         # full path to source file
        lno=r["line"],
        func=r["function"],
        msg=msg,
        args=(),
        exc_info=exc_info,
        extra={"module": r["module"], "file": r["file"].name},
    )
    # Preserve Loguru's timestamp
    record.created = r["time"].timestamp()

    _prefect_bridge.handle(record)

# Add the sink; keep existing Loguru outputs if you want
# loguru_logger.remove()  # uncomment if you only want Prefect logs
loguru_logger.add(_loguru_to_prefect_sink, level="DEBUG", enqueue=True, backtrace=False, diagnose=False)

This will give you Prefect logs like:

2025-10-17 07:30:35.157 | WARNING | sync.cli.prefect.experimental:_failure_hook:80 - failure hook triggered for task: run_joy

in the UI (because we include the modulefunctionline in the message string and set the logger name to Loguru’s

name

). Optional: also show file/line in local console output - If you want your local console to include function/line without modifying messages, customize Prefect’s console formatter via a logging config: - Put a YAML at ~/.prefect/logging.yml (or point to it with PREFECT_LOGGING_SETTINGS_PATH) with a format like:

version: 1
formatters:
  console:
    "()": "prefect.logging.formatters.PrefectFormatter"
    format: "%(asctime)s | %(levelname)s | %(name)s:%(funcName)s:%(lineno)d - %(message)s"
handlers:
  console:
    class: "prefect.logging.handlers.PrefectConsoleHandler"
    level: "INFO"
    formatter: "console"
loggers:
  prefect.flow_runs:
    level: "INFO"
    handlers: ["console"]
    propagate: false
  prefect.task_runs:
    level: "INFO"
    handlers: ["console"]
    propagate: false

Caveats: - Prefect UI primarily displays the message text; adding the call-site into the message (as above) ensures you always see it in the UI. - If you use

enqueue=True

(recommended), stacklevel tricks won’t work to infer the original frame; explicitly setting

fn/lno/func

as shown is the reliable way. - Logs only appear in Prefect when there’s an active flow/task run context. Refs: - Prefect logging customization - APILogHandler API If you want me to tweak the prefix to exactly match your Loguru format (including date formatting), share your Loguru formatter and I’ll mirror it.

Thanks for the traceback — the KeyError is because

logging.makeRecord(..., extra=...)

cannot overwrite reserved LogRecord attributes like

module

,

filename

, etc. That’s why passing

extra={"module": ...}

blows up. Here’s a corrected bridge that: - avoids reserved keys - preserves the original call site (modulefunctionline) - preserves the original timestamp - includes formatted traceback in the message when present - sets the logger name to Loguru’s

name

so the UI shows your module path instead of the bridge logger

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

# Logger that ships to Prefect's API
_bridge_logger = logging.getLogger("prefect.loguru.bridge")
_bridge_logger.setLevel(logging.DEBUG)
_bridge_logger.addHandler(APILogHandler())
_bridge_logger.propagate = False  # avoid duplicate console output

def _loguru_to_prefect_sink(message):
    r = message.record  # Loguru record dict
    level = r["level"].no

    # Build message with call-site info so it’s visible in the Prefect UI
    msg = f"{r['module']}:{r['function']}:{r['line']} - {r['message']}"

    # If there’s an exception, append a formatted traceback
    if r["exception"]:
        try:
            tb_str = "".join(r["exception"].traceback.format())
            msg = f"{msg}\n{tb_str}"
        except Exception:
            pass

    # Create a stdlib record with original origin and name
    record = _bridge_logger.makeRecord(
        name=r["name"],            # e.g. "sync.cli.prefect.experimental"
        level=level,
        fn=r["file"].path,         # absolute path to the source file
        lno=r["line"],
        func=r["function"],
        msg=msg,
        args=(),
        exc_info=None,             # already appended to msg (Loguru's traceback isn't a stdlib traceback)
    )
    # Preserve Loguru’s timestamp
    record.created = r["time"].timestamp()

    _bridge_logger.handle(record)

# Important: keep enqueue=False so logs are emitted in the same thread
# This preserves Prefect's run context (flow/task IDs); using enqueue=True can drop context
loguru_logger.add(_loguru_to_prefect_sink, level="DEBUG", enqueue=False, backtrace=False, diagnose=False)

Why this fixes both issues: - The KeyError is gone because we’re not trying to set reserved fields via

extra

. - Prefect UI will now show the logger as your module (from

name=r["name"]

) and the message will include

module:function:line

, so you’ll see the offending codeline inline (matching your Loguru format). - The original timestamp is preserved by setting

record.created

. Tip: - If you want the call-site shown by formatting instead of in the message body, you can also customize Prefect’s console formatter; however, the Prefect UI primarily displays the message string, so embedding the call-site in the message is the most reliable way to see it there. If you still prefer

enqueue=True

, let me know — we can discuss strategies to preserve Prefect run context across the background queue, but

enqueue=False

is the safest for accurate run attribution.

Short answer: - Console: Yes — Prefect 3.x uses Rich for console logs and can render Rich markup like [bold green]…[/bold green]. - Prefect UI: No — the web UI renders logs as plain text (no colors/ANSI). To see colors in your local/worker console, enable Rich markup and colors: - Set environment variables before starting your worker/flow:

export PREFECT_LOGGING_MARKUP=true
export PREFECT_LOGGING_COLORS=true

- Then log with markup:

from prefect import flow, get_run_logger

@flow
def demo():
    log = get_run_logger()
    <http://log.info|log.info>("[bold green]Starting...[/bold green]")

demo()

Your console output will render the green “Starting…” text. If you’re forwarding Loguru messages into Prefect (via the bridge we discussed), leaving the markup tags in the message will render as color in the console once markup is enabled. Notes and tips: - The Prefect UI will still show the literal “[bold green]Starting…[/bold green]” — that’s expected. - Markup can misinterpret strings with square brackets; if needed, escape per Rich’s rules. - You can further customize console styling/highlighting via a logging config file (~/.prefect/logging.yml) and Prefect’s console handler/highlighters. Docs: - Advanced logging customization - Settings reference (see PREFECT_LOGGING_MARKUP, PREFECT_LOGGING_COLORS) If you want, I can share a sample logging.yml that tweaks colors for levels and names too.

You’re still seeing the literal “[bold green]…” because that line is coming from Loguru’s own console sink, not from Prefect’s Rich-powered console handler. Prefect can render Rich markup, but only for logs that go through its console handler. Do this: 1) Enable markup/colors for Prefect’s console

export PREFECT_LOGGING_MARKUP=true
export PREFECT_LOGGING_COLORS=true

2) Stop Loguru from writing its own console output and let Prefect handle console + API - Remove Loguru’s default sink - Attach both Prefect’s API and console handlers to the bridge logger

import logging
from loguru import logger as loguru_logger
from prefect.logging.handlers import APILogHandler, PrefectConsoleHandler

_bridge = logging.getLogger("prefect.loguru.bridge")
_bridge.setLevel(logging.DEBUG)
_bridge.addHandler(APILogHandler())          # send to Prefect backend
_bridge.addHandler(PrefectConsoleHandler())  # render to console (Rich), honors PREFECT_LOGGING_MARKUP/COLORS
_bridge.propagate = False

def _loguru_to_prefect_sink(message):
    r = message.record
    level = r["level"].no
    msg = f"{r['module']}:{r['function']}:{r['line']} - {r['message']}"
    record = _bridge.makeRecord(
        name=r["name"],
        level=level,
        fn=r["file"].path,
        lno=r["line"],
        func=r["function"],
        msg=msg,
        args=(),
        exc_info=None,
    )
    record.created = r["time"].timestamp()
    _bridge.handle(record)

# Important: remove Loguru's default console sink so you don't see the unrendered markup
loguru_logger.remove()
loguru_logger.add(_loguru_to_prefect_sink, level="DEBUG", enqueue=False, backtrace=False, diagnose=False)

What to expect: - Your terminal/worker console will now render “[bold green]…[/bold green]” as colored text. - The Prefect UI still shows plain text (no colors/ANSI), by design. If it’s still not colored: - Double-check the env vars are set in the same process that runs the flow/worker. - Some environments disable color if there’s no TTY; PREFECT_LOGGING_COLORS=true helps, but some log aggregators will strip ANSI. - You can also force via a logging.yml if you prefer: Advanced logging customization - Settings reference: PREFECT_LOGGING_MARKUP, PREFECT_LOGGING_COLORS Where are you viewing the logs (local terminal, worker container stdout, or Prefect UI)? I can tailor this to your setup.

Yes—route Loguru into Prefect’s run logger so Prefect formats it exactly like its own logs. This sink uses Prefect’s

get_run_logger()

inside the Loguru sink. That means your console output will match Prefect’s default format (e.g., “082036.529 | INFO | Task run 'run_joy-materialize' - …”) and the logs will be attributed to the active run.

import logging
from contextlib import suppress
from loguru import logger as loguru_logger

# Optional: a stdlib fallback logger for messages emitted outside a flow/task run
_fallback = logging.getLogger("loguru")  # If you want Prefect to capture these too, set PREFECT_LOGGING_EXTRA_LOGGERS=loguru
_fallback.setLevel(logging.DEBUG)

def _loguru_to_prefect_sink(message):
    r = message.record
    level = r["level"].no
    msg = r["message"]

    # Append traceback text (get_run_logger doesn't accept Loguru's traceback object)
    if r["exception"]:
        try:
            msg = f"{msg}\n{''.join(r['exception'].traceback.format())}"
        except Exception:
            pass

    # Prefer Prefect’s run logger for exact formatting/prefix and API attribution
    with suppress(Exception):
        from prefect import get_run_logger
        logger = get_run_logger()
        logger.log(level, msg)  # Prefect formats this exactly like built-in logs
        return

    # Fallback (outside of a run): emit via stdlib; Prefect can still capture with EXTRA_LOGGERS
    record = _fallback.makeRecord(
        name=_fallback.name,
        level=level,
        fn=r["file"].path,
        lno=r["line"],
        func=r["function"],
        msg=msg,
        args=(),
        exc_info=None,
    )
    record.created = r["time"].timestamp()
    _fallback.handle(record)

# Replace Loguru's default console sink so only Prefect's formatting is shown
loguru_logger.remove()
loguru_logger.add(_loguru_to_prefect_sink, level="DEBUG", enqueue=False, backtrace=False, diagnose=False)

Also enable markup/colors in your console if you want Rich styling:

export PREFECT_LOGGING_MARKUP=true
export PREFECT_LOGGING_COLORS=true

Optional (for the fallback path): - To have non-run logs also flow through Prefect’s handlers, set:

export PREFECT_LOGGING_EXTRA_LOGGERS=loguru

Notes: - Using

get_run_logger()

ensures the exact same console line shape and run-context prefix as native Prefect logs. - The Prefect UI shows plain text (no colors), even if markup is enabled in the console. - If you emit Loguru levels like TRACE/SUCCESS, they’ll pass through as custom numeric levels; if you prefer mapping them to stdlib levels, I can add that mapping. Links: Add logging • Logging customization

Got it — that’s happening because Prefect’s default log level is INFO, so DEBUG is being filtered out when we route through

get_run_logger()

. Pick one of these to enable DEBUG: Option A: Set level via environment (simplest) - Set before starting your worker/flow process:

export PREFECT_LOGGING_LEVEL=DEBUG
# optional: keep Prefect’s own internal logs quiet
export PREFECT_INTERNAL_LOGGING_LEVEL=WARNING

This applies to both console and API logs. Make sure these env vars are present in the process that actually runs your flow (e.g., worker container/pod). Option B: Configure in a Prefect profile or project - In your profile (e.g., ~/.prefect/profiles.toml) or project config (prefect.toml/pyproject.toml):

[logging]
level = "DEBUG"

Option C: Use a logging.yml for fine-grained control - Create ~/.prefect/logging.yml (or point to it with PREFECT_LOGGING_SETTINGS_PATH) to force DEBUG for run loggers/handlers:

version: 1
formatters:
  console:
    "()": "prefect.logging.formatters.PrefectFormatter"
handlers:
  console:
    class: "prefect.logging.handlers.PrefectConsoleHandler"
    level: "DEBUG"
    formatter: "console"
  api:
    class: "prefect.logging.handlers.APILogHandler"
    level: "DEBUG"
loggers:
  prefect.flow_runs:
    level: "DEBUG"
    handlers: ["console", "api"]
    propagate: false
  prefect.task_runs:
    level: "DEBUG"
    handlers: ["console", "api"]
    propagate: false

Notes: - Your Loguru →

get_run_logger()

sink is fine; once Prefect’s level is DEBUG you’ll see those messages. - If you also want colored/markup output in the console, keep:

export PREFECT_LOGGING_MARKUP=true
export PREFECT_LOGGING_COLORS=true

Docs: - Advanced logging customization - Add logging to flows and tasks - Settings and profiles If you tell me how you’re running (local, Docker, K8s worker), I can show exactly where to set the env vars.

Totally doable. You can keep Prefect’s own logs at INFO while allowing your Loguru messages to come through at DEBUG. Two approaches: Option 1 — Bump only the run logger level programmatically (per run) Keep your Loguru→get_run_logger bridge, but raise the level of the active run logger to DEBUG just before you emit. This does not change Prefect’s internal loggers.

import logging
from contextlib import suppress
from loguru import logger as loguru_logger

def _loguru_to_prefect_sink(message):
    r = message.record
    msg = r["message"]

    # Map Loguru TRACE to DEBUG so it's not dropped
    level = r["level"].no
    if r["level"].name == "TRACE":
        level = logging.DEBUG

    # Append traceback text if present
    if r["exception"]:
        with suppress(Exception):
            msg = f"{msg}\n{''.join(r['exception'].traceback.format())}"

    with suppress(Exception):
        from prefect import get_run_logger
        plog = get_run_logger()

        # Raise only the run logger’s level to DEBUG for this run
        base = getattr(plog, "logger", None)  # underlying stdlib logger
        if base and base.level > logging.DEBUG:
            base.setLevel(logging.DEBUG)

        plog.log(level, msg)
        return

    # If we're outside a run, you can no-op or send to stdlib fallback
    # (omitted for brevity)

loguru_logger.remove()
loguru_logger.add(_loguru_to_prefect_sink, level="DEBUG", enqueue=False, backtrace=False, diagnose=False)

- This keeps Prefect’s internal loggers (e.g., worker/engine) at their default levels, but ensures your run-scoped logs aren’t filtered. - We map Loguru TRACE→DEBUG so it’s not dropped by stdlib’s level thresholds. Option 2 — Configure only run loggers to DEBUG via logging.yml If you prefer config over code, set just the run loggers to DEBUG and leave Prefect internals alone:

version: 1
formatters:
  console:
    "()": "prefect.logging.formatters.PrefectFormatter"
handlers:
  console:
    class: "prefect.logging.handlers.PrefectConsoleHandler"
    level: "DEBUG"
    formatter: "console"
  api:
    class: "prefect.logging.handlers.APILogHandler"
    level: "DEBUG"
loggers:
  prefect.flow_runs:
    level: "DEBUG"
    handlers: ["console", "api"]
    propagate: false
  prefect.task_runs:
    level: "DEBUG"
    handlers: ["console", "api"]
    propagate: false
  # Keep Prefect internals at INFO (or higher)
  prefect:
    level: "INFO"
    propagate: true

- Save as ~/.prefect/logging.yml (or point PREFECT_LOGGING_SETTINGS_PATH to it). - This shows all DEBUG emitted via get_run_logger() (including your Loguru-bridged messages) but not Prefect’s internal DEBUG. Tip for colors/markup in console: - If you want [bold green]…[/bold green] to render in your terminal, set:

export PREFECT_LOGGING_MARKUP=true
export PREFECT_LOGGING_COLORS=true

The Prefect UI will still show plain text. Docs: - Advanced logging customization - Add logging to flows and tasks If you tell me where you run (local, Docker, K8s), I can show exactly where to place the env/config.