Python:logging

import logging
loggers = [name for name in logging.root.manager.loggerDict]

# -*- coding: utf-8 -*-

import sys
import logging
from typing import Union

RECC_LOGGER = "recc"
RECC_HTTP_LOGGER = "recc.http"
RECC_RPC_LOGGER = "recc.rpc"

DEFAULT_LOGGING_FORMAT = "{levelname[0]} {asctime} [{name}] {message}"
DEFAULT_LOGGING_STYLE = "{"


def create_default_logger(
    name: str,
    level: Union[int, str] = logging.INFO,
) -> logging.Logger:

    logger = logging.getLogger(name)
    console = logging.StreamHandler(sys.stdout)
    console.setFormatter(
        logging.Formatter(fmt=DEFAULT_LOGGING_FORMAT, style=DEFAULT_LOGGING_STYLE)
    )
    logger.handlers.clear()
    logger.addHandler(console)
    logger.setLevel(level)

    return logger


recc_logger = create_default_logger(RECC_LOGGER)
recc_http_logger = logging.getLogger(RECC_HTTP_LOGGER)
recc_rpc_logger = logging.getLogger(RECC_RPC_LOGGER)

import logging

# create logger
logger = logging.getLogger('simple_example')
logger.setLevel(logging.DEBUG)

# create console handler and set level to debug
ch = logging.StreamHandler()
ch.setLevel(logging.DEBUG)

# create formatter
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')

# add formatter to ch
ch.setFormatter(formatter)

# add ch to logger
logger.addHandler(ch)

# 'application' code
logger.debug('debug message')
logger.info('info message')
logger.warning('warn message')
logger.error('error message')
logger.critical('critical message')

DEFAULT_LOGGING_FORMAT = "{levelname[0]} {asctime} [{name}] {message}"
DEFAULT_LOGGING_STYLE = "{"

logger = logging.getLogger("library")
console = logging.StreamHandler(sys.stdout)
console.setFormatter(
    logging.Formatter(fmt=DEFAULT_LOGGING_FORMAT, style=DEFAULT_LOGGING_STYLE)
)
logger.handlers.clear()
logger.addHandler(console)

import logging.handlers

file_handler = logging.handlers.TimedRotatingFileHandler(
  filename='test', when='midnight', interval=1,  encoding='utf-8'
  )
file_handler.suffix = 'log-%Y%m%d' # 파일명 끝에 붙여줌; ex. log-20190811

myLogger.addHandler(file_handler)

import logging.config, os, yaml


loggingConfigPath = 'logging.yaml'
if os.path.exists(loggingConfigPath):
    with open(loggingConfigPath, 'rt') as f:
        loggingConfig = yaml.load(f.read())
        logging.config.dictConfig(loggingConfig)
else:
    logging.basicConfig(level=logging.INFO)

logging.getLogger(__name__)

logging.debug("debug")
logging.info("info")

# -*- coding: utf-8 -*-

from functools import wraps
from inspect import iscoroutinefunction
from logging import (
    CRITICAL,
    DEBUG,
    ERROR,
    FATAL,
    INFO,
    NOTSET,
    WARN,
    WARNING,
    Logger,
    getLogger,
)
from time import time
from typing import Callable, Final, Optional, Union

LoggingElapsedCallable = Callable[[str, float], None]

SEVERITY_NAME_CRITICAL = "critical"
SEVERITY_NAME_FATAL = "fatal"
SEVERITY_NAME_ERROR = "error"
SEVERITY_NAME_WARNING = "warning"
SEVERITY_NAME_WARN = "warn"
SEVERITY_NAME_INFO = "info"
SEVERITY_NAME_DEBUG = "debug"
SEVERITY_NAME_NOTSET = "notset"
SEVERITY_NAME_OFF = "off"

DEFAULT_ROUND_PRECISION: Final[int] = 3


def convert_level_number(level: Optional[Union[str, int]] = None) -> int:
    if level is None:
        return DEBUG

    if isinstance(level, str):
        ll = level.lower()
        if ll == SEVERITY_NAME_CRITICAL:
            return CRITICAL
        elif ll == SEVERITY_NAME_FATAL:
            return FATAL
        elif ll == SEVERITY_NAME_ERROR:
            return ERROR
        elif ll == SEVERITY_NAME_WARNING:
            return WARNING
        elif ll == SEVERITY_NAME_WARN:
            return WARN
        elif ll == SEVERITY_NAME_INFO:
            return INFO
        elif ll == SEVERITY_NAME_DEBUG:
            return DEBUG
        elif ll == SEVERITY_NAME_NOTSET:
            return NOTSET
        elif ll == SEVERITY_NAME_OFF:
            return CRITICAL + 100
        else:
            try:
                return int(ll)
            except ValueError:
                raise ValueError(f"Unknown level: {level}")
    elif isinstance(level, int):
        return level
    else:
        raise TypeError(f"Unsupported level type: {type(level)}")


def convert_printable_level(level: Union[str, int]) -> str:
    if isinstance(level, str):
        return level
    if isinstance(level, int):
        if level > CRITICAL:
            return "OverCritical"
        if level == CRITICAL:
            return "Critical"
        if level > ERROR:
            return "OverError"
        if level == ERROR:
            return "Error"
        if level > WARNING:
            return "OverWarning"
        if level == WARNING:
            return "Warning"
        if level > INFO:
            return "OverInfo"
        if level == INFO:
            return "Info"
        if level > DEBUG:
            return "OverDebug"
        if level == DEBUG:
            return "Debug"
        if level > NOTSET:
            return "OverNotSet"
        if level == NOTSET:
            return "NotSet"
    return str(level)


def _logging_elapsed_seconds(
    function_name: str,
    elapsed: float,
    logger: Optional[Union[str, Logger]] = None,
    level: Optional[Union[str, int]] = None,
    callback: Optional[LoggingElapsedCallable] = None,
    round_precision: Optional[int] = None,
) -> None:
    if round_precision is not None:
        elapsed_seconds = round(elapsed, round_precision)
    else:
        elapsed_seconds = elapsed

    if callback:
        callback(function_name, elapsed_seconds)
        return

    _logger = getLogger(logger)
    _level = convert_level_number(level)

    _logger.log(
        _level,
        f"Elapsed time of function `{function_name}` is {elapsed_seconds} seconds",
    )


def logging_elapsed(
    logger: Optional[Union[str, Logger]] = None,
    level: Optional[Union[str, int]] = None,
    callback: Optional[LoggingElapsedCallable] = None,
    round_precision: Optional[int] = DEFAULT_ROUND_PRECISION,
):
    def _wrapper(func):
        if iscoroutinefunction(func):

            @wraps(func)
            async def _wrap1(*args, **kwargs):
                _begin = time()
                result = await func(*args, **kwargs)
                _logging_elapsed_seconds(
                    func.__name__,
                    time() - _begin,
                    logger,
                    level,
                    callback,
                    round_precision,
                )
                return result

            return _wrap1
        else:

            @wraps(func)
            def _wrap2(*args, **kwargs):
                _begin = time()
                result = func(*args, **kwargs)
                _logging_elapsed_seconds(
                    func.__name__,
                    time() - _begin,
                    logger,
                    level,
                    callback,
                    round_precision,
                )
                return result

            return _wrap2

    return _wrapper

# -*- coding: utf-8 -*-

from io import StringIO
from logging import DEBUG, WARNING, StreamHandler, getLogger
from time import sleep
from unittest import IsolatedAsyncioTestCase, TestCase, main

from recc.logging.logging_elapsed import logging_elapsed


class AsyncLoggingElapsedTestCase(IsolatedAsyncioTestCase):
    async def asyncSetUp(self):
        self.name = "test.recc.logging.elapsed.async"
        self.buffer = StringIO()
        self.logger = getLogger(self.name)
        self.logger.addHandler(StreamHandler(self.buffer))
        self.logger.setLevel(DEBUG)

    async def asyncTearDown(self):
        self.buffer.close()

    async def test_logging_elapsed(self):
        @logging_elapsed(self.logger, DEBUG)
        async def _test() -> None:
            sleep(0.1)

        await _test()
        self.assertTrue(self.buffer.getvalue())


class LoggingElapsedTestCase(TestCase):
    def setUp(self):
        self.name = "test.recc.logging.elapsed"
        self.buffer = StringIO()
        self.logger = getLogger(self.name)
        self.logger.addHandler(StreamHandler(self.buffer))
        self.logger.setLevel(WARNING)

    def tearDown(self):
        self.buffer.close()

    def test_logging_elapsed(self):
        @logging_elapsed(self.logger, WARNING)
        def _test() -> None:
            sleep(0.1)

        _test()
        self.assertTrue(self.buffer.getvalue())


if __name__ == "__main__":
    main()