A few things to check to make sure your task’s warning/prints show up in the Prefect UI (assuming Prefect 3.x): 1) Make sure the logger is created inside the task body and you’re using Prefect’s run logger - You’re doing this correctly with

logger = get_run_logger()

inside

get_container

. Avoid creating the logger at import time. 2) Confirm you’re looking in the right place in the UI - Task logs appear both in the Flow Run log stream and in the specific Task Run details. If you don’t see the message in the flow log stream, click into the task run to check its logs. 3) If you rely on print(), ensure print-capture is enabled - You set

@task(log_prints=True)

, which is good. Alternatively, set it on the flow so all tasks inherit it:

@flow(log_prints=True)
def my_flow():
    container = get_container("MY_VAR")

- Or set it globally:

PREFECT_LOGGING_LOG_PRINTS=true

4) Ensure logs are being sent to the API - If API logging is disabled, you’ll see logs in your console but not in the UI. Check:

prefect config view --show-defaults

Look for: -

PREFECT_LOGGING_TO_API_ENABLED = true

(should be true) -

PREFECT_LOGGING_LOG_PRINTS

(true if you’re relying on prints) - Overall log level (e.g.,

PREFECT_LOGGING_LEVEL = INFO

or

DEBUG

) 5) Only Prefect’s run logger is forwarded with task/flow context - Using

get_run_logger().warning(...)

is correct. If any helper function uses the standard Python logger (e.g.,

logging.getLogger(__name__)

) those logs won’t carry Prefect context and may not reach the API unless you customize logging. Stick to

get_run_logger()

(or prints with

log_prints=True

) inside tasks/flows. 6) Don’t call tasks outside a flow - Calling a task outside a

@flow

won’t have a TaskRunContext, and

get_run_logger()

can’t route logs to the API. Minimal example that should show both a warning and a print in the UI:

# my_module.py
from prefect import task
from prefect.logging import get_run_logger

@task(log_prints=True)
def get_container(prod_image: str) -> str:
    logger = get_run_logger()

    prod_img = "foo"
    flow_container = "bar"

    if flow_container:
        if flow_container != prod_img:
            logger.warning("** Using a DEBUG_IMG container **")
            print("WARNING - ** Using a DEBUG_IMG container **")
        return flow_container

    if not prod_img:
        raise RuntimeError("Could not find an app container to work with.")

    return prod_img

# flow.py
from prefect import flow
from my_module import get_container

@flow(log_prints=True)
def my_flow():
    container = get_container("MY_VAR")
    return container

if __name__ == "__main__":
    my_flow()

If this still doesn’t show the warning in the UI: - Please confirm whether other logs from the flow appear in the UI. - Share the output of

prefect config view --show-defaults

for the logging-related settings. - Let me know if you’re running in Prefect Cloud or Prefect Server, and how you’re starting the run (e.g., python script locally, worker + deployment, etc.). Helpful docs: - Add logging to flows and tasks - Advanced logging customization

Thanks for the details — the message “logging records are not forwarded to the API” indicates Prefect’s APILogHandler received a log without an active run context (flow/task IDs). Since flow logs do reach the API but task logs don’t, the most likely causes are: - The task is being executed as a plain function (context not established), often due to using

.fn

somewhere or calling it at import time. - The runtime logging config does not attach the API handler to

prefect.task_runs

(a common customization pitfall). - Something inside your helpers emits Prefect logs before a TaskRunContext exists (e.g., module-level logging during import or background threads). Let’s quickly isolate which one it is. 1) Verify the task is actually running as a Prefect task (not bypassed) - Ensure there is no

.fn

usage or top-level calls to the task:

# BAD: runs at import time, outside any run context
container = get_container("MY_VAR")

# GOOD: only call inside a @flow
@flow
def my_flow():
    container = get_container("MY_VAR")

- Add a probe task to confirm TaskRunContext is present:

from prefect import task, flow
from prefect.context import TaskRunContext, FlowRunContext
from prefect.logging import get_run_logger

@task(log_prints=True)
def probe_task_logger():
    logger = get_run_logger()
    print("probe-print")
    trc = TaskRunContext.get()
    frc = FlowRunContext.get()
    <http://logger.info|logger.info>(f"probe-info task_run_id={getattr(trc, 'task_run', None) and trc.task_run.id}")
    logger.warning("probe-warning")

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

If the probe logs don’t appear in the UI, we’re dealing with environment/logging config or context propagation. 2) Check the effective logging config in the runtime environment The config you pasted is from your local shell; deployments run in the worker’s environment. Check settings inside the same environment that executes your flow/task: Option A: Inspect from within a task (shows what the runtime is using)

@task
def inspect_runtime_logging():
    import logging
    from prefect.logging import get_run_logger
    # get_run_logger returns a LoggingAdapter; its .logger holds the underlying logger
    l = get_run_logger()
    underlying = getattr(l, "logger", l)
    print(f"task-logger-name={underlying.name}")
    print(f"task-logger-handlers={[type(h).__name__ for h in underlying.handlers]}")

Expected: the task logger should be

prefect.task_runs

and include

APILogHandler

. If APILogHandler is missing here, your logging.yml in the runtime is overriding Prefect’s defaults. Option B: Check the logging.yml the runtime is loading - In the same runtime, print the path and (if possible) cat it:

@task
def show_logging_settings_path():
    import os
    from prefect.settings import PREFECT_LOGGING_SETTINGS_PATH
    print(f"PREFECT_LOGGING_SETTINGS_PATH={PREFECT_LOGGING_SETTINGS_PATH.value()}")

Open that file and confirm that: -

loggers.prefect.task_runs.handlers

includes

api

-

handlers.api.class

is Prefect’s APILogHandler If your runtime has a custom

.prefect/logging.yml

without the

api

handler on

prefect.task_runs

, task logs won’t reach the API. Remove/replace the custom file or restore the

api

handler there. 3) Rule out logs emitted outside a run context If your helpers (e.g.,

get_local_module_var

,

get_debug_container

) do any of the following, APILogHandler will warn and drop the record: - Call

get_run_logger()

at module import time - Use Prefect’s run logger from a background thread or process - Use Prefect loggers in code that runs before the task context is entered Move any Prefect logging into the body of a

@task

or

@flow

and avoid background threads for logging, or pass the context explicitly to background workers. 4) Enable print capture at the flow level (good practice) Even though your task has

log_prints=True

, enabling it at the flow removes doubt about inheritance and print capture: ```

@flow(log_prints=True) def my_flow(): container = get_container("MY_VAR")

Or set it via env in your deployment/worker: `PREFECT_LOGGING_LOG_PRINTS=true`.

If you want, here’s a minimal flow you can deploy to validate the environment without any of your module code:

from prefect import flow, task from prefect.logging import get_run_logger @task(log_prints=True) def t(): logger = get_run_logger() print("hello-from-task-print") logger.warning("hello-from-task-warning") @flow(log_prints=True) def f(): logger = get_run_logger() logger.info("hello-from-flow-info") t() if name == "__main__": f() ``` - If the flow info shows in UI but the task print/warning do not, it’s almost certainly the runtime logging config for

prefect.task_runs

(missing

APILogHandler

) or the task is not being executed as a task in that environment. Next steps - Can you run the probe/inspect snippets above in the same deployment/worker environment and share: - The handlers listed for the task logger - The value of

PREFECT_LOGGING_SETTINGS_PATH

and, if customized, the

prefect.task_runs

logger section from that file - Also confirm there is no

.fn

usage or module-level invocation of

get_container

. References - Add logging to flows and tasks: Add logging to flows and tasks - Logging customization: Advanced logging customization

Got it — since the single-file test works but your modular import does not, let’s surgically verify that: - the imported object is actually a Prefect Task (not the plain function), - the task runs with a TaskRunContext, - the task logger has the APILogHandler attached in the worker/deployment environment, - the module you think is running is actually the one that’s imported at runtime. Here’s a compact debugging recipe tailored for the modular setup. Step 1 — Prove you’re importing a Prefect Task (not a plain function) Add this to your flow (temporarily) right after the import:

# flow.py
from prefect import flow
from my_module import get_container

@flow(log_prints=True)
def my_flow():
    print(f"type(get_container)={type(get_container)}")
    print(f"hasattr(get_container, 'fn')={hasattr(get_container, 'fn')}")
    # Force a task run to ensure it’s executed as a task
    get_container.submit("MY_VAR")

Expected: -

hasattr(get_container, 'fn')

should be True. If False, you imported the plain function, not the Prefect Task (root cause). Step 2 — Introspect context and handlers inside the task Temporarily add this debug task and call it from your flow:

# my_module.py (or add a new file if you prefer not to touch get_container)
from prefect import task
from prefect.context import TaskRunContext, FlowRunContext
from prefect.logging import get_run_logger

@task(log_prints=True)
def inspect_task_runtime():
    trc = TaskRunContext.get()
    frc = FlowRunContext.get()
    print(f"TRC present: {bool(trc)}; FRC present: {bool(frc)}")

    try:
        logger = get_run_logger()
        underlying = getattr(logger, "logger", logger)  # unwrap adapter
        print(f"logger_name={underlying.name}")
        print(f"handlers={[type(h).__name__ for h in underlying.handlers]}")
        logger.warning("probe-warning-from-inspect")
    except Exception as e:
        print(f"get_run_logger failed: {e!r}")

# flow.py
from prefect import flow
from my_module import inspect_task_runtime

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

Expected: - TRC present: True - logger_name: prefect.task_runs - handlers include APILogHandler If TRC is False or handlers don’t include APILogHandler, that’s why task logs aren’t reaching the API. Step 3 — Confirm the module actually used at runtime It’s common for deployments to load code from a different path or installed package. Log the file path:

# my_module.py
import sys
import my_module  # self-import

@task(log_prints=True)
def show_module_origin():
    print(f"my_module.__file__={getattr(my_module, '__file__', 'unknown')}")
    print(f"sys.path={sys.path[:5]}")  # first few entries for brevity

# flow.py
from prefect import flow
from my_module import show_module_origin

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

Verify

my_module.__file__

points to the code version you expect in the worker/deployment environment. If it points somewhere else (or is missing), that explains the mismatch. Step 4 — Check the logging config used by the worker In the runtime, print what logging.yml is used and ensure the

api

handler is attached to `prefect.task_runs`:

# flow.py
from prefect import flow, task
from prefect.settings import PREFECT_LOGGING_SETTINGS_PATH

@task(log_prints=True)
def show_logging_settings_path():
    print(f"PREFECT_LOGGING_SETTINGS_PATH={PREFECT_LOGGING_SETTINGS_PATH.value()}")

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

Open that file in your runtime environment and confirm: - loggers.prefect.task_runs.handlers includes

api

- handlers.api class is Prefect’s APILogHandler If a custom logging.yml removed

api

from

prefect.task_runs

, task logs won’t reach the API. Step 5 — Common pitfalls to rule out in modular code - No

.fn

usage on tasks:

get_container.fn("MY_VAR")

bypasses the task engine/context; use

get_container(...)

or

get_container.submit(...)

.

- No module-level logging or

get_run_logger()

at import time in

my_module.py

- No background threads/processes emitting logs from inside the task (context does not propagate to other threads) - Ensure you’re not importing the undecorated function (e.g., shadowing names, conditionally applying @task, or importing from a different module path/package) Optional environment aids (in the deployment/worker env) - Set flow-level print capture to be safe:

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

- Temporarily increase verbosity to see more in the UI:

PREFECT_LOGGING_LEVEL=DEBUG

Fast sanity test - Force task execution path:

@flow(log_prints=True)
def my_flow():
    get_container.submit("MY_VAR")  # submit guarantees a TaskRun

If logs now appear, your earlier call path was bypassing the task engine (likely via

.fn

or an undecorated import). If you share: - Output of Step 1 (type/hasattr) - Output of Step 2 (TRC/handlers) - Output of Step 3 (module file path) - The

prefect.task_runs

section from your runtime logging.yml …I can pinpoint the exact root cause and provide the minimal fix. Helpful docs: - Add logging to flows and tasks - Advanced logging customization