FileMgr

✏️ /usr/local/lib/python3.9/site-packages/kombu/log.py

✕ Tutup

"""Logging Utilities."""

from __future__ import annotations

import logging
import numbers
import os
import sys
from logging.handlers import WatchedFileHandler
from typing import TYPE_CHECKING

from .utils.encoding import safe_repr, safe_str
from .utils.functional import maybe_evaluate
from .utils.objects import cached_property

if TYPE_CHECKING:
    from logging import Logger

__all__ = ('LogMixin', 'LOG_LEVELS', 'get_loglevel', 'setup_logging')

LOG_LEVELS = dict(logging._nameToLevel)
LOG_LEVELS.update(logging._levelToName)
LOG_LEVELS.setdefault('FATAL', logging.FATAL)
LOG_LEVELS.setdefault(logging.FATAL, 'FATAL')
DISABLE_TRACEBACKS = os.environ.get('DISABLE_TRACEBACKS')

def get_logger(logger: str | Logger):
    """Get logger by name."""
    if isinstance(logger, str):
        logger = logging.getLogger(logger)
    if not logger.handlers:
        logger.addHandler(logging.NullHandler())
    return logger

def get_loglevel(level):
    """Get loglevel by name."""
    if isinstance(level, str):
        return LOG_LEVELS[level]
    return level

def naive_format_parts(fmt):
    parts = fmt.split('%')
    for i, e in enumerate(parts[1:]):
        yield None if not e or not parts[i - 1] else e[0]

def safeify_format(fmt, args, filters=None):
    filters = {'s': safe_str, 'r': safe_repr} if not filters else filters
    for index, type in enumerate(naive_format_parts(fmt)):
        filt = filters.get(type)
        yield filt(args[index]) if filt else args[index]

class LogMixin:
    """Mixin that adds severity methods to any class."""

def debug(self, *args, **kwargs):
        return self.log(logging.DEBUG, *args, **kwargs)

def info(self, *args, **kwargs):
        return self.log(logging.INFO, *args, **kwargs)

def warn(self, *args, **kwargs):
        return self.log(logging.WARN, *args, **kwargs)

def error(self, *args, **kwargs):
        kwargs.setdefault('exc_info', True)
        return self.log(logging.ERROR, *args, **kwargs)

def critical(self, *args, **kwargs):
        kwargs.setdefault('exc_info', True)
        return self.log(logging.CRITICAL, *args, **kwargs)

def annotate(self, text):
        return f'{self.logger_name} - {text}'

def log(self, severity, *args, **kwargs):
        if DISABLE_TRACEBACKS:
            kwargs.pop('exc_info', None)
        if self.logger.isEnabledFor(severity):
            log = self.logger.log
            if len(args) > 1 and isinstance(args[0], str):
                expand = [maybe_evaluate(arg) for arg in args[1:]]
                return log(severity,
                           self.annotate(args[0].replace('%r', '%s')),
                           *list(safeify_format(args[0], expand)), **kwargs)
            else:
                return self.logger.log(
                    severity, self.annotate(' '.join(map(safe_str, args))),
                    **kwargs)

def get_logger(self):
        return get_logger(self.logger_name)

def is_enabled_for(self, level):
        return self.logger.isEnabledFor(self.get_loglevel(level))

def get_loglevel(self, level):
        if not isinstance(level, numbers.Integral):
            return LOG_LEVELS[level]
        return level

@cached_property
    def logger(self):
        return self.get_logger()

@property
    def logger_name(self):
        return self.__class__.__name__

class Log(LogMixin):

def __init__(self, name, logger=None):
        self._logger_name = name
        self._logger = logger

def get_logger(self):
        if self._logger:
            return self._logger
        return super().get_logger()

@property
    def logger_name(self):
        return self._logger_name

def setup_logging(loglevel=None, logfile=None):
    """Setup logging."""
    logger = logging.getLogger()
    loglevel = get_loglevel(loglevel or 'ERROR')
    logfile = logfile if logfile else sys.__stderr__
    if not logger.handlers:
        if hasattr(logfile, 'write'):
            handler = logging.StreamHandler(logfile)
        else:
            handler = WatchedFileHandler(logfile)
        logger.addHandler(handler)
        logger.setLevel(loglevel)
    return logger

Batal

⬅ site-packages 23 item

Nama	Tipe	Ukuran	Diubah	Perm	Aksi
📁 __pycache__	dir	—	08/05/25 20:25	0755	📂
📁 asynchronous	dir	—	08/05/25 20:25	0755	📂
📁 transport	dir	—	08/05/25 20:25	0755	📂
📁 utils	dir	—	08/05/25 20:25	0755	📂
🐍 __init__.py	py	3.8 KB	08/05/25 20:25	0644	✏️ ⬇
🐍 abstract.py	py	4.3 KB	08/05/25 20:25	0644	✏️ ⬇
🐍 clocks.py	py	4.7 KB	08/05/25 20:25	0644	✏️ ⬇
🐍 common.py	py	13.4 KB	08/05/25 20:25	0644	✏️ ⬇
🐍 compat.py	py	6.6 KB	08/05/25 20:25	0644	✏️ ⬇
🐍 compression.py	py	2.9 KB	08/05/25 20:25	0644	✏️ ⬇
🐍 connection.py	py	40.5 KB	08/05/25 20:25	0644	✏️ ⬇
🐍 entity.py	py	32.4 KB	08/05/25 20:25	0644	✏️ ⬇
🐍 exceptions.py	py	2.8 KB	08/05/25 20:25	0644	✏️ ⬇
🐍 log.py	py	4.1 KB	08/05/25 20:25	0644	✏️ ⬇
🐍 matcher.py	py	4.2 KB	08/05/25 20:25	0644	✏️ ⬇
🐍 message.py	py	8 KB	08/05/25 20:25	0644	✏️ ⬇
🐍 messaging.py	py	24.5 KB	08/05/25 20:25	0644	✏️ ⬇
🐍 mixins.py	py	9.5 KB	08/05/25 20:25	0644	✏️ ⬇
🐍 pidbox.py	py	14.4 KB	08/05/25 20:25	0644	✏️ ⬇
🐍 pools.py	py	3.9 KB	08/05/25 20:25	0644	✏️ ⬇
🐍 resource.py	py	7.9 KB	08/05/25 20:25	0644	✏️ ⬇
🐍 serialization.py	py	15.1 KB	08/05/25 20:25	0644	✏️ ⬇
🐍 simple.py	py	5.2 KB	08/05/25 20:25	0644	✏️ ⬇

✏️ Rename

🔑 Ubah Permission