Prefect 2.3 introduced a new context manager (PR#6575) to disable the logger (i.e., so the function won’t fail when running outside of a prefect context).
That solves half of the problem, because now, when testing outside of prefect you do not see the logs.
A better solution would be to use a logger that works both inside and outside of Prefect (if the logger is called inside: it acts like a Prefect logger and the logs are sent to Prefect Cloud, whereas if the logger is called outside: it acts like a regular logger and the logs can be seen in the terminal).
Here is a small proof-of-concept:
import inspect
import logging
import os
os.environ["PREFECT_LOGGING_EXTRA_LOGGERS"] = "my_hybrid_logger"
try:
from prefect import context, flow, task
except ModuleNotFoundError:
print("prefect is not installed")
class HybridLogger(logging.Logger):
def __init__(self, level="info"):
self._level = level.upper()
self.caller_filename = inspect.stack()[1].filename.split("/")[-1]
self.logger = self._get_regular_logger()
def _get_regular_logger(self) -> logging.Logger:
"""Create, configure and return a regular logger instance."""
logging.basicConfig()
regular_logger = logging.getLogger("my_hybrid_logger")
regular_logger.setLevel(logging.getLevelName(self._level))
return regular_logger
def _get_prefect_context(self):
"""Return the current Prefect context. Otherwise return an empty string."""
# Check for existing contexts
try:
task_run_context = context.TaskRunContext.get()
flow_run_context = context.FlowRunContext.get()
except NameError:
# prefect is not installed
return ""
# Determine if this is a task or flow run logger
if task_run_context:
return f"Task run '{task_run_context.task_run.name}' - "
elif flow_run_context:
return f"Flow run '{flow_run_context.flow_run.name}' - "
else:
return ""
def _fmt_msg(self, msg, caller_function_name):
"""Format the message adding some context."""
return f"{self._get_prefect_context()}{self.caller_filename}:{caller_function_name} - {msg}"
def debug(self, msg, *args, **kwargs):
msg = self._fmt_msg(msg, caller_function_name=inspect.stack()[1].function)
self.logger.debug(msg, *args, **kwargs)
def info(self, msg, *args, **kwargs):
msg = self._fmt_msg(msg, caller_function_name=inspect.stack()[1].function)
self.logger.info(msg, *args, **kwargs)
def warning(self, msg, *args, **kwargs):
msg = self._fmt_msg(msg, caller_function_name=inspect.stack()[1].function)
self.logger.warning(msg, *args, **kwargs)
def error(self, msg, *args, **kwargs):
msg = self._fmt_msg(msg, caller_function_name=inspect.stack()[1].function)
self.logger.error(msg, *args, **kwargs)
def exception(self, msg, *args, **kwargs):
msg = self._fmt_msg(msg, caller_function_name=inspect.stack()[1].function)
self.logger.exception(msg, *args, **kwargs)
def critical(self, msg, *args, **kwargs):
msg = self._fmt_msg(msg, caller_function_name=inspect.stack()[1].function)
self.logger.critical(msg, *args, **kwargs)
def demo_outside_prefect():
logger = HybridLogger(level="info")
logger.debug("not showing")
logger.warning("warning logging outside of a Prefect context")
logger = HybridLogger(level="debug")
logger.debug("debug logging outside of a Prefect context")
@flow
def demo_inside_prefect():
logger = HybridLogger(level="info")
logger.debug("not showing")
logger.warning("warning logging inside of a Prefect Flow")
logger = HybridLogger(level="debug")
logger.debug("debug logging inside of a Prefect Flow")
demo_inside_prefect_task()
@task
def demo_inside_prefect_task():
logger = HybridLogger(level="info")
logger.debug("not showing")
logger.warning("warning logging inside of a Prefect Task")
logger = HybridLogger(level="debug")
logger.debug("debug logging inside of a Prefect Task")
if __name__ == "__main__":
demo_outside_prefect()
demo_inside_prefect()
Here is why someone would want to use such a logger:
- Works both inside and outside of Prefect context (i.e. can be used for testing while still seeing the logs)
- You no longer have to write “logger = get_run_logger()” in every functions. Instead, you can simply instantiate the logger once in the top of the file (with “logger = HybridLogger()”)
- [extra] You can see the filename and the function_name that called the logger (very useful if the function that calls the logger is a regular function (i.e., not a flow or a task), because otherwise it’s hard to tell where the log is coming from)
Downsides:
- You have to implement it yourself
- You have to set PREFECT_LOGGING_EXTRA_LOGGERS
- Not tested & not supported by Prefect
Questions:
- Could Prefect make their logger hybrid (so it works like a regular logger when outside of a Prefect context)?
- Could Prefect make their logger importable like that : “from prefect import logger” (so you don’t have to write “logger = get_run_logger(level)” in every functions)?