Add on_missing_context parameter to get_run_logger()

src/prefect/logging/loggers.py

@@ -7,7 +7,16 @@
 from contextlib import contextmanager
 from functools import lru_cache
 from logging import LogRecord
-from typing import TYPE_CHECKING, Any, List, Mapping, MutableMapping, Optional, Union
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    List,
+    Literal,
+    Mapping,
+    MutableMapping,
+    Optional,
+    Union,
+)
 
 from typing_extensions import Self
 
@@ -89,25 +98,34 @@ def get_logger(name: str | None = None) -> logging.Logger:
 
 
 def get_run_logger(
-    context: Optional["RunContext"] = None, **kwargs: Any
+    context: Optional["RunContext"] = None,
+    on_missing_context: Literal["raise", "warn", "ignore"] = "raise",
+    **kwargs: Any,
 ) -> Union[logging.Logger, LoggingAdapter]:
     """
     Get a Prefect logger for the current task run or flow run.
 
-    The logger will be named either `prefect.task_runs` or `prefect.flow_runs`.
+    The logger will be named either ``prefect.task_runs`` or ``prefect.flow_runs``.
     Contextual data about the run will be attached to the log records.
 
-    These loggers are connected to the `APILogHandler` by default to send log records to
+    These loggers are connected to the ``APILogHandler`` by default to send log records to
     the API.
 
     Arguments:
         context: A specific context may be provided as an override. By default, the
             context is inferred from global state and this should not be needed.
+        on_missing_context: Controls behavior when no active flow or task run context
+            exists. ``"raise"`` (default) raises ``MissingContextError`` for backward
+            compatibility. ``"warn"`` returns a standard ``prefect`` logger and emits a
+            warning. ``"ignore"`` returns a standard ``prefect`` logger silently. Use
+            ``"warn"`` or ``"ignore"`` when calling from threads or other contexts where
+            a Prefect run context may not be available.
         **kwargs: Additional keyword arguments will be attached to the log records in
             addition to the run metadata
 
     Raises:
-        MissingContextError: If no context can be found
+        MissingContextError: If no context can be found and ``on_missing_context`` is
+            ``"raise"``
     """
     from prefect.context import FlowRunContext, TaskRunContext
 
@@ -148,6 +166,15 @@ def get_run_logger(
     ):
         logger = logging.getLogger("null")
         logger.disabled = True
+    elif on_missing_context == "warn":
+        logger = get_logger("prefect.run_fallback")
+        logger.debug(
+            "No active flow or task run context found. "
+            "Using fallback logger. This is expected when logging from "
+            "threads or other contexts without an active Prefect run."
+        )
+    elif on_missing_context == "ignore":
+        logger = get_logger("prefect.run_fallback")
     else:
         raise MissingContextError("There is no active flow or task run context.")
 

tests/logging/test_on_missing_context.py

@@ -0,0 +1,63 @@
+"""Tests for the on_missing_context parameter of get_run_logger."""
+
+import logging
+
+import pytest
+
+from prefect.exceptions import MissingContextError
+from prefect.logging.loggers import get_run_logger
+
+
+class TestOnMissingContext:
+    """Tests for the on_missing_context parameter."""
+
+    def test_raise_is_default_behavior(self):
+        """Default behavior raises MissingContextError outside a run context."""
+        with pytest.raises(
+            MissingContextError, match="no active flow or task run context"
+        ):
+            get_run_logger()
+
+    def test_raise_explicit(self):
+        """Explicitly passing on_missing_context='raise' raises MissingContextError."""
+        with pytest.raises(
+            MissingContextError, match="no active flow or task run context"
+        ):
+            get_run_logger(on_missing_context="raise")
+
+    def test_warn_returns_logger(self):
+        """on_missing_context='warn' returns a fallback logger instead of raising."""
+        logger = get_run_logger(on_missing_context="warn")
+        assert isinstance(logger, logging.Logger)
+        assert logger.name == "prefect.run_fallback"
+
+    def test_warn_emits_debug_message(self, caplog):
+        """on_missing_context='warn' emits a debug message about the fallback."""
+        with caplog.at_level(logging.DEBUG, logger="prefect.run_fallback"):
+            get_run_logger(on_missing_context="warn")
+        assert any(
+            "No active flow or task run context found" in record.message
+            for record in caplog.records
+        )
+
+    def test_ignore_returns_logger(self):
+        """on_missing_context='ignore' returns a fallback logger silently."""
+        logger = get_run_logger(on_missing_context="ignore")
+        assert isinstance(logger, logging.Logger)
+        assert logger.name == "prefect.run_fallback"
+
+    def test_ignore_does_not_emit_message(self, caplog):
+        """on_missing_context='ignore' does not emit any log messages."""
+        with caplog.at_level(logging.DEBUG, logger="prefect.run_fallback"):
+            get_run_logger(on_missing_context="ignore")
+        assert not any(
+            "No active flow or task run context found" in record.message
+            for record in caplog.records
+        )
+
+    def test_fallback_logger_is_functional(self):
+        """The fallback logger can actually log messages."""
+        logger = get_run_logger(on_missing_context="ignore")
+        # Should not raise
+        logger.info("test message from fallback logger")
+        logger.warning("test warning from fallback logger")