Spaces:
Runtime error
Runtime error
# coding=utf-8 | |
# Copyright 2021-present, the Recognai S.L. team. | |
# | |
# Licensed under the Apache License, Version 2.0 (the "License"); | |
# you may not use this file except in compliance with the License. | |
# You may obtain a copy of the License at | |
# | |
# http://www.apache.org/licenses/LICENSE-2.0 | |
# | |
# Unless required by applicable law or agreed to in writing, software | |
# distributed under the License is distributed on an "AS IS" BASIS, | |
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | |
# See the License for the specific language governing permissions and | |
# limitations under the License. | |
""" | |
This module centralizes all configuration and logging management | |
""" | |
import logging | |
from logging import Logger | |
from typing import Type | |
try: | |
from loguru import logger | |
except ModuleNotFoundError: | |
logger = None | |
def full_qualified_class_name(_class: Type) -> str: | |
"""Calculates the full qualified name (module + class) of a class""" | |
class_module = _class.__module__ | |
if class_module is None or class_module == str.__class__.__module__: | |
return _class.__name__ # Avoid reporting __builtin__ | |
else: | |
return f"{class_module}.{_class.__name__}" | |
def get_logger_for_class(_class: Type) -> Logger: | |
"""Return the logger for a given class""" | |
return logging.getLogger(full_qualified_class_name(_class)) | |
class LoggingMixin: | |
""" | |
Main logging class methods. Classes that inherit from this, have | |
available a `logger` properly configured property | |
""" | |
__logger__: Logger = None | |
def __new__(cls, *args, **kwargs): | |
cls.__logger__ = get_logger_for_class(cls) | |
return super().__new__(cls) | |
def logger(self) -> logging.Logger: | |
"""Return the logger configured for the class""" | |
return self.__logger__ | |
class LoguruLoggerHandler(logging.Handler): | |
"""This logging handler enables an easy way to use loguru fo all built-in logger traces""" | |
__LOGLEVEL_MAPPING__ = { | |
50: "CRITICAL", | |
40: "ERROR", | |
30: "WARNING", | |
20: "INFO", | |
10: "DEBUG", | |
0: "NOTSET", | |
} | |
def is_available(self) -> bool: | |
"""Return True if handler can tackle log records. False otherwise""" | |
return logger is not None | |
def __init__(self, *args, **kwargs): | |
super().__init__(*args, **kwargs) | |
if not self.is_available: | |
self.emit = lambda record: None | |
def emit(self, record: logging.LogRecord): | |
try: | |
level = logger.level(record.levelname).name | |
except AttributeError: | |
level = self.__LOGLEVEL_MAPPING__[record.levelno] | |
frame, depth = logging.currentframe(), 2 | |
while frame.f_code.co_filename == logging.__file__: | |
frame = frame.f_back | |
depth += 1 | |
log = logger.bind(request_id="argilla") | |
log.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage()) | |
def configure_logging(): | |
"""Normalizes logging configuration for argilla and its dependencies""" | |
intercept_handler = LoguruLoggerHandler() | |
if not intercept_handler.is_available: | |
return | |
logging.basicConfig(handlers=[intercept_handler], level=logging.WARNING) | |
for name in logging.root.manager.loggerDict: | |
logger_ = logging.getLogger(name) | |
logger_.handlers = [] | |
for name in [ | |
"uvicorn", | |
"uvicorn.lifespan", | |
"uvicorn.error", | |
"uvicorn.access", | |
"fastapi", | |
"argilla", | |
"argilla.server", | |
]: | |
logger_ = logging.getLogger(name) | |
logger_.propagate = False | |
logger_.handlers = [intercept_handler] | |