Python:logging

로깅은 포맷 문자열을 지원한다.

LogRecord Attributes

logging — 파이썬 로깅 시설 — Python 3.9.2 문서

어트리뷰트 이름	포맷
args	직접 포맷할 필요는 없습니다.
asctime	`(asctime)s`
created	`(created)f`
exc_info	직접 포맷할 필요는 없습니다.
filename	`(filename)s`
funcName	`(funcName)s`
levelname	`(levelname)s`
levelno	`(levelno)s`
lineno	`(lineno)d`
message	`(message)s`
module	`(module)s`
msecs	`(msecs)d`
msg	직접 포맷할 필요는 없습니다.
name	`(name)s`
pathname	`(pathname)s`
process	`(process)d`
processName	`(processName)s`
relativeCreated	`(relativeCreated)d`
stack_info	직접 포맷할 필요는 없습니다.
thread	`(thread)d`
threadName	`(threadName)s`

Config file

logging.config — 로깅 구성 — Python 3.9.2 문서 # 구성 파일 형식

List of all logger names

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

recc 개발중 사용한 로거 생성 파일

# -*- 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)

Example

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)

외부 객체에 대한 액세스

logging.config — 로깅 구성 # 외부 객체에 대한 액세스

구성에서 구성 외부의 객체를 참조해야 하는 경우가 있습니다, 예를 들어 sys.stderr. 구성 딕셔너리가 파이썬 코드를 사용하여 만들어질 때는 간단하지만, 구성이 텍스트 파일(예를 들어, JSON, YAML)을 통해 제공될 때 문제가 발생합니다. 텍스트 파일에서는, sys.stderr를 리터럴 문자열 'sys.stderr'과 구별하는 표준 방법이 없습니다. 이 구별을 쉽게 하기 위해, 구성 시스템은 문자열 값에서 특정 접두사를 찾아 특수하게 처리합니다. 예를 들어, 리터럴 문자열 'ext://sys.stderr'이 구성에서 값으로 제공되면, ext://는 제거되고 값의 나머지 부분을 일반 임포트 메커니즘을 사용하여 처리합니다.

이러한 접두사의 처리는 프로토콜 처리와 유사한 방식으로 수행됩니다: 정규식 ^(?P<prefix>[a-z]+)://(?P<suffix>.*)$와 일치하는 접두사를 찾는 일반 메커니즘이 있습니다. prefix가 인식되면 suffix는 접두사 종속적 방식으로 처리되고 처리 결과가 문자열 값을 대체합니다. 접두사가 인식되지 않으면, 문자열 값은 그대로 남습니다.

Rotate logging

Python rotating log 설정하기

TimedRotatingFileHandler

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)

YAML 로깅 파일 적용 방법

python logging 사용법 – SecretMuse

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")

logging.yaml:

---

version: 1

disable_existing_loggers: False

formatters:
    simple:
        format: "%(asctime)s - %(name)s - %(levelname)s - %(message)s"

handlers:
    console:
        class: logging.StreamHandler
        level: INFO
        formatter: simple
        stream: ext://sys.stdout

    logfile:
        class: logging.handlers.RotatingFileHandler
        level: DEBUG            
        formatter: simple
        filename: ../log/info.log
        maxBytes: 10485760 # 10MB
        backupCount: 20

root:
    level: DEBUG
    handlers: [console, logfile]

Config file example

logging.config — 로깅 구성

[handler_hand02]
class=FileHandler
level=DEBUG
formatter=form02
args=('python.log', 'w')

[handler_hand03]
class=handlers.SocketHandler
level=INFO
formatter=form03
args=('localhost', handlers.DEFAULT_TCP_LOGGING_PORT)

[handler_hand04]
class=handlers.DatagramHandler
level=WARN
formatter=form04
args=('localhost', handlers.DEFAULT_UDP_LOGGING_PORT)

[handler_hand05]
class=handlers.SysLogHandler
level=ERROR
formatter=form05
args=(('localhost', handlers.SYSLOG_UDP_PORT), handlers.SysLogHandler.LOG_USER)

[handler_hand06]
class=handlers.NTEventLogHandler
level=CRITICAL
formatter=form06
args=('Python Application', '', 'Application')

[handler_hand07]
class=handlers.SMTPHandler
level=WARN
formatter=form07
args=('localhost', 'from@abc', ['user1@abc', 'user2@xyz'], 'Logger Subject')
kwargs={'timeout': 10.0}

[handler_hand08]
class=handlers.MemoryHandler
level=NOTSET
formatter=form08
target=
args=(10, ERROR)

[handler_hand09]
class=handlers.HTTPHandler
level=NOTSET
formatter=form09
args=('localhost:9022', '/log', 'GET')
kwargs={'secure': True}

YAML Logging Config Example

recc프로젝트에 사용한 yaml 로깅 설정 파일:

version: 1

disable_existing_loggers: False

formatters:
  default: &default_formatter
    format: "%(asctime)s.%(msecs)03d %(process)d/%(thread)s %(name)s %(levelname)s %(message)s"
    datefmt: "%Y-%m-%d %H:%M:%S"
    style: "%"

  simple:
    format: "{levelname[0]} {asctime} {name} {message}"
    datefmt: "%Y%m%d %H%M%S"
    style: "{"

  color:
    <<: *default_formatter
    class: recc.logging.colored_formatter.ColoredFormatter

handlers:
  stream_stdout: &stream_stdout_handler
    class: logging.StreamHandler
    level: DEBUG
    stream: ext://sys.stdout

  console_default:
    <<: *stream_stdout_handler
    formatter: default

  console_simple:
    <<: *stream_stdout_handler
    formatter: simple

  console_color:
    <<: *stream_stdout_handler
    formatter: color

  file_default:
    class: logging.FileHandler
    level: DEBUG
    formatter: default
    filename: recc.log
    mode: a
    encoding: utf-8
    delay: false

loggers:
  zeep:
    level: WARNING
  aiortc:
    level: ERROR
  aioice:
    level: ERROR
  matplotlib:
    level: INFO

#  aiohttp:
#    level: DEBUG
#  asyncio:
#    level: DEBUG
#  docker:
#    level: DEBUG
#  elasticsearch:
#    level: DEBUG
#  grpc:
#    level: DEBUG
#  recc:
#    level: DEBUG
#  urllib3:
#    level: DEBUG
#  mmdet:
#    level: DEBUG

root:
  level: DEBUG
  handlers: [console_color]

Logging elapsed decorator

# -*- 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

TestCase

# -*- 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()