"""Logger Handler integrated with Carol."""
import logging
import os
import sys
from .tasks import Tasks
from .carol import Carol
_carol_levels = dict(
NOTSET="NOTSET",
DEBUG="DEBUG",
INFO="INFO",
WARNING="WARN",
WARN="WARN",
ERROR="ERROR",
CRITICAL="ERROR",
FATAL="ERROR",
)
[docs]class CarolHandler(logging.StreamHandler):
"""Carol logger handler.
This class can be used to log information in long tasks in Carol.
Args:
carol: Carol object.
task_id: Task id.
Usage:
.. code:: python
from pycarol import Carol, CarolHandler
import logging
logger = logging.getLogger(__name__)
carol = CarolHandler(Carol())
carol.setLevel(logging.INFO)
logger.addHandler(carol)
# This will not be logged in Carol. Level is set to INFO
logger.debug('This is a debug message')
logger.info('This is an info message')
logger.warning('This is a warning message')
logger.error('This is an error message')
logger.critical('This is a critical message')
# These methods will use the current long task id provided by Carol when running
# your application.
# For local environments you need to set that manually first on the beginning of
# your code:
import os
os.environ['LONGTASKID'] = TASK_ID
If no TASK ID is passed it works as a Console Handler.
"""
def __init__(self, carol: Carol = None, task_id: str = None):
super().__init__(stream=sys.stdout)
self._use_console = False
if carol is None:
domain = os.getenv("CAROLTENANT")
app_name = os.getenv("CAROLAPPNAME")
auth_token = os.getenv("CAROLAPPOAUTH")
connector_id = os.getenv("CAROLCONNECTORID")
if (
(domain is not None)
and (app_name is not None)
and (auth_token is not None)
and (connector_id is not None)
):
carol = Carol()
self._use_console = False
else:
self._use_console = True
self.carol = carol
self._task = Tasks(self.carol)
self.task_id = task_id or os.getenv("LONGTASKID", None)
self._task.task_id = self.task_id
self._first_pending = True
def _log_carol(self, record):
msg = self.format(record)
log_level = _carol_levels.get(record.levelname)
self._task.add_log(msg, log_level=log_level)
[docs] def emit(self, record) -> None:
"""Log the message.
Args:
record: `str`
Message to log.
"""
if (self.task_id is None) or (self._use_console):
super().emit(record)
else:
self._log_carol(record)