Сборник рецептов по журналированию

Использование журналирования в нескольких модулях

Несколько вызовов logging.getLogger('someLogger') возвращают ссылку на один и тот же объект логгера. Это справедливо не только в пределах одного модуля, но и для разных модулей, если они работают в одном процессе интерпретатора Python. Речь идёт о ссылках на один и тот же объект; кроме того, код приложения может определить и настроить родительский логгер в одном модуле, а в другом создать дочерний логгер (но не настраивать его), и все вызовы логгера для дочернего логгера будут передаваться родительскому. Вот основной модуль:

import logging
import auxiliary_module

# create logger with 'spam_application'
logger = logging.getLogger('spam_application')
logger.setLevel(logging.DEBUG)
# create file handler which logs even debug messages
fh = logging.FileHandler('spam.log')
fh.setLevel(logging.DEBUG)
# create console handler with a higher log level
ch = logging.StreamHandler()
ch.setLevel(logging.ERROR)
# create formatter and add it to the handlers
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
fh.setFormatter(formatter)
ch.setFormatter(formatter)
# add the handlers to the logger
logger.addHandler(fh)
logger.addHandler(ch)

logger.info('creating an instance of auxiliary_module.Auxiliary')
a = auxiliary_module.Auxiliary()
logger.info('created an instance of auxiliary_module.Auxiliary')
logger.info('calling auxiliary_module.Auxiliary.do_something')
a.do_something()
logger.info('finished auxiliary_module.Auxiliary.do_something')
logger.info('calling auxiliary_module.some_function()')
auxiliary_module.some_function()
logger.info('done with auxiliary_module.some_function()')

А вот вспомогательный модуль:

import logging

# create logger
module_logger = logging.getLogger('spam_application.auxiliary')

class Auxiliary:
    def __init__(self):
        self.logger = logging.getLogger('spam_application.auxiliary.Auxiliary')
        self.logger.info('creating an instance of Auxiliary')

    def do_something(self):
        self.logger.info('doing something')
        a = 1 + 1
        self.logger.info('done doing something')

def some_function():
    module_logger.info('received a call to "some_function"')

Результат будет выглядеть так:

2005-03-23 23:47:11,663 - spam_application - INFO -
   creating an instance of auxiliary_module.Auxiliary
2005-03-23 23:47:11,665 - spam_application.auxiliary.Auxiliary - INFO -
   creating an instance of Auxiliary
2005-03-23 23:47:11,665 - spam_application - INFO -
   created an instance of auxiliary_module.Auxiliary
2005-03-23 23:47:11,668 - spam_application - INFO -
   calling auxiliary_module.Auxiliary.do_something
2005-03-23 23:47:11,668 - spam_application.auxiliary.Auxiliary - INFO -
   doing something
2005-03-23 23:47:11,669 - spam_application.auxiliary.Auxiliary - INFO -
   done doing something
2005-03-23 23:47:11,670 - spam_application - INFO -
   finished auxiliary_module.Auxiliary.do_something
2005-03-23 23:47:11,671 - spam_application - INFO -
   calling auxiliary_module.some_function()
2005-03-23 23:47:11,672 - spam_application.auxiliary - INFO -
   received a call to 'some_function'
2005-03-23 23:47:11,673 - spam_application - INFO -
   done with auxiliary_module.some_function()

Журналирование из нескольких потоков

Для журналирования из нескольких потоков не требуется никаких особых усилий. В следующем примере журналирование выполняется из главного (первичного) потока и ещё одного потока:

import logging
import threading
import time

def worker(arg):
    while not arg['stop']:
        logging.debug('Hi from myfunc')
        time.sleep(0.5)

def main():
    logging.basicConfig(level=logging.DEBUG, format='%(relativeCreated)6d %(threadName)s %(message)s')
    info = {'stop': False}
    thread = threading.Thread(target=worker, args=(info,))
    thread.start()
    while True:
        try:
            logging.debug('Hello from main')
            time.sleep(0.75)
        except KeyboardInterrupt:
            info['stop'] = True
            break
    thread.join()

if __name__ == '__main__':
    main()

При запуске скрипт должен вывести примерно следующее:

   0 Thread-1 Hi from myfunc
   3 MainThread Hello from main
 505 Thread-1 Hi from myfunc
 755 MainThread Hello from main
1007 Thread-1 Hi from myfunc
1507 MainThread Hello from main
1508 Thread-1 Hi from myfunc
2010 Thread-1 Hi from myfunc
2258 MainThread Hello from main
2512 Thread-1 Hi from myfunc
3009 MainThread Hello from main
3013 Thread-1 Hi from myfunc
3515 Thread-1 Hi from myfunc
3761 MainThread Hello from main
4017 Thread-1 Hi from myfunc
4513 MainThread Hello from main
4518 Thread-1 Hi from myfunc

Как видно, вывод журнала перемежается так, как и следовало ожидать. Разумеется, этот подход работает и с большим количеством потоков.

Несколько обработчиков и форматировщиков

Логгеры — это обычные объекты Python. Метод addHandler() не устанавливает минимального или максимального ограничения на количество добавляемых обработчиков. В некоторых случаях приложению полезно записывать все сообщения любого уровня важности в текстовый файл и одновременно выводить в консоль сообщения об ошибках и более высокого уровня. Для этого достаточно настроить соответствующие обработчики. Вызовы журналирования в коде приложения при этом останутся без изменений. Вот немного изменённый пример простой конфигурации на основе модулей из предыдущего раздела:

import logging

logger = logging.getLogger('simple_example')
logger.setLevel(logging.DEBUG)
# create file handler which logs even debug messages
fh = logging.FileHandler('spam.log')
fh.setLevel(logging.DEBUG)
# create console handler with a higher log level
ch = logging.StreamHandler()
ch.setLevel(logging.ERROR)
# create formatter and add it to the handlers
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
ch.setFormatter(formatter)
fh.setFormatter(formatter)
# add the handlers to logger
logger.addHandler(ch)
logger.addHandler(fh)

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

Обратите внимание, что код «приложения» не зависит от количества обработчиков. Изменилось лишь то, что был добавлен и настроен новый обработчик с именем fh.

Возможность создавать новые обработчики с фильтрами более высокого или низкого уровня важности может быть очень полезна при разработке и тестировании приложения. Вместо того чтобы использовать множество инструкций print для отладки, используйте logger.debug: в отличие от инструкций print, которые позже придётся удалить или закомментировать, инструкции logger.debug могут оставаться в исходном коде и не выполнять никаких действий, пока они снова не понадобятся. Тогда потребуется лишь изменить уровень важности логгера и/или обработчика на debug.

Журналирование в несколько мест назначения

Предположим, вы хотите выводить журнал в консоль и файл, используя разные форматы сообщений и разные условия. Допустим, вы хотите записывать в файл сообщения уровня DEBUG и выше, а в консоль — сообщения уровня INFO и выше. Также предположим, что в файле должны быть временные метки, а в сообщениях консоли — нет. Вот как этого добиться:

import logging

# set up logging to file - see previous section for more details
logging.basicConfig(level=logging.DEBUG,
                    format='%(asctime)s %(name)-12s %(levelname)-8s %(message)s',
                    datefmt='%m-%d %H:%M',
                    filename='/tmp/myapp.log',
                    filemode='w')
# define a Handler which writes INFO messages or higher to the sys.stderr
console = logging.StreamHandler()
console.setLevel(logging.INFO)
# set a format which is simpler for console use
formatter = logging.Formatter('%(name)-12s: %(levelname)-8s %(message)s')
# tell the handler to use this format
console.setFormatter(formatter)
# add the handler to the root logger
logging.getLogger('').addHandler(console)

# Now, we can log to the root logger, or any other logger. First the root...
logging.info('Jackdaws love my big sphinx of quartz.')

# Now, define a couple of other loggers which might represent areas in your
# application:

logger1 = logging.getLogger('myapp.area1')
logger2 = logging.getLogger('myapp.area2')

logger1.debug('Quick zephyrs blow, vexing daft Jim.')
logger1.info('How quickly daft jumping zebras vex.')
logger2.warning('Jail zesty vixen who grabbed pay from quack.')
logger2.error('The five boxing wizards jump quickly.')

При запуске в консоли появится

root        : INFO     Jackdaws love my big sphinx of quartz.
myapp.area1 : INFO     How quickly daft jumping zebras vex.
myapp.area2 : WARNING  Jail zesty vixen who grabbed pay from quack.
myapp.area2 : ERROR    The five boxing wizards jump quickly.

а в файле вы увидите примерно следующее:

10-22 22:19 root         INFO     Jackdaws love my big sphinx of quartz.
10-22 22:19 myapp.area1  DEBUG    Quick zephyrs blow, vexing daft Jim.
10-22 22:19 myapp.area1  INFO     How quickly daft jumping zebras vex.
10-22 22:19 myapp.area2  WARNING  Jail zesty vixen who grabbed pay from quack.
10-22 22:19 myapp.area2  ERROR    The five boxing wizards jump quickly.

Как видно, сообщение DEBUG появляется только в файле. Остальные сообщения отправляются в оба места назначения.

В этом примере используются обработчики консоли и файла, но можно выбрать любое количество обработчиков и любую их комбинацию.

Обратите внимание, что выбранное выше имя файла журнала /tmp/myapp.log предполагает использование стандартного расположения для временных файлов в системах POSIX. В Windows может потребоваться выбрать другое имя каталога для журнала — главное, чтобы каталог существовал и у вас были разрешения на создание и обновление файлов в нём.

Особая обработка уровней

Иногда может потребоваться немного изменить стандартную обработку уровней в обработчиках, при которой обработчик обрабатывает все уровни выше заданного порога. Для этого нужно использовать фильтры. Рассмотрим ситуацию, когда необходимо организовать работу следующим образом:

• Отправлять сообщения уровня важности INFO и WARNING в sys.stdout
• Отправлять сообщения уровня важности ERROR и выше в sys.stderr
• Отправлять сообщения уровня важности DEBUG и выше в файл app.log

Предположим, вы настроили журналирование с помощью следующего JSON:

{
    "version": 1,
    "disable_existing_loggers": false,
    "formatters": {
        "simple": {
            "format": "%(levelname)-8s - %(message)s"
        }
    },
    "handlers": {
        "stdout": {
            "class": "logging.StreamHandler",
            "level": "INFO",
            "formatter": "simple",
            "stream": "ext://sys.stdout"
        },
        "stderr": {
            "class": "logging.StreamHandler",
            "level": "ERROR",
            "formatter": "simple",
            "stream": "ext://sys.stderr"
        },
        "file": {
            "class": "logging.FileHandler",
            "formatter": "simple",
            "filename": "app.log",
            "mode": "w"
        }
    },
    "root": {
        "level": "DEBUG",
        "handlers": [
            "stderr",
            "stdout",
            "file"
        ]
    }
}

Эта конфигурация делает почти то, что нужно, за исключением того, что sys.stdout будет отображать сообщения уровня важности ERROR, а также отслеживать только события этого уровня и выше, как и сообщения INFO и WARNING. Чтобы этого избежать, можно настроить фильтр, исключающий эти сообщения, и добавить его к соответствующему обработчику. Для этого добавим раздел filters на том же уровне, что и formatters и handlers:

{
    "filters": {
        "warnings_and_below": {
            "()" : "__main__.filter_maker",
            "level": "WARNING"
        }
    }
}

и изменим раздел обработчика stdout, добавив в него этот фильтр:

{
    "stdout": {
        "class": "logging.StreamHandler",
        "level": "INFO",
        "formatter": "simple",
        "stream": "ext://sys.stdout",
        "filters": ["warnings_and_below"]
    }
}

Фильтр — это просто функция, поэтому можно определить filter_maker (функцию-фабрику) следующим образом:

def filter_maker(level):
    level = getattr(logging, level)

    def filter(record):
        return record.levelno <= level

    return filter

Она преобразует строковый аргумент в числовой уровень и возвращает функцию, которая возвращает True только в том случае, если уровень переданной записи не выше указанного. Обратите внимание: в этом примере я определил filter_maker в тестовом скрипте main.py, который запускаю из командной строки, поэтому его модулем будет __main__ — отсюда и __main__.filter_maker в конфигурации фильтра. Если вы определите его в другом модуле, это значение нужно будет изменить.

Добавив фильтр, можно запустить main.py, полный код которого выглядит так:

import json
import logging
import logging.config

CONFIG = '''
{
    "version": 1,
    "disable_existing_loggers": false,
    "formatters": {
        "simple": {
            "format": "%(levelname)-8s - %(message)s"
        }
    },
    "filters": {
        "warnings_and_below": {
            "()" : "__main__.filter_maker",
            "level": "WARNING"
        }
    },
    "handlers": {
        "stdout": {
            "class": "logging.StreamHandler",
            "level": "INFO",
            "formatter": "simple",
            "stream": "ext://sys.stdout",
            "filters": ["warnings_and_below"]
        },
        "stderr": {
            "class": "logging.StreamHandler",
            "level": "ERROR",
            "formatter": "simple",
            "stream": "ext://sys.stderr"
        },
        "file": {
            "class": "logging.FileHandler",
            "formatter": "simple",
            "filename": "app.log",
            "mode": "w"
        }
    },
    "root": {
        "level": "DEBUG",
        "handlers": [
            "stderr",
            "stdout",
            "file"
        ]
    }
}
'''

def filter_maker(level):
    level = getattr(logging, level)

    def filter(record):
        return record.levelno <= level

    return filter

logging.config.dictConfig(json.loads(CONFIG))
logging.debug('A DEBUG message')
logging.info('An INFO message')
logging.warning('A WARNING message')
logging.error('An ERROR message')
logging.critical('A CRITICAL message')

После запуска следующим образом:

python main.py 2>stderr.log >stdout.log

мы увидим ожидаемые результаты:

$ more *.log
::::::::::::::
app.log
::::::::::::::
DEBUG    - A DEBUG message
INFO     - An INFO message
WARNING  - A WARNING message
ERROR    - An ERROR message
CRITICAL - A CRITICAL message
::::::::::::::
stderr.log
::::::::::::::
ERROR    - An ERROR message
CRITICAL - A CRITICAL message
::::::::::::::
stdout.log
::::::::::::::
INFO     - An INFO message
WARNING  - A WARNING message

Пример сервера конфигурации

Вот пример модуля, использующего сервер конфигурации журналирования:

import logging
import logging.config
import time
import os

# read initial config file
logging.config.fileConfig('logging.conf')

# create and start listener on port 9999
t = logging.config.listen(9999)
t.start()

logger = logging.getLogger('simpleExample')

try:
    # loop through logging calls to see the difference
    # new configurations make, until Ctrl+C is pressed
    while True:
        logger.debug('debug message')
        logger.info('info message')
        logger.warning('warn message')
        logger.error('error message')
        logger.critical('critical message')
        time.sleep(5)
except KeyboardInterrupt:
    # cleanup
    logging.config.stopListening()
    t.join()

А вот скрипт, который принимает имя файла и отправляет этот файл серверу в качестве новой конфигурации журналирования, предварительно добавив его длину в двоичном формате:

#!/usr/bin/env python
import socket, sys, struct

with open(sys.argv[1], 'rb') as f:
    data_to_send = f.read()

HOST = 'localhost'
PORT = 9999
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
print('connecting...')
s.connect((HOST, PORT))
print('sending config...')
s.send(struct.pack('>L', len(data_to_send)))
s.send(data_to_send)
s.close()
print('complete')

Работа с блокирующими обработчиками

Иногда необходимо, чтобы обработчики журналирования выполняли свою работу, не блокируя поток, из которого ведётся журналирование. Это часто встречается в веб-приложениях, хотя, разумеется, бывает и в других ситуациях.

Распространённой причиной замедления работы служит SMTPHandler: отправка электронных писем может занять много времени по причинам, не зависящим от разработчика (например, из-за медленно работающей почтовой или сетевой инфраструктуры). Однако блокировать работу может почти любой обработчик, использующий сеть: даже операция SocketHandler может незаметно для вас выполнить слишком медленный DNS-запрос (причём этот запрос может выполняться глубоко в коде библиотеки сокетов, ниже уровня Python и вне вашего контроля).

Одно из решений — использовать двухэтапный подход. На первом этапе к логгерам, вызываемым из критичных по производительности потоков, следует подключить только QueueHandler. Они просто записывают данные в очередь, размер которой можно задать достаточно большим или не ограничивать вовсе. Запись в очередь обычно выполняется быстро, хотя в качестве меры предосторожности в коде, вероятно, следует перехватывать исключение queue.Full. Если вы разрабатываете библиотеку, в коде которой есть критичные по производительности потоки, обязательно задокументируйте это (и порекомендуйте подключать к логгерам только QueueHandlers), чтобы помочь другим разработчикам, использующим ваш код.

Второй этап решения — это QueueListener, созданный как дополнение к QueueHandler. Устройство QueueListener очень простое: ему передают очередь и несколько обработчиков, после чего он запускает внутренний поток, который ожидает в очереди записи LogRecord, отправленные из QueueHandlers (или из любого другого источника LogRecords). LogRecords извлекаются из очереди и передаются обработчикам для обработки.

Преимущество отдельного класса QueueListener в том, что один его экземпляр может обслуживать несколько QueueHandlers. Это экономнее расходует ресурсы, чем, например, создание потоковых версий существующих классов обработчиков, которые без особой пользы занимали бы по одному потоку на каждый обработчик.

Пример использования этих двух классов (импорты опущены):

que = queue.Queue(-1)  # no limit on size
queue_handler = QueueHandler(que)
handler = logging.StreamHandler()
listener = QueueListener(que, handler)
root = logging.getLogger()
root.addHandler(queue_handler)
formatter = logging.Formatter('%(threadName)s: %(message)s')
handler.setFormatter(formatter)
listener.start()
# The log output will display the thread which generated
# the event (the main thread) rather than the internal
# thread which monitors the internal queue. This is what
# you want to happen.
root.warning('Look out!')
listener.stop()

При запуске он выведет:

MainThread: Look out!

Отправка и получение событий журналирования по сети

Предположим, вы хотите отправлять события журналирования по сети и обрабатывать их на принимающей стороне. Простой способ сделать это — подключить экземпляр SocketHandler к корневому логгеру на отправляющей стороне:

import logging, logging.handlers

rootLogger = logging.getLogger('')
rootLogger.setLevel(logging.DEBUG)
socketHandler = logging.handlers.SocketHandler('localhost',
                    logging.handlers.DEFAULT_TCP_LOGGING_PORT)
# don't bother with a formatter, since a socket handler sends the event as
# an unformatted pickle
rootLogger.addHandler(socketHandler)

# Now, we can log to the root logger, or any other logger. First the root...
logging.info('Jackdaws love my big sphinx of quartz.')

# Now, define a couple of other loggers which might represent areas in your
# application:

logger1 = logging.getLogger('myapp.area1')
logger2 = logging.getLogger('myapp.area2')

logger1.debug('Quick zephyrs blow, vexing daft Jim.')
logger1.info('How quickly daft jumping zebras vex.')
logger2.warning('Jail zesty vixen who grabbed pay from quack.')
logger2.error('The five boxing wizards jump quickly.')

На принимающей стороне можно настроить получатель с помощью модуля socketserver. Вот простой рабочий пример:

import pickle
import logging
import logging.handlers
import socketserver
import struct


class LogRecordStreamHandler(socketserver.StreamRequestHandler):
    """Handler for a streaming logging request.

    This basically logs the record using whatever logging policy is
    configured locally.
    """

    def handle(self):
        """
        Handle multiple requests - each expected to be a 4-byte length,
        followed by the LogRecord in pickle format. Logs the record
        according to whatever policy is configured locally.
        """
        while True:
            chunk = self.connection.recv(4)
            if len(chunk) < 4:
                break
            slen = struct.unpack('>L', chunk)[0]
            chunk = self.connection.recv(slen)
            while len(chunk) < slen:
                chunk = chunk + self.connection.recv(slen - len(chunk))
            obj = self.unPickle(chunk)
            record = logging.makeLogRecord(obj)
            self.handleLogRecord(record)

    def unPickle(self, data):
        return pickle.loads(data)

    def handleLogRecord(self, record):
        # if a name is specified, we use the named logger rather than the one
        # implied by the record.
        if self.server.logname is not None:
            name = self.server.logname
        else:
            name = record.name
        logger = logging.getLogger(name)
        # N.B. EVERY record gets logged. This is because Logger.handle
        # is normally called AFTER logger-level filtering. If you want
        # to do filtering, do it at the client end to save wasting
        # cycles and network bandwidth!
        logger.handle(record)

class LogRecordSocketReceiver(socketserver.ThreadingTCPServer):
    """
    Simple TCP socket-based logging receiver suitable for testing.
    """

    allow_reuse_address = True

    def __init__(self, host='localhost',
                 port=logging.handlers.DEFAULT_TCP_LOGGING_PORT,
                 handler=LogRecordStreamHandler):
        socketserver.ThreadingTCPServer.__init__(self, (host, port), handler)
        self.abort = 0
        self.timeout = 1
        self.logname = None

    def serve_until_stopped(self):
        import select
        abort = 0
        while not abort:
            rd, wr, ex = select.select([self.socket.fileno()],
                                       [], [],
                                       self.timeout)
            if rd:
                self.handle_request()
            abort = self.abort

def main():
    logging.basicConfig(
        format='%(relativeCreated)5d %(name)-15s %(levelname)-8s %(message)s')
    tcpserver = LogRecordSocketReceiver()
    print('About to start TCP server...')
    tcpserver.serve_until_stopped()

if __name__ == '__main__':
    main()

Сначала запустите сервер, а затем клиент. На стороне клиента в консоли ничего не выводится; на стороне сервера вы должны увидеть примерно следующее:

About to start TCP server...
   59 root            INFO     Jackdaws love my big sphinx of quartz.
   59 myapp.area1     DEBUG    Quick zephyrs blow, vexing daft Jim.
   69 myapp.area1     INFO     How quickly daft jumping zebras vex.
   69 myapp.area2     WARNING  Jail zesty vixen who grabbed pay from quack.
   69 myapp.area2     ERROR    The five boxing wizards jump quickly.

Обратите внимание: в некоторых ситуациях использование pickle связано с проблемами безопасности. Если это актуально для вас, можно использовать альтернативный способ сериализации, переопределив метод makePickle() и реализовав в нём альтернативный вариант, а также адаптировав приведённый выше скрипт для его использования.

Добавление контекстной информации в вывод журнала

Иногда помимо параметров, переданных вызову журналирования, необходимо включить в вывод журнала контекстную информацию. Например, в сетевом приложении может быть полезно записывать в журнал сведения о клиенте (например, имя пользователя удалённого клиента или IP-адрес). Для этого можно использовать параметр extra, но передавать таким образом информацию не всегда удобно. Может возникнуть соблазн создавать экземпляры Logger для каждого соединения, но это плохая идея, поскольку такие экземпляры не подвергаются сборке мусора. На практике это не проблема, если количество экземпляров Logger зависит от требуемой степени детализации журналирования приложения, однако управлять ими может быть сложно, если количество экземпляров Logger станет практически неограниченным.

def debug(self, msg, /, *args, **kwargs):
    """
    Delegate a debug call to the underlying logger, after adding
    contextual information from this adapter instance.
    """
    msg, kwargs = self.process(msg, kwargs)
    self.logger.debug(msg, *args, **kwargs)

class CustomAdapter(logging.LoggerAdapter):
    """
    This example adapter expects the passed in dict-like object to have a
    'connid' key, whose value in brackets is prepended to the log message.
    """
    def process(self, msg, kwargs):
        return '[%s] %s' % (self.extra['connid'], msg), kwargs

logger = logging.getLogger(__name__)
adapter = CustomAdapter(logger, {'connid': some_conn_id})

import logging
from random import choice

class ContextFilter(logging.Filter):
    """
    This is a filter which injects contextual information into the log.

    Rather than use actual contextual information, we just use random
    data in this demo.
    """

    USERS = ['jim', 'fred', 'sheila']
    IPS = ['123.231.231.123', '127.0.0.1', '192.168.0.1']

    def filter(self, record):

        record.ip = choice(ContextFilter.IPS)
        record.user = choice(ContextFilter.USERS)
        return True

if __name__ == '__main__':
    levels = (logging.DEBUG, logging.INFO, logging.WARNING, logging.ERROR, logging.CRITICAL)
    logging.basicConfig(level=logging.DEBUG,
                        format='%(asctime)-15s %(name)-5s %(levelname)-8s IP: %(ip)-15s User: %(user)-8s %(message)s')
    a1 = logging.getLogger('a.b.c')
    a2 = logging.getLogger('d.e.f')

    f = ContextFilter()
    a1.addFilter(f)
    a2.addFilter(f)
    a1.debug('A debug message')
    a1.info('An info message with %s', 'some parameters')
    for x in range(10):
        lvl = choice(levels)
        lvlname = logging.getLevelName(lvl)
        a2.log(lvl, 'A message at %s level with %d %s', lvlname, 2, 'parameters')

2010-09-06 22:38:15,292 a.b.c DEBUG    IP: 123.231.231.123 User: fred     A debug message
2010-09-06 22:38:15,300 a.b.c INFO     IP: 192.168.0.1     User: sheila   An info message with some parameters
2010-09-06 22:38:15,300 d.e.f CRITICAL IP: 127.0.0.1       User: sheila   A message at CRITICAL level with 2 parameters
2010-09-06 22:38:15,300 d.e.f ERROR    IP: 127.0.0.1       User: jim      A message at ERROR level with 2 parameters
2010-09-06 22:38:15,300 d.e.f DEBUG    IP: 127.0.0.1       User: sheila   A message at DEBUG level with 2 parameters
2010-09-06 22:38:15,300 d.e.f ERROR    IP: 123.231.231.123 User: fred     A message at ERROR level with 2 parameters
2010-09-06 22:38:15,300 d.e.f CRITICAL IP: 192.168.0.1     User: jim      A message at CRITICAL level with 2 parameters
2010-09-06 22:38:15,300 d.e.f CRITICAL IP: 127.0.0.1       User: sheila   A message at CRITICAL level with 2 parameters
2010-09-06 22:38:15,300 d.e.f DEBUG    IP: 192.168.0.1     User: jim      A message at DEBUG level with 2 parameters
2010-09-06 22:38:15,301 d.e.f ERROR    IP: 127.0.0.1       User: sheila   A message at ERROR level with 2 parameters
2010-09-06 22:38:15,301 d.e.f DEBUG    IP: 123.231.231.123 User: fred     A message at DEBUG level with 2 parameters
2010-09-06 22:38:15,301 d.e.f INFO     IP: 123.231.231.123 User: fred     A message at INFO level with 2 parameters

Использование contextvars

Начиная с Python 3.7 модуль contextvars предоставляет хранилище контекстных данных, работающее как с threading, так и с задачами обработки в asyncio. Поэтому такой тип хранилища может быть предпочтительнее локальных переменных потоков. В следующем примере показано, как в многопоточной среде добавлять в журналы контекстную информацию, например атрибуты запросов, обрабатываемых веб-приложениями.

Для наглядности предположим, что у вас есть несколько веб-приложений, каждое из которых независимо от остальных, но работает в том же процессе Python и использует общую библиотеку. Как добиться того, чтобы у каждого приложения был собственный журнал, куда направляются все сообщения журналирования из библиотеки (и другого кода обработки запросов), при этом в журнал включается дополнительная контекстная информация, например IP-адрес клиента, метод HTTP-запроса и имя пользователя клиента?

Предположим, что библиотеку можно смоделировать следующим кодом:

# webapplib.py
import logging
import time

logger = logging.getLogger(__name__)

def useful():
    # Just a representative event logged from the library
    logger.debug('Hello from webapplib!')
    # Just sleep for a bit so other threads get to run
    time.sleep(0.01)

Смоделировать несколько веб-приложений можно с помощью двух простых классов: Request и WebApp. Они имитируют работу настоящих многопоточных веб-приложений, где каждый запрос обрабатывается в отдельном потоке:

# main.py
import argparse
from contextvars import ContextVar
import logging
import os
from random import choice
import threading
import webapplib

logger = logging.getLogger(__name__)
root = logging.getLogger()
root.setLevel(logging.DEBUG)

class Request:
    """
    A simple dummy request class which just holds dummy HTTP request method,
    client IP address and client username
    """
    def __init__(self, method, ip, user):
        self.method = method
        self.ip = ip
        self.user = user

# A dummy set of requests which will be used in the simulation - we'll just pick
# from this list randomly. Note that all GET requests are from 192.168.2.XXX
# addresses, whereas POST requests are from 192.16.3.XXX addresses. Three users
# are represented in the sample requests.

REQUESTS = [
    Request('GET', '192.168.2.20', 'jim'),
    Request('POST', '192.168.3.20', 'fred'),
    Request('GET', '192.168.2.21', 'sheila'),
    Request('POST', '192.168.3.21', 'jim'),
    Request('GET', '192.168.2.22', 'fred'),
    Request('POST', '192.168.3.22', 'sheila'),
]

# Note that the format string includes references to request context information
# such as HTTP method, client IP and username

formatter = logging.Formatter('%(threadName)-11s %(appName)s %(name)-9s %(user)-6s %(ip)s %(method)-4s %(message)s')

# Create our context variables. These will be filled at the start of request
# processing, and used in the logging that happens during that processing

ctx_request = ContextVar('request')
ctx_appname = ContextVar('appname')

class InjectingFilter(logging.Filter):
    """
    A filter which injects context-specific information into logs and ensures
    that only information for a specific webapp is included in its log
    """
    def __init__(self, app):
        self.app = app

    def filter(self, record):
        request = ctx_request.get()
        record.method = request.method
        record.ip = request.ip
        record.user = request.user
        record.appName = appName = ctx_appname.get()
        return appName == self.app.name

class WebApp:
    """
    A dummy web application class which has its own handler and filter for a
    webapp-specific log.
    """
    def __init__(self, name):
        self.name = name
        handler = logging.FileHandler(name + '.log', 'w')
        f = InjectingFilter(self)
        handler.setFormatter(formatter)
        handler.addFilter(f)
        root.addHandler(handler)
        self.num_requests = 0

    def process_request(self, request):
        """
        This is the dummy method for processing a request. It's called on a
        different thread for every request. We store the context information into
        the context vars before doing anything else.
        """
        ctx_request.set(request)
        ctx_appname.set(self.name)
        self.num_requests += 1
        logger.debug('Request processing started')
        webapplib.useful()
        logger.debug('Request processing finished')

def main():
    fn = os.path.splitext(os.path.basename(__file__))[0]
    adhf = argparse.ArgumentDefaultsHelpFormatter
    ap = argparse.ArgumentParser(formatter_class=adhf, prog=fn,
                                 description='Simulate a couple of web '
                                             'applications handling some '
                                             'requests, showing how request '
                                             'context can be used to '
                                             'populate logs')
    aa = ap.add_argument
    aa('--count', '-c', type=int, default=100, help='How many requests to simulate')
    options = ap.parse_args()

    # Create the dummy webapps and put them in a list which we can use to select
    # from randomly
    app1 = WebApp('app1')
    app2 = WebApp('app2')
    apps = [app1, app2]
    threads = []
    # Add a common handler which will capture all events
    handler = logging.FileHandler('app.log', 'w')
    handler.setFormatter(formatter)
    root.addHandler(handler)

    # Generate calls to process requests
    for i in range(options.count):
        try:
            # Pick an app at random and a request for it to process
            app = choice(apps)
            request = choice(REQUESTS)
            # Process the request in its own thread
            t = threading.Thread(target=app.process_request, args=(request,))
            threads.append(t)
            t.start()
        except KeyboardInterrupt:
            break

    # Wait for the threads to terminate
    for t in threads:
        t.join()

    for app in apps:
        print('%s processed %s requests' % (app.name, app.num_requests))

if __name__ == '__main__':
    main()

Если запустить приведённый выше код, примерно половина запросов должна попасть в app1.log, а остальные — в app2.log; при этом все запросы будут записаны в app.log. В журнале каждого веб-приложения будут только записи, относящиеся к этому приложению, а сведения о запросах будут последовательно отображаться в журнале (то есть информация о каждом фиктивном запросе всегда будет находиться вместе в одной строке журнала). Это показано в следующем выводе оболочки:

~/logging-contextual-webapp$ python main.py
app1 processed 51 requests
app2 processed 49 requests
~/logging-contextual-webapp$ wc -l *.log
  153 app1.log
  147 app2.log
  300 app.log
  600 total
~/logging-contextual-webapp$ head -3 app1.log
Thread-3 (process_request) app1 __main__  jim    192.168.3.21 POST Request processing started
Thread-3 (process_request) app1 webapplib jim    192.168.3.21 POST Hello from webapplib!
Thread-5 (process_request) app1 __main__  jim    192.168.3.21 POST Request processing started
~/logging-contextual-webapp$ head -3 app2.log
Thread-1 (process_request) app2 __main__  sheila 192.168.2.21 GET  Request processing started
Thread-1 (process_request) app2 webapplib sheila 192.168.2.21 GET  Hello from webapplib!
Thread-2 (process_request) app2 __main__  jim    192.168.2.20 GET  Request processing started
~/logging-contextual-webapp$ head app.log
Thread-1 (process_request) app2 __main__  sheila 192.168.2.21 GET  Request processing started
Thread-1 (process_request) app2 webapplib sheila 192.168.2.21 GET  Hello from webapplib!
Thread-2 (process_request) app2 __main__  jim    192.168.2.20 GET  Request processing started
Thread-3 (process_request) app1 __main__  jim    192.168.3.21 POST Request processing started
Thread-2 (process_request) app2 webapplib jim    192.168.2.20 GET  Hello from webapplib!
Thread-3 (process_request) app1 webapplib jim    192.168.3.21 POST Hello from webapplib!
Thread-4 (process_request) app2 __main__  fred   192.168.2.22 GET  Request processing started
Thread-5 (process_request) app1 __main__  jim    192.168.3.21 POST Request processing started
Thread-4 (process_request) app2 webapplib fred   192.168.2.22 GET  Hello from webapplib!
Thread-6 (process_request) app1 __main__  jim    192.168.3.21 POST Request processing started
~/logging-contextual-webapp$ grep app1 app1.log | wc -l
153
~/logging-contextual-webapp$ grep app2 app2.log | wc -l
147
~/logging-contextual-webapp$ grep app1 app.log | wc -l
153
~/logging-contextual-webapp$ grep app2 app.log | wc -l
147

Передача контекстной информации в обработчиках

У каждого Handler есть собственная цепочка фильтров. Если нужно добавить контекстную информацию в LogRecord, не передавая её другим обработчикам, можно использовать фильтр, который возвращает новую LogRecord, а не изменяет существующую запись на месте, как показано в следующем скрипте:

import copy
import logging

def filter(record: logging.LogRecord):
    record = copy.copy(record)
    record.user = 'jim'
    return record

if __name__ == '__main__':
    logger = logging.getLogger()
    logger.setLevel(logging.INFO)
    handler = logging.StreamHandler()
    formatter = logging.Formatter('%(message)s from %(user)-8s')
    handler.setFormatter(formatter)
    handler.addFilter(filter)
    logger.addHandler(handler)

    logger.info('A log message')

Ведение журнала в одном файле из нескольких процессов

Хотя ведение журнала потокобезопасно и поддерживается запись в один файл из нескольких потоков одного процесса, запись в один файл из нескольких процессов не поддерживается, поскольку в Python нет стандартного способа сериализовать доступ к одному файлу из нескольких процессов. Если вам нужно вести журнал в одном файле из нескольких процессов, один из способов — настроить все процессы на запись в SocketHandler, а в отдельном процессе реализовать сервер сокетов, который считывает данные из сокета и записывает их в файл. (При желании для этой задачи можно выделить один поток в одном из существующих процессов.) В этом разделе этот подход описан подробнее и приведён работающий приёмник сокетов, который можно использовать в качестве основы для адаптации под собственные приложения.

Можно также написать собственный обработчик, который использует класс Lock из модуля multiprocessing для сериализации доступа процессов к файлу. Стандартные библиотечные классы FileHandler и их подклассы не используют multiprocessing.

В качестве альтернативы можно использовать Queue и QueueHandler, чтобы отправлять все события журнала одному из процессов многопроцессного приложения. В следующем примере скрипта показано, как это сделать: отдельный процесс-приёмник получает события, отправленные другими процессами, и записывает их в соответствии со своей конфигурацией ведения журнала. Хотя в примере показан только один способ (например, вместо отдельного процесса-приёмника можно использовать поток-приёмник — реализация будет аналогичной), он позволяет задать совершенно разные конфигурации ведения журнала для приёмника и остальных процессов приложения и может послужить основой для кода, отвечающего вашим конкретным требованиям:

# You'll need these imports in your own code
import logging
import logging.handlers
import multiprocessing

# Next two import lines for this demo only
from random import choice, random
import time

#
# Because you'll want to define the logging configurations for listener and workers, the
# listener and worker process functions take a configurer parameter which is a callable
# for configuring logging for that process. These functions are also passed the queue,
# which they use for communication.
#
# In practice, you can configure the listener however you want, but note that in this
# simple example, the listener does not apply level or filter logic to received records.
# In practice, you would probably want to do this logic in the worker processes, to avoid
# sending events which would be filtered out between processes.
#
# The size of the rotated files is made small so you can see the results easily.
def listener_configurer():
    root = logging.getLogger()
    h = logging.handlers.RotatingFileHandler('mptest.log', 'a', 300, 10)
    f = logging.Formatter('%(asctime)s %(processName)-10s %(name)s %(levelname)-8s %(message)s')
    h.setFormatter(f)
    root.addHandler(h)

# This is the listener process top-level loop: wait for logging events
# (LogRecords)on the queue and handle them, quit when you get a None for a
# LogRecord.
def listener_process(queue, configurer):
    configurer()
    while True:
        try:
            record = queue.get()
            if record is None:  # We send this as a sentinel to tell the listener to quit.
                break
            logger = logging.getLogger(record.name)
            logger.handle(record)  # No level or filter logic applied - just do it!
        except Exception:
            import sys, traceback
            print('Whoops! Problem:', file=sys.stderr)
            traceback.print_exc(file=sys.stderr)

# Arrays used for random selections in this demo

LEVELS = [logging.DEBUG, logging.INFO, logging.WARNING,
          logging.ERROR, logging.CRITICAL]

LOGGERS = ['a.b.c', 'd.e.f']

MESSAGES = [
    'Random message #1',
    'Random message #2',
    'Random message #3',
]

# The worker configuration is done at the start of the worker process run.
# Note that on Windows you can't rely on fork semantics, so each process
# will run the logging configuration code when it starts.
def worker_configurer(queue):
    h = logging.handlers.QueueHandler(queue)  # Just the one handler needed
    root = logging.getLogger()
    root.addHandler(h)
    # send all messages, for demo; no other level or filter logic applied.
    root.setLevel(logging.DEBUG)

# This is the worker process top-level loop, which just logs ten events with
# random intervening delays before terminating.
# The print messages are just so you know it's doing something!
def worker_process(queue, configurer):
    configurer(queue)
    name = multiprocessing.current_process().name
    print('Worker started: %s' % name)
    for i in range(10):
        time.sleep(random())
        logger = logging.getLogger(choice(LOGGERS))
        level = choice(LEVELS)
        message = choice(MESSAGES)
        logger.log(level, message)
    print('Worker finished: %s' % name)

# Here's where the demo gets orchestrated. Create the queue, create and start
# the listener, create ten workers and start them, wait for them to finish,
# then send a None to the queue to tell the listener to finish.
def main():
    queue = multiprocessing.Queue(-1)
    listener = multiprocessing.Process(target=listener_process,
                                       args=(queue, listener_configurer))
    listener.start()
    workers = []
    for i in range(10):
        worker = multiprocessing.Process(target=worker_process,
                                         args=(queue, worker_configurer))
        workers.append(worker)
        worker.start()
    for w in workers:
        w.join()
    queue.put_nowait(None)
    listener.join()

if __name__ == '__main__':
    main()

Вариант приведённого выше скрипта ведёт журнал в главном процессе, в отдельном потоке:

import logging
import logging.config
import logging.handlers
from multiprocessing import Process, Queue
import random
import threading
import time

def logger_thread(q):
    while True:
        record = q.get()
        if record is None:
            break
        logger = logging.getLogger(record.name)
        logger.handle(record)


def worker_process(q):
    qh = logging.handlers.QueueHandler(q)
    root = logging.getLogger()
    root.setLevel(logging.DEBUG)
    root.addHandler(qh)
    levels = [logging.DEBUG, logging.INFO, logging.WARNING, logging.ERROR,
              logging.CRITICAL]
    loggers = ['foo', 'foo.bar', 'foo.bar.baz',
               'spam', 'spam.ham', 'spam.ham.eggs']
    for i in range(100):
        lvl = random.choice(levels)
        logger = logging.getLogger(random.choice(loggers))
        logger.log(lvl, 'Message no. %d', i)

if __name__ == '__main__':
    q = Queue()
    d = {
        'version': 1,
        'formatters': {
            'detailed': {
                'class': 'logging.Formatter',
                'format': '%(asctime)s %(name)-15s %(levelname)-8s %(processName)-10s %(message)s'
            }
        },
        'handlers': {
            'console': {
                'class': 'logging.StreamHandler',
                'level': 'INFO',
            },
            'file': {
                'class': 'logging.FileHandler',
                'filename': 'mplog.log',
                'mode': 'w',
                'formatter': 'detailed',
            },
            'foofile': {
                'class': 'logging.FileHandler',
                'filename': 'mplog-foo.log',
                'mode': 'w',
                'formatter': 'detailed',
            },
            'errors': {
                'class': 'logging.FileHandler',
                'filename': 'mplog-errors.log',
                'mode': 'w',
                'level': 'ERROR',
                'formatter': 'detailed',
            },
        },
        'loggers': {
            'foo': {
                'handlers': ['foofile']
            }
        },
        'root': {
            'level': 'DEBUG',
            'handlers': ['console', 'file', 'errors']
        },
    }
    workers = []
    for i in range(5):
        wp = Process(target=worker_process, name='worker %d' % (i + 1), args=(q,))
        workers.append(wp)
        wp.start()
    logging.config.dictConfig(d)
    lp = threading.Thread(target=logger_thread, args=(q,))
    lp.start()
    # At this point, the main process could do some useful work of its own
    # Once it's done that, it can wait for the workers to terminate...
    for wp in workers:
        wp.join()
    # And now tell the logging thread to finish up, too
    q.put(None)
    lp.join()

Этот вариант показывает, как, например, применить конфигурацию для определённых регистраторов: у регистратора foo есть специальный обработчик, который сохраняет все события подсистемы foo в файл mplog-foo.log. Механизм ведения журнала главного процесса будет использовать эту конфигурацию (хотя события журнала создаются в рабочих процессах), чтобы направлять сообщения в соответствующие места назначения.

queue = multiprocessing.Queue(-1)

queue = multiprocessing.Manager().Queue(-1)  # also works with the examples above

workers = []
for i in range(10):
    worker = multiprocessing.Process(target=worker_process,
                                     args=(queue, worker_configurer))
    workers.append(worker)
    worker.start()
for w in workers:
    w.join()

with concurrent.futures.ProcessPoolExecutor(max_workers=10) as executor:
    for i in range(10):
        executor.submit(worker_process, queue, worker_configurer)

Ротация файлов

Иногда нужно позволить файлу журнала вырасти до определённого размера, а затем открыть новый файл и продолжить запись в него. Может потребоваться хранить определённое число таких файлов, а после создания нужного количества выполнять их ротацию, чтобы ограничить и количество файлов, и их размер. Для такого сценария пакет logging предоставляет обработчик RotatingFileHandler:

import glob
import logging
import logging.handlers

LOG_FILENAME = 'logging_rotatingfile_example.out'

# Set up a specific logger with our desired output level
my_logger = logging.getLogger('MyLogger')
my_logger.setLevel(logging.DEBUG)

# Add the log message handler to the logger
handler = logging.handlers.RotatingFileHandler(
              LOG_FILENAME, maxBytes=20, backupCount=5)

my_logger.addHandler(handler)

# Log some messages
for i in range(20):
    my_logger.debug('i = %d' % i)

# See what files are created
logfiles = glob.glob('%s*' % LOG_FILENAME)

for filename in logfiles:
    print(filename)

В результате должны получиться 6 отдельных файлов, каждый из которых содержит часть журнала приложения:

logging_rotatingfile_example.out
logging_rotatingfile_example.out.1
logging_rotatingfile_example.out.2
logging_rotatingfile_example.out.3
logging_rotatingfile_example.out.4
logging_rotatingfile_example.out.5

Самый актуальный файл всегда называется logging_rotatingfile_example.out, и при достижении ограничения размера ему присваивается имя с суффиксом .1. Имена всех существующих резервных файлов меняются так, чтобы суффикс увеличился (.1 превращается в .2 и т. д.), а файл .6 удаляется.

Очевидно, что в этом примере длина журнала задана слишком маленькой — это сделано для наглядности. Параметру maxBytes следует задать подходящее значение.

Использование альтернативных стилей форматирования

Когда в стандартную библиотеку Python добавили ведение журнала, единственным способом форматирования сообщений с переменным содержимым был %-стиль форматирования. С тех пор в Python появились ещё два способа форматирования: string.Template (добавлен в Python 2.4) и str.format() (добавлен в Python 2.6).

Начиная с версии 3.2, ведение журнала обеспечивает улучшенную поддержку этих двух дополнительных стилей форматирования. Класс Formatter был дополнен необязательным именованным параметром style. По умолчанию ему присваивается значение '%', но также можно указать '{' или '$' — эти значения соответствуют двум другим стилям форматирования. Обратная совместимость сохраняется по умолчанию, как и следовало ожидать, но явное указание параметра style позволяет задавать строки форматирования, совместимые с str.format() или string.Template. Возможности демонстрирует следующий пример сеанса работы в консоли:

>>> import logging
>>> root = logging.getLogger()
>>> root.setLevel(logging.DEBUG)
>>> handler = logging.StreamHandler()
>>> bf = logging.Formatter('{asctime} {name} {levelname:8s} {message}',
...                        style='{')
>>> handler.setFormatter(bf)
>>> root.addHandler(handler)
>>> logger = logging.getLogger('foo.bar')
>>> logger.debug('This is a DEBUG message')
2010-10-28 15:11:55,341 foo.bar DEBUG    This is a DEBUG message
>>> logger.critical('This is a CRITICAL message')
2010-10-28 15:12:11,526 foo.bar CRITICAL This is a CRITICAL message
>>> df = logging.Formatter('$asctime $name ${levelname} $message',
...                        style='$')
>>> handler.setFormatter(df)
>>> logger.debug('This is a DEBUG message')
2010-10-28 15:13:06,924 foo.bar DEBUG This is a DEBUG message
>>> logger.critical('This is a CRITICAL message')
2010-10-28 15:13:11,494 foo.bar CRITICAL This is a CRITICAL message
>>>

Обратите внимание, что форматирование сообщений журнала для окончательного вывода полностью независимо от способа формирования отдельного сообщения журнала. Для этого по-прежнему можно использовать %-форматирование, как показано здесь:

>>> logger.error('This is an%s %s %s', 'other,', 'ERROR,', 'message')
2010-10-28 15:19:29,833 foo.bar ERROR This is another, ERROR, message
>>>

Вызовы ведения журнала (logger.debug(), logger.info() и т. д.) принимают позиционные параметры только для самого сообщения журнала; именованные параметры используются только для задания параметров обработки вызова (например, параметр exc_info указывает, что нужно записать сведения об обратной трассировке, а параметр extra — что в журнал нужно добавить дополнительную контекстную информацию). Поэтому нельзя напрямую вызывать функции ведения журнала с синтаксисом str.format() или string.Template, поскольку внутри пакет logging использует %-форматирование для объединения строки формата и переменных аргументов. Изменить это без нарушения обратной совместимости невозможно, поскольку существующий код повсеместно вызывает функции ведения журнала, используя строки формата с %-форматированием.

Однако есть способ использовать форматирование с {} и $ для формирования отдельных сообщений журнала. Вспомните, что в качестве строки формата сообщения можно использовать произвольный объект, а пакет logging вызовет у этого объекта метод str(), чтобы получить фактическую строку формата. Рассмотрим следующие два класса:

class BraceMessage:
    def __init__(self, fmt, /, *args, **kwargs):
        self.fmt = fmt
        self.args = args
        self.kwargs = kwargs

    def __str__(self):
        return self.fmt.format(*self.args, **self.kwargs)

class DollarMessage:
    def __init__(self, fmt, /, **kwargs):
        self.fmt = fmt
        self.kwargs = kwargs

    def __str__(self):
        from string import Template
        return Template(self.fmt).substitute(**self.kwargs)

Любой из них можно использовать вместо строки формата, чтобы с помощью форматирования {} или $ сформировать фактическую часть «сообщения», которая появится в отформатированном выводе журнала вместо «%(message)s», «{message}» или «$message». Каждый раз указывать имена классов при записи в журнал немного неудобно, но это вполне приемлемо, если использовать псевдоним, например __ (двойное подчёркивание; не путайте его с _, одиночным подчёркиванием, которое служит синонимом или псевдонимом для gettext.gettext() и связанных с ней функций).

Описанные выше классы не входят в состав Python, но их можно легко скопировать и вставить в собственный код. Их можно использовать следующим образом (предполагается, что они объявлены в модуле с именем wherever):

>>> from wherever import BraceMessage as __
>>> print(__('Message with {0} {name}', 2, name='placeholders'))
Message with 2 placeholders
>>> class Point: pass
...
>>> p = Point()
>>> p.x = 0.5
>>> p.y = 0.5
>>> print(__('Message with coordinates: ({point.x:.2f}, {point.y:.2f})',
...       point=p))
Message with coordinates: (0.50, 0.50)
>>> from wherever import DollarMessage as __
>>> print(__('Message with $num $what', num=2, what='placeholders'))
Message with 2 placeholders
>>>

В приведённых выше примерах для демонстрации работы форматирования используется print(), но для реальной записи в журнал таким способом, конечно же, следует использовать logger.debug() или аналогичный вызов.

Обратите внимание: этот подход не приводит к значительному снижению производительности. Фактическое форматирование происходит не при вызове функции ведения журнала, а только тогда, когда обработчик собирается вывести сообщение в журнал (если вообще собирается). Единственная непривычная особенность, которая может сбить с толку, — круглые скобки охватывают строку формата и аргументы, а не только строку формата. Это связано с тем, что нотация __ — всего лишь синтаксический сахар для вызова конструктора одного из классов XXXMessage.

При желании можно использовать LoggerAdapter, чтобы добиться аналогичного эффекта, как в следующем примере:

import logging

class Message:
    def __init__(self, fmt, args):
        self.fmt = fmt
        self.args = args

    def __str__(self):
        return self.fmt.format(*self.args)

class StyleAdapter(logging.LoggerAdapter):
    def log(self, level, msg, /, *args, stacklevel=1, **kwargs):
        if self.isEnabledFor(level):
            msg, kwargs = self.process(msg, kwargs)
            self.logger.log(level, Message(msg, args), **kwargs,
                            stacklevel=stacklevel+1)

logger = StyleAdapter(logging.getLogger(__name__))

def main():
    logger.debug('Hello, {}', 'world!')

if __name__ == '__main__':
    logging.basicConfig(level=logging.DEBUG)
    main()

При запуске в Python 3.8 или более поздней версии приведённый выше скрипт должен записать в журнал сообщение Hello, world!.

Настройка LogRecord

Каждое событие ведения журнала представлено экземпляром LogRecord. Когда событие записывается в журнал и не отфильтровывается уровнем регистратора, создаётся экземпляр LogRecord, заполняется информацией о событии, а затем передаётся обработчикам этого регистратора (и его предков — вплоть до регистратора, на котором отключено дальнейшее распространение по иерархии). До Python 3.2 этот объект создавался только в двух местах:

• Logger.makeRecord(), вызываемый в обычном процессе записи события в журнал. Он напрямую вызывал LogRecord для создания экземпляра.
• makeLogRecord(), которому передаётся словарь с атрибутами для добавления в LogRecord. Обычно он вызывается, когда получен подходящий словарь по сети (например, в виде pickle через SocketHandler или в формате JSON через HTTPHandler).

Обычно это означало, что при необходимости особым образом обработать LogRecord приходилось выбирать один из следующих способов.

• Создать собственный подкласс Logger, переопределить в нём Logger.makeRecord() и задать его с помощью setLoggerClass() до создания интересующих вас регистраторов.
• Добавить к регистратору или обработчику Filter, который выполняет необходимые специальные действия при вызове его метода filter().

Первый подход был бы несколько неудобен в ситуации, когда, например, несколько разных библиотек хотят выполнять разные действия. Каждая из них попыталась бы задать собственный подкласс Logger, и в итоге использовался бы последний заданный подкласс.

Второй подход хорошо подходит для многих случаев, но не позволяет, например, использовать специализированный подкласс LogRecord. Разработчики библиотек могут задать подходящий фильтр для своих регистраторов, но им придётся помнить об этом каждый раз, когда они добавляют новый регистратор (например, создавая новые пакеты или модули и выполняя

logger = logging.getLogger(__name__)

на уровне модуля). Вероятно, это слишком много дополнительных действий, о которых нужно помнить. Разработчики могут также добавить фильтр к NullHandler, прикреплённому к регистратору верхнего уровня, но он не будет вызван, если разработчик приложения прикрепит обработчик к регистратору библиотеки более низкого уровня. Поэтому вывод этого обработчика не будет учитывать намерения разработчика библиотеки.

Начиная с Python 3.2 создание LogRecord выполняется с помощью фабрики, которую можно задать самостоятельно. Фабрика — это вызываемый объект, который можно установить с помощью setLogRecordFactory() и получить с помощью getLogRecordFactory(). Фабрика вызывается с той же сигнатурой, что и конструктор LogRecord, поскольку LogRecord является фабрикой по умолчанию.

Этот подход позволяет пользовательской фабрике управлять всеми аспектами создания LogRecord. Например, можно возвращать подкласс или просто добавлять к созданной записи дополнительные атрибуты, используя следующий шаблон:

old_factory = logging.getLogRecordFactory()

def record_factory(*args, **kwargs):
    record = old_factory(*args, **kwargs)
    record.custom_attribute = 0xdecafbad
    return record

logging.setLogRecordFactory(record_factory)

Этот шаблон позволяет разным библиотекам объединять фабрики в цепочку. Если они не перезаписывают атрибуты друг друга и случайно не перезаписывают стандартные атрибуты, проблем возникнуть не должно. Однако следует учитывать, что каждое звено цепочки увеличивает накладные расходы во время выполнения для всех операций ведения журнала, поэтому этот подход следует использовать только в тех случаях, когда нужного результата нельзя добиться с помощью Filter.

Создание подклассов QueueHandler и QueueListener: пример с ZeroMQ

import zmq   # using pyzmq, the Python binding for ZeroMQ
import json  # for serializing records portably

ctx = zmq.Context()
sock = zmq.Socket(ctx, zmq.PUB)  # or zmq.PUSH, or other suitable value
sock.bind('tcp://*:5556')        # or wherever

class ZeroMQSocketHandler(QueueHandler):
    def enqueue(self, record):
        self.queue.send_json(record.__dict__)


handler = ZeroMQSocketHandler(sock)

class ZeroMQSocketHandler(QueueHandler):
    def __init__(self, uri, socktype=zmq.PUB, ctx=None):
        self.ctx = ctx or zmq.Context()
        socket = zmq.Socket(self.ctx, socktype)
        socket.bind(uri)
        super().__init__(socket)

    def enqueue(self, record):
        self.queue.send_json(record.__dict__)

    def close(self):
        self.queue.close()

class ZeroMQSocketListener(QueueListener):
    def __init__(self, uri, /, *handlers, **kwargs):
        self.ctx = kwargs.get('ctx') or zmq.Context()
        socket = zmq.Socket(self.ctx, zmq.SUB)
        socket.setsockopt_string(zmq.SUBSCRIBE, '')  # subscribe to everything
        socket.connect(uri)
        super().__init__(socket, *handlers, **kwargs)

    def dequeue(self):
        msg = self.queue.recv_json()
        return logging.makeLogRecord(msg)

Создание подклассов QueueHandler и QueueListener: пример с pynng

Как и в предыдущем разделе, можно реализовать приёмник и обработчик с помощью pynng — привязки Python к NNG, которую называют духовным преемником ZeroMQ. Следующие фрагменты кода иллюстрируют этот подход; их можно проверить в среде, где установлен pynng. Для разнообразия сначала покажем приёмник.

# listener.py
import json
import logging
import logging.handlers

import pynng

DEFAULT_ADDR = "tcp://localhost:13232"

interrupted = False

class NNGSocketListener(logging.handlers.QueueListener):

    def __init__(self, uri, /, *handlers, **kwargs):
        # Have a timeout for interruptability, and open a
        # subscriber socket
        socket = pynng.Sub0(listen=uri, recv_timeout=500)
        # The b'' subscription matches all topics
        topics = kwargs.pop('topics', None) or b''
        socket.subscribe(topics)
        # We treat the socket as a queue
        super().__init__(socket, *handlers, **kwargs)

    def dequeue(self, block):
        data = None
        # Keep looping while not interrupted and no data received over the
        # socket
        while not interrupted:
            try:
                data = self.queue.recv(block=block)
                break
            except pynng.Timeout:
                pass
            except pynng.Closed:  # sometimes happens when you hit Ctrl-C
                break
        if data is None:
            return None
        # Get the logging event sent from a publisher
        event = json.loads(data.decode('utf-8'))
        return logging.makeLogRecord(event)

    def enqueue_sentinel(self):
        # Not used in this implementation, as the socket isn't really a
        # queue
        pass

logging.getLogger('pynng').propagate = False
listener = NNGSocketListener(DEFAULT_ADDR, logging.StreamHandler(), topics=b'')
listener.start()
print('Press Ctrl-C to stop.')
try:
    while True:
        pass
except KeyboardInterrupt:
    interrupted = True
finally:
    listener.stop()

# sender.py
import json
import logging
import logging.handlers
import time
import random

import pynng

DEFAULT_ADDR = "tcp://localhost:13232"

class NNGSocketHandler(logging.handlers.QueueHandler):

    def __init__(self, uri):
        socket = pynng.Pub0(dial=uri, send_timeout=500)
        super().__init__(socket)

    def enqueue(self, record):
        # Send the record as UTF-8 encoded JSON
        d = dict(record.__dict__)
        data = json.dumps(d)
        self.queue.send(data.encode('utf-8'))

    def close(self):
        self.queue.close()

logging.getLogger('pynng').propagate = False
handler = NNGSocketHandler(DEFAULT_ADDR)
# Make sure the process ID is in the output
logging.basicConfig(level=logging.DEBUG,
                    handlers=[logging.StreamHandler(), handler],
                    format='%(levelname)-8s %(name)10s %(process)6s %(message)s')
levels = (logging.DEBUG, logging.INFO, logging.WARNING, logging.ERROR,
          logging.CRITICAL)
logger_names = ('myapp', 'myapp.lib1', 'myapp.lib2')
msgno = 1
while True:
    # Just randomly select some loggers and levels and log away
    level = random.choice(levels)
    logger = logging.getLogger(random.choice(logger_names))
    logger.log(level, 'Message no. %5d' % msgno)
    msgno += 1
    delay = random.random() * 2 + 0.5
    time.sleep(delay)

$ python sender.py
DEBUG         myapp    613 Message no.     1
WARNING  myapp.lib2    613 Message no.     2
CRITICAL myapp.lib2    613 Message no.     3
WARNING  myapp.lib2    613 Message no.     4
CRITICAL myapp.lib1    613 Message no.     5
DEBUG         myapp    613 Message no.     6
CRITICAL myapp.lib1    613 Message no.     7
INFO     myapp.lib1    613 Message no.     8
(and so on)

$ python sender.py
INFO     myapp.lib2    657 Message no.     1
CRITICAL myapp.lib2    657 Message no.     2
CRITICAL      myapp    657 Message no.     3
CRITICAL myapp.lib1    657 Message no.     4
INFO     myapp.lib1    657 Message no.     5
WARNING  myapp.lib2    657 Message no.     6
CRITICAL      myapp    657 Message no.     7
DEBUG    myapp.lib1    657 Message no.     8
(and so on)

$ python listener.py
Press Ctrl-C to stop.
DEBUG         myapp    613 Message no.     1
WARNING  myapp.lib2    613 Message no.     2
INFO     myapp.lib2    657 Message no.     1
CRITICAL myapp.lib2    613 Message no.     3
CRITICAL myapp.lib2    657 Message no.     2
CRITICAL      myapp    657 Message no.     3
WARNING  myapp.lib2    613 Message no.     4
CRITICAL myapp.lib1    613 Message no.     5
CRITICAL myapp.lib1    657 Message no.     4
INFO     myapp.lib1    657 Message no.     5
DEBUG         myapp    613 Message no.     6
WARNING  myapp.lib2    657 Message no.     6
CRITICAL      myapp    657 Message no.     7
CRITICAL myapp.lib1    613 Message no.     7
INFO     myapp.lib1    613 Message no.     8
DEBUG    myapp.lib1    657 Message no.     8
(and so on)

Пример конфигурации на основе словаря

Ниже приведён пример словаря конфигурации ведения журнала, взятый из документации проекта Django. Этот словарь передаётся в dictConfig(), чтобы применить конфигурацию:

LOGGING = {
    'version': 1,
    'disable_existing_loggers': False,
    'formatters': {
        'verbose': {
            'format': '{levelname} {asctime} {module} {process:d} {thread:d} {message}',
            'style': '{',
        },
        'simple': {
            'format': '{levelname} {message}',
            'style': '{',
        },
    },
    'filters': {
        'special': {
            '()': 'project.logging.SpecialFilter',
            'foo': 'bar',
        },
    },
    'handlers': {
        'console': {
            'level': 'INFO',
            'class': 'logging.StreamHandler',
            'formatter': 'simple',
        },
        'mail_admins': {
            'level': 'ERROR',
            'class': 'django.utils.log.AdminEmailHandler',
            'filters': ['special']
        }
    },
    'loggers': {
        'django': {
            'handlers': ['console'],
            'propagate': True,
        },
        'django.request': {
            'handlers': ['mail_admins'],
            'level': 'ERROR',
            'propagate': False,
        },
        'myproject.custom': {
            'handlers': ['console', 'mail_admins'],
            'level': 'INFO',
            'filters': ['special']
        }
    }
}

Дополнительную информацию об этой конфигурации можно найти в соответствующем разделе документации Django.

Настройка ротации журнала с помощью rotator и namer

Пример определения namer и rotator приведён в следующем исполняемом скрипте, который демонстрирует сжатие файла журнала с помощью gzip:

import gzip
import logging
import logging.handlers
import os
import shutil

def namer(name):
    return name + ".gz"

def rotator(source, dest):
    with open(source, 'rb') as f_in:
        with gzip.open(dest, 'wb') as f_out:
            shutil.copyfileobj(f_in, f_out)
    os.remove(source)


rh = logging.handlers.RotatingFileHandler('rotated.log', maxBytes=128, backupCount=5)
rh.rotator = rotator
rh.namer = namer

root = logging.getLogger()
root.setLevel(logging.INFO)
root.addHandler(rh)
f = logging.Formatter('%(asctime)s %(message)s')
rh.setFormatter(f)
for i in range(1000):
    root.info(f'Message no. {i + 1}')

После запуска будут созданы шесть новых файлов, пять из которых будут сжаты:

$ ls rotated.log*
rotated.log       rotated.log.2.gz  rotated.log.4.gz
rotated.log.1.gz  rotated.log.3.gz  rotated.log.5.gz
$ zcat rotated.log.1.gz
2023-01-20 02:28:17,767 Message no. 996
2023-01-20 02:28:17,767 Message no. 997
2023-01-20 02:28:17,767 Message no. 998

Более сложный пример многопроцессной обработки

В следующем рабочем примере показано, как использовать ведение журнала с многопроцессной обработкой и файлами конфигурации. Конфигурации довольно просты, но иллюстрируют, как можно реализовать более сложные варианты в реальном сценарии с несколькими процессами.

В примере главный процесс создаёт процесс-приёмник и несколько рабочих процессов. Для главного процесса, приёмника и рабочих процессов задаются три отдельные конфигурации (все рабочие процессы используют одну и ту же конфигурацию). Можно увидеть, как ведётся журнал в главном процессе, как рабочие процессы записывают сообщения через QueueHandler, а приёмник использует QueueListener и более сложную конфигурацию ведения журнала, направляя полученные через очередь события обработчикам, указанным в конфигурации. Обратите внимание, что эти конфигурации приведены только для иллюстрации, но вы сможете адаптировать пример под собственные задачи.

Вот скрипт; надеемся, что строки документации и комментарии объясняют принцип его работы:

import logging
import logging.config
import logging.handlers
from multiprocessing import Process, Queue, Event, current_process
import os
import random
import time

class MyHandler:
    """
    A simple handler for logging events. It runs in the listener process and
    dispatches events to loggers based on the name in the received record,
    which then get dispatched, by the logging system, to the handlers
    configured for those loggers.
    """

    def handle(self, record):
        if record.name == "root":
            logger = logging.getLogger()
        else:
            logger = logging.getLogger(record.name)

        if logger.isEnabledFor(record.levelno):
            # The process name is transformed just to show that it's the listener
            # doing the logging to files and console
            record.processName = '%s (for %s)' % (current_process().name, record.processName)
            logger.handle(record)

def listener_process(q, stop_event, config):
    """
    This could be done in the main process, but is just done in a separate
    process for illustrative purposes.

    This initialises logging according to the specified configuration,
    starts the listener and waits for the main process to signal completion
    via the event. The listener is then stopped, and the process exits.
    """
    logging.config.dictConfig(config)
    listener = logging.handlers.QueueListener(q, MyHandler())
    listener.start()
    if os.name == 'posix':
        # On POSIX, the setup logger will have been configured in the
        # parent process, but should have been disabled following the
        # dictConfig call.
        # On Windows, since fork isn't used, the setup logger won't
        # exist in the child, so it would be created and the message
        # would appear - hence the "if posix" clause.
        logger = logging.getLogger('setup')
        logger.critical('Should not appear, because of disabled logger ...')
    stop_event.wait()
    listener.stop()

def worker_process(config):
    """
    A number of these are spawned for the purpose of illustration. In
    practice, they could be a heterogeneous bunch of processes rather than
    ones which are identical to each other.

    This initialises logging according to the specified configuration,
    and logs a hundred messages with random levels to randomly selected
    loggers.

    A small sleep is added to allow other processes a chance to run. This
    is not strictly needed, but it mixes the output from the different
    processes a bit more than if it's left out.
    """
    logging.config.dictConfig(config)
    levels = [logging.DEBUG, logging.INFO, logging.WARNING, logging.ERROR,
              logging.CRITICAL]
    loggers = ['foo', 'foo.bar', 'foo.bar.baz',
               'spam', 'spam.ham', 'spam.ham.eggs']
    if os.name == 'posix':
        # On POSIX, the setup logger will have been configured in the
        # parent process, but should have been disabled following the
        # dictConfig call.
        # On Windows, since fork isn't used, the setup logger won't
        # exist in the child, so it would be created and the message
        # would appear - hence the "if posix" clause.
        logger = logging.getLogger('setup')
        logger.critical('Should not appear, because of disabled logger ...')
    for i in range(100):
        lvl = random.choice(levels)
        logger = logging.getLogger(random.choice(loggers))
        logger.log(lvl, 'Message no. %d', i)
        time.sleep(0.01)

def main():
    q = Queue()
    # The main process gets a simple configuration which prints to the console.
    config_initial = {
        'version': 1,
        'handlers': {
            'console': {
                'class': 'logging.StreamHandler',
                'level': 'INFO'
            }
        },
        'root': {
            'handlers': ['console'],
            'level': 'DEBUG'
        }
    }
    # The worker process configuration is just a QueueHandler attached to the
    # root logger, which allows all messages to be sent to the queue.
    # We disable existing loggers to disable the "setup" logger used in the
    # parent process. This is needed on POSIX because the logger will
    # be there in the child following a fork().
    config_worker = {
        'version': 1,
        'disable_existing_loggers': True,
        'handlers': {
            'queue': {
                'class': 'logging.handlers.QueueHandler',
                'queue': q
            }
        },
        'root': {
            'handlers': ['queue'],
            'level': 'DEBUG'
        }
    }
    # The listener process configuration shows that the full flexibility of
    # logging configuration is available to dispatch events to handlers however
    # you want.
    # We disable existing loggers to disable the "setup" logger used in the
    # parent process. This is needed on POSIX because the logger will
    # be there in the child following a fork().
    config_listener = {
        'version': 1,
        'disable_existing_loggers': True,
        'formatters': {
            'detailed': {
                'class': 'logging.Formatter',
                'format': '%(asctime)s %(name)-15s %(levelname)-8s %(processName)-10s %(message)s'
            },
            'simple': {
                'class': 'logging.Formatter',
                'format': '%(name)-15s %(levelname)-8s %(processName)-10s %(message)s'
            }
        },
        'handlers': {
            'console': {
                'class': 'logging.StreamHandler',
                'formatter': 'simple',
                'level': 'INFO'
            },
            'file': {
                'class': 'logging.FileHandler',
                'filename': 'mplog.log',
                'mode': 'w',
                'formatter': 'detailed'
            },
            'foofile': {
                'class': 'logging.FileHandler',
                'filename': 'mplog-foo.log',
                'mode': 'w',
                'formatter': 'detailed'
            },
            'errors': {
                'class': 'logging.FileHandler',
                'filename': 'mplog-errors.log',
                'mode': 'w',
                'formatter': 'detailed',
                'level': 'ERROR'
            }
        },
        'loggers': {
            'foo': {
                'handlers': ['foofile']
            }
        },
        'root': {
            'handlers': ['console', 'file', 'errors'],
            'level': 'DEBUG'
        }
    }
    # Log some initial events, just to show that logging in the parent works
    # normally.
    logging.config.dictConfig(config_initial)
    logger = logging.getLogger('setup')
    logger.info('About to create workers ...')
    workers = []
    for i in range(5):
        wp = Process(target=worker_process, name='worker %d' % (i + 1),
                     args=(config_worker,))
        workers.append(wp)
        wp.start()
        logger.info('Started worker: %s', wp.name)
    logger.info('About to create listener ...')
    stop_event = Event()
    lp = Process(target=listener_process, name='listener',
                 args=(q, stop_event, config_listener))
    lp.start()
    logger.info('Started listener')
    # We now hang around for the workers to finish their work.
    for wp in workers:
        wp.join()
    # Workers all done, listening can now stop.
    # Logging in the parent still works normally.
    logger.info('Telling listener to stop ...')
    stop_event.set()
    lp.join()
    logger.info('All done.')

if __name__ == '__main__':
    main()

Вставка BOM в сообщения, отправляемые обработчику SysLogHandler

RFC 5424 требует отправлять сообщение Unicode демону syslog в виде набора байтов следующей структуры: необязательный компонент, состоящий только из ASCII, за которым следует маркер порядка байтов UTF-8 (BOM), а затем данные Unicode, закодированные в UTF-8. (См. соответствующий раздел спецификации.)

В Python 3.1 в SysLogHandler добавили код для вставки BOM в сообщение, но, к сожалению, реализовали его неправильно: BOM оказывался в начале сообщения, не позволяя перед ним разместить компонент, состоящий только из ASCII.

Поскольку это поведение некорректно, ошибочный код вставки BOM удалён в Python 3.2.4 и более поздних версиях. Однако ему не предоставили замену. Поэтому, если вам нужно создавать соответствующие RFC 5424 сообщения с BOM, необязательной последовательностью символов ASCII перед ним и произвольными данными Unicode после него, закодированными в UTF-8, выполните следующие действия:

1. Добавьте экземпляр Formatter к экземпляру SysLogHandler, указав строку формата, например:

   'ASCII section\ufeffUnicode section'
   

   При кодировании в UTF-8 кодовая точка Unicode U+FEFF преобразуется в BOM UTF-8 — последовательность байтов b'\xef\xbb\xbf'.

2. Замените секцию ASCII любыми необходимыми заполнителями, но убедитесь, что после подстановки данные в ней всегда состоят только из символов ASCII (тогда после кодирования UTF-8 они останутся неизменными).
3. Замените секцию Unicode любыми необходимыми заполнителями; если после подстановки данные в ней содержат символы за пределами диапазона ASCII, это не проблема — они будут закодированы в UTF-8.

Отформатированное сообщение будет закодировано в UTF-8 с помощью SysLogHandler. Если следовать приведённым выше правилам, можно создавать сообщения, соответствующие RFC 5424. Если этого не сделать, ведение журнала может не выдать предупреждение, однако сообщения не будут соответствовать RFC 5424, и демон syslog может сообщить об ошибке.

Реализация структурированного ведения журнала

Хотя большинство сообщений журнала предназначены для чтения людьми и поэтому плохо поддаются машинному разбору, иногда может понадобиться выводить сообщения в структурированном формате, который можно разбирать программно (без сложных регулярных выражений для анализа сообщения журнала). Это несложно сделать с помощью пакета logging. Существует несколько способов реализации, но ниже приведён простой подход, в котором для сериализации события в машиночитаемом виде используется JSON:

import json
import logging

class StructuredMessage:
    def __init__(self, message, /, **kwargs):
        self.message = message
        self.kwargs = kwargs

    def __str__(self):
        return '%s >>> %s' % (self.message, json.dumps(self.kwargs))

_ = StructuredMessage   # optional, to improve readability

logging.basicConfig(level=logging.INFO, format='%(message)s')
logging.info(_('message 1', foo='bar', bar='baz', num=123, fnum=123.456))

При запуске приведённого выше скрипта выводится:

message 1 >>> {"fnum": 123.456, "num": 123, "bar": "baz", "foo": "bar"}

Обратите внимание, что порядок элементов может различаться в зависимости от используемой версии Python.

Если требуется более специализированная обработка, можно использовать пользовательский кодировщик JSON, как в следующем полном примере:

import json
import logging


class Encoder(json.JSONEncoder):
    def default(self, o):
        if isinstance(o, set):
            return tuple(o)
        elif isinstance(o, str):
            return o.encode('unicode_escape').decode('ascii')
        return super().default(o)

class StructuredMessage:
    def __init__(self, message, /, **kwargs):
        self.message = message
        self.kwargs = kwargs

    def __str__(self):
        s = Encoder().encode(self.kwargs)
        return '%s >>> %s' % (self.message, s)

_ = StructuredMessage   # optional, to improve readability

def main():
    logging.basicConfig(level=logging.INFO, format='%(message)s')
    logging.info(_('message 1', set_value={1, 2, 3}, snowman='\u2603'))

if __name__ == '__main__':
    main()

При запуске приведённого выше скрипта выводится:

message 1 >>> {"snowman": "\u2603", "set_value": [1, 2, 3]}

Обратите внимание, что порядок элементов может различаться в зависимости от используемой версии Python.

Настройка обработчиков с помощью dictConfig()

Иногда требуется особым образом настроить обработчики ведения журнала. Если использовать dictConfig(), это можно сделать без создания подклассов. Например, может понадобиться задать владельца файла журнала. В POSIX это легко сделать с помощью shutil.chown(), но файловые обработчики стандартной библиотеки не поддерживают это напрямую. Создать обработчик с нужными параметрами можно с помощью обычной функции, например:

def owned_file_handler(filename, mode='a', encoding=None, owner=None):
    if owner:
        if not os.path.exists(filename):
            open(filename, 'a').close()
        shutil.chown(filename, *owner)
    return logging.FileHandler(filename, mode, encoding)

Затем в конфигурации ведения журнала, передаваемой в dictConfig(), можно указать, что обработчик следует создать вызовом этой функции:

LOGGING = {
    'version': 1,
    'disable_existing_loggers': False,
    'formatters': {
        'default': {
            'format': '%(asctime)s %(levelname)s %(name)s %(message)s'
        },
    },
    'handlers': {
        'file':{
            # The values below are popped from this dictionary and
            # used to create the handler, set the handler's level and
            # its formatter.
            '()': owned_file_handler,
            'level':'DEBUG',
            'formatter': 'default',
            # The values below are passed to the handler creator callable
            # as keyword arguments.
            'owner': ['pulse', 'pulse'],
            'filename': 'chowntest.log',
            'mode': 'w',
            'encoding': 'utf-8',
        },
    },
    'root': {
        'handlers': ['file'],
        'level': 'DEBUG',
    },
}

В этом примере для наглядности владельцем назначаются пользователь и группа pulse. Объединим всё это в рабочий скрипт chowntest.py:

import logging, logging.config, os, shutil

def owned_file_handler(filename, mode='a', encoding=None, owner=None):
    if owner:
        if not os.path.exists(filename):
            open(filename, 'a').close()
        shutil.chown(filename, *owner)
    return logging.FileHandler(filename, mode, encoding)

LOGGING = {
    'version': 1,
    'disable_existing_loggers': False,
    'formatters': {
        'default': {
            'format': '%(asctime)s %(levelname)s %(name)s %(message)s'
        },
    },
    'handlers': {
        'file':{
            # The values below are popped from this dictionary and
            # used to create the handler, set the handler's level and
            # its formatter.
            '()': owned_file_handler,
            'level':'DEBUG',
            'formatter': 'default',
            # The values below are passed to the handler creator callable
            # as keyword arguments.
            'owner': ['pulse', 'pulse'],
            'filename': 'chowntest.log',
            'mode': 'w',
            'encoding': 'utf-8',
        },
    },
    'root': {
        'handlers': ['file'],
        'level': 'DEBUG',
    },
}

logging.config.dictConfig(LOGGING)
logger = logging.getLogger('mylogger')
logger.debug('A debug message')

Для запуска, вероятно, потребуются права пользователя root:

$ sudo python3.3 chowntest.py
$ cat chowntest.log
2013-11-05 09:34:51,128 DEBUG mylogger A debug message
$ ls -l chowntest.log
-rw-r--r-- 1 pulse pulse 55 2013-11-05 09:34 chowntest.log

Обратите внимание, что в этом примере используется Python 3.3, поскольку именно в этой версии появился shutil.chown(). Этот подход должен работать во всех версиях Python, поддерживающих dictConfig(), то есть в Python 2.7, 3.2 и более поздних версиях. В версиях до 3.3 фактическую смену владельца потребуется реализовать, например, с помощью os.chown().

На практике функция создания обработчика может находиться в служебном модуле вашего проекта. Вместо строки конфигурации:

'()': owned_file_handler,

можно, например, использовать:

'()': 'ext://project.util.owned_file_handler',

где project.util можно заменить фактическим именем пакета, в котором находится функция. В приведённом выше рабочем скрипте должно подойти 'ext://__main__.owned_file_handler'. В этом случае вызываемый объект фактически определяется функцией dictConfig() на основе спецификации ext://.

Надеемся, этот пример также подскажет, как аналогичным способом реализовать другие изменения файлов, например установить определённые биты разрешений POSIX с помощью os.chmod().

Разумеется, этот подход можно применять не только к FileHandler, но и к другим типам обработчиков: например, к одному из обработчиков с ротацией файлов или к обработчику совершенно другого типа.

Использование определённых стилей форматирования во всём приложении

В Python 3.2 в Formatter был добавлен именованный параметр style, который, по умолчанию принимая значение % для обратной совместимости, позволял задавать { или $, чтобы поддержать подходы к форматированию, используемые в str.format() и string.Template. Обратите внимание: этот параметр управляет форматированием сообщений журнала при выводе в журналы и совершенно не связан с тем, как создаётся отдельное сообщение журнала.

Вызовы журналирования (debug(), info() и т. д.) принимают только позиционные параметры для самого сообщения журнала; именованные параметры используются лишь для определения способа обработки вызова журналирования (например, параметр exc_info указывает, что следует записать информацию о трассировке, а параметр extra — что в журнал нужно добавить дополнительные контекстные сведения). Поэтому нельзя напрямую вызывать функции журналирования, используя синтаксис str.format() или string.Template, поскольку внутри пакет журналирования использует форматирование с %-оператором для объединения строки формата и переменных аргументов. Изменить это без нарушения обратной совместимости невозможно, так как во всём существующем коде вызовы журналирования используют строки формата с %-оператором.

Предлагалось связывать стили форматирования с конкретными регистраторами, но такой подход также приводит к проблемам с обратной совместимостью, поскольку в существующем коде может использоваться заданное имя регистратора и форматирование с %-оператором.

Чтобы журналирование работало совместно с любыми сторонними библиотеками и вашим кодом, решения о форматировании нужно принимать на уровне отдельного вызова журналирования. Это позволяет использовать несколько способов поддержки альтернативных стилей форматирования.

class BraceMessage:
    def __init__(self, fmt, /, *args, **kwargs):
        self.fmt = fmt
        self.args = args
        self.kwargs = kwargs

    def __str__(self):
        return self.fmt.format(*self.args, **self.kwargs)

class DollarMessage:
    def __init__(self, fmt, /, **kwargs):
        self.fmt = fmt
        self.kwargs = kwargs

    def __str__(self):
        from string import Template
        return Template(self.fmt).substitute(**self.kwargs)

>>> __ = BraceMessage
>>> print(__('Message with {0} {1}', 2, 'placeholders'))
Message with 2 placeholders
>>> class Point: pass
...
>>> p = Point()
>>> p.x = 0.5
>>> p.y = 0.5
>>> print(__('Message with coordinates: ({point.x:.2f}, {point.y:.2f})', point=p))
Message with coordinates: (0.50, 0.50)

>>> __ = DollarMessage
>>> print(__('Message with $num $what', num=2, what='placeholders'))
Message with 2 placeholders
>>>

Настройка фильтров с помощью dictConfig()

Фильтры можно настраивать с помощью dictConfig(), хотя с первого взгляда это может быть неочевидно (поэтому и появился этот пример). Поскольку Filter — единственный класс фильтров в стандартной библиотеке и вряд ли он подойдёт для многих задач (он существует лишь как базовый класс), обычно потребуется определить собственный подкласс Filter с переопределённым методом filter(). Для этого укажите ключ () в словаре конфигурации фильтра и задайте вызываемый объект, который будет создавать фильтр (очевиднее всего использовать класс, но можно предоставить любой вызываемый объект, возвращающий экземпляр Filter). Ниже приведён полный пример:

import logging
import logging.config
import sys

class MyFilter(logging.Filter):
    def __init__(self, param=None):
        self.param = param

    def filter(self, record):
        if self.param is None:
            allow = True
        else:
            allow = self.param not in record.msg
        if allow:
            record.msg = 'changed: ' + record.msg
        return allow

LOGGING = {
    'version': 1,
    'filters': {
        'myfilter': {
            '()': MyFilter,
            'param': 'noshow',
        }
    },
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
            'filters': ['myfilter']
        }
    },
    'root': {
        'level': 'DEBUG',
        'handlers': ['console']
    },
}

if __name__ == '__main__':
    logging.config.dictConfig(LOGGING)
    logging.debug('hello')
    logging.debug('hello - noshow')

В этом примере показано, как передать данные конфигурации вызываемому объекту, создающему экземпляр, в виде именованных параметров. При запуске приведённый выше сценарий выведет:

changed: hello

что показывает, что фильтр работает согласно настройкам.

Стоит отметить ещё пару моментов:

• Если на вызываемый объект нельзя сослаться напрямую в конфигурации (например, если он находится в другом модуле и его нельзя импортировать непосредственно там, где задан словарь конфигурации), можно использовать форму ext://..., описанную в разделе Доступ к внешним объектам. Например, в приведённом выше примере вместо MyFilter можно было бы использовать текст 'ext://__main__.MyFilter'.
• Этот способ подходит не только для фильтров, но и для настройки пользовательских обработчиков и форматировщиков. Дополнительную информацию о поддержке пользовательских объектов в конфигурации журналирования см. в разделе Объекты, определённые пользователем, а также в другом примере из руководства — Настройка обработчиков с помощью dictConfig() выше.

Пользовательское форматирование исключений

Иногда может понадобиться по-своему форматировать исключения. Скажем, вам нужно, чтобы на каждое зарегистрированное событие приходилась ровно одна строка, даже если присутствует информация об исключении. Это можно сделать с помощью пользовательского класса форматировщика, как показано в следующем примере:

import logging

class OneLineExceptionFormatter(logging.Formatter):
    def formatException(self, exc_info):
        """
        Format an exception so that it prints on a single line.
        """
        result = super().formatException(exc_info)
        return repr(result)  # or format into one line however you want to

    def format(self, record):
        s = super().format(record)
        if record.exc_text:
            s = s.replace('\n', '') + '|'
        return s

def configure_logging():
    fh = logging.FileHandler('output.txt', 'w')
    f = OneLineExceptionFormatter('%(asctime)s|%(levelname)s|%(message)s|',
                                  '%d/%m/%Y %H:%M:%S')
    fh.setFormatter(f)
    root = logging.getLogger()
    root.setLevel(logging.DEBUG)
    root.addHandler(fh)

def main():
    configure_logging()
    logging.info('Sample message')
    try:
        x = 1 / 0
    except ZeroDivisionError as e:
        logging.exception('ZeroDivisionError: %s', e)

if __name__ == '__main__':
    main()

При запуске будет создан файл ровно с двумя строками:

28/01/2015 07:21:23|INFO|Sample message|
28/01/2015 07:21:23|ERROR|ZeroDivisionError: integer division or modulo by zero|'Traceback (most recent call last):\n  File "logtest7.py", line 30, in main\n    x = 1 / 0\nZeroDivisionError: integer division or modulo by zero'|

Хотя приведённый выше способ и упрощён, он показывает, как форматировать информацию об исключениях по своему усмотрению. Для более специализированных задач может пригодиться модуль traceback.

Озвучивание сообщений журнала

Иногда сообщения журнала желательно не отображать, а озвучивать. Это легко сделать, если в вашей системе доступно преобразование текста в речь (TTS), даже если для него нет привязки к Python. В большинстве систем TTS есть программа командной строки, которую можно запустить; её можно вызвать из обработчика с помощью subprocess. Предполагается, что программы TTS, запускаемые из командной строки, не требуют взаимодействия с пользователями и не выполняются долго, а частота регистрируемых сообщений не настолько высока, чтобы завалить пользователя сообщениями. Также предполагается, что сообщения допустимо озвучивать по очереди, а не одновременно. Приведённая ниже реализация ждёт, пока будет произнесено одно сообщение, прежде чем обработать следующее, поэтому другие обработчики могут задерживаться. Вот короткий пример, демонстрирующий этот подход и предполагающий, что пакет TTS espeak доступен:

import logging
import subprocess
import sys

class TTSHandler(logging.Handler):
    def emit(self, record):
        msg = self.format(record)
        # Speak slowly in a female English voice
        cmd = ['espeak', '-s150', '-ven+f3', msg]
        p = subprocess.Popen(cmd, stdout=subprocess.PIPE,
                             stderr=subprocess.STDOUT)
        # wait for the program to finish
        p.communicate()

def configure_logging():
    h = TTSHandler()
    root = logging.getLogger()
    root.addHandler(h)
    # the default formatter just returns the message
    root.setLevel(logging.DEBUG)

def main():
    logging.info('Hello')
    logging.debug('Goodbye')

if __name__ == '__main__':
    configure_logging()
    sys.exit(main())

При запуске этот сценарий должен произнести «Hello», а затем «Goodbye» женским голосом.

Описанный подход, разумеется, можно адаптировать для других систем TTS и даже для совершенно иных систем, которые могут обрабатывать сообщения с помощью внешних программ, запускаемых из командной строки.

Буферизация сообщений журнала и их условный вывод

Иногда сообщения нужно записывать во временную область и выводить, только если произошло определённое событие. Например, можно начать записывать отладочные события в функции и не загромождать журнал собранной отладочной информацией, если функция завершилась без ошибок. Но если возникла ошибка, потребуется вывести вместе с ней и всю отладочную информацию.

Ниже приведён пример того, как это можно сделать с помощью декоратора для функций, журналирование в которых должно работать именно так. В нём используется logging.handlers.MemoryHandler, который позволяет буферизовать зарегистрированные события до наступления некоторого условия, после чего буферизованные события flushed — передаются другому обработчику (обработчику target) для обработки. По умолчанию MemoryHandler сбрасывает буфер, когда он заполняется или встречается событие с уровнем, большим или равным заданному пороговому значению. Если вам требуется собственное поведение при сбросе буфера, можно использовать этот пример с более специализированным подклассом MemoryHandler.

В примере сценария есть простая функция foo, которая перебирает все уровни журналирования, записывает в sys.stderr сообщение о том, на каком уровне она собирается вести журналирование, а затем действительно регистрирует сообщение на этом уровне. Функции foo можно передать параметр, который при истинном значении включает журналирование на уровнях ERROR и CRITICAL; в противном случае сообщения регистрируются только на уровнях DEBUG, INFO и WARNING.

Сценарий просто декорирует foo декоратором, выполняющим требуемое условное журналирование. Декоратор принимает регистратор в качестве параметра и на время вызова декорируемой функции подключает обработчик памяти. Дополнительно декоратор можно настроить, указав целевой обработчик, уровень, при котором должен происходить сброс буфера, и его вместимость (число буферизуемых записей). По умолчанию это соответственно StreamHandler, записывающий данные в sys.stderr, logging.ERROR и 100.

Вот сценарий:

import logging
from logging.handlers import MemoryHandler
import sys

logger = logging.getLogger(__name__)
logger.addHandler(logging.NullHandler())

def log_if_errors(logger, target_handler=None, flush_level=None, capacity=None):
    if target_handler is None:
        target_handler = logging.StreamHandler()
    if flush_level is None:
        flush_level = logging.ERROR
    if capacity is None:
        capacity = 100
    handler = MemoryHandler(capacity, flushLevel=flush_level, target=target_handler)

    def decorator(fn):
        def wrapper(*args, **kwargs):
            logger.addHandler(handler)
            try:
                return fn(*args, **kwargs)
            except Exception:
                logger.exception('call failed')
                raise
            finally:
                super(MemoryHandler, handler).flush()
                logger.removeHandler(handler)
        return wrapper

    return decorator

def write_line(s):
    sys.stderr.write('%s\n' % s)

def foo(fail=False):
    write_line('about to log at DEBUG ...')
    logger.debug('Actually logged at DEBUG')
    write_line('about to log at INFO ...')
    logger.info('Actually logged at INFO')
    write_line('about to log at WARNING ...')
    logger.warning('Actually logged at WARNING')
    if fail:
        write_line('about to log at ERROR ...')
        logger.error('Actually logged at ERROR')
        write_line('about to log at CRITICAL ...')
        logger.critical('Actually logged at CRITICAL')
    return fail

decorated_foo = log_if_errors(logger)(foo)

if __name__ == '__main__':
    logger.setLevel(logging.DEBUG)
    write_line('Calling undecorated foo with False')
    assert not foo(False)
    write_line('Calling undecorated foo with True')
    assert foo(True)
    write_line('Calling decorated foo with False')
    assert not decorated_foo(False)
    write_line('Calling decorated foo with True')
    assert decorated_foo(True)

При запуске сценария должен появиться следующий вывод:

Calling undecorated foo with False
about to log at DEBUG ...
about to log at INFO ...
about to log at WARNING ...
Calling undecorated foo with True
about to log at DEBUG ...
about to log at INFO ...
about to log at WARNING ...
about to log at ERROR ...
about to log at CRITICAL ...
Calling decorated foo with False
about to log at DEBUG ...
about to log at INFO ...
about to log at WARNING ...
Calling decorated foo with True
about to log at DEBUG ...
about to log at INFO ...
about to log at WARNING ...
about to log at ERROR ...
Actually logged at DEBUG
Actually logged at INFO
Actually logged at WARNING
Actually logged at ERROR
about to log at CRITICAL ...
Actually logged at CRITICAL

Как видите, фактический вывод в журнал происходит только при регистрации события с уровнем ERROR или выше, но в этом случае также регистрируются все предшествующие события с более низким уровнем.

Разумеется, можно использовать и обычный способ декорирования:

@log_if_errors(logger)
def foo(fail=False):
    ...

Отправка сообщений журнала по электронной почте с буферизацией

Чтобы показать, как отправлять сообщения журнала по электронной почте так, чтобы в каждом письме было заданное количество сообщений, можно создать подкласс BufferingHandler. В следующем примере, который можно адаптировать под свои задачи, приведён простой тестовый сценарий. Он позволяет запускать скрипт с аргументами командной строки, задающими типичные параметры для отправки сообщений через SMTP. (Запустите загруженный сценарий с аргументом -h, чтобы увидеть обязательные и необязательные аргументы.)

import logging
import logging.handlers
import smtplib

class BufferingSMTPHandler(logging.handlers.BufferingHandler):
    def __init__(self, mailhost, port, username, password, fromaddr, toaddrs,
                 subject, capacity):
        logging.handlers.BufferingHandler.__init__(self, capacity)
        self.mailhost = mailhost
        self.mailport = port
        self.username = username
        self.password = password
        self.fromaddr = fromaddr
        if isinstance(toaddrs, str):
            toaddrs = [toaddrs]
        self.toaddrs = toaddrs
        self.subject = subject
        self.setFormatter(logging.Formatter("%(asctime)s %(levelname)-5s %(message)s"))

    def flush(self):
        if len(self.buffer) > 0:
            try:
                smtp = smtplib.SMTP(self.mailhost, self.mailport)
                smtp.starttls()
                smtp.login(self.username, self.password)
                msg = "From: %s\r\nTo: %s\r\nSubject: %s\r\n\r\n" % (self.fromaddr, ','.join(self.toaddrs), self.subject)
                for record in self.buffer:
                    s = self.format(record)
                    msg = msg + s + "\r\n"
                smtp.sendmail(self.fromaddr, self.toaddrs, msg)
                smtp.quit()
            except Exception:
                if logging.raiseExceptions:
                    raise
            self.buffer = []

if __name__ == '__main__':
    import argparse

    ap = argparse.ArgumentParser()
    aa = ap.add_argument
    aa('host', metavar='HOST', help='SMTP server')
    aa('--port', '-p', type=int, default=587, help='SMTP port')
    aa('user', metavar='USER', help='SMTP username')
    aa('password', metavar='PASSWORD', help='SMTP password')
    aa('to', metavar='TO', help='Addressee for emails')
    aa('sender', metavar='SENDER', help='Sender email address')
    aa('--subject', '-s',
       default='Test Logging email from Python logging module (buffering)',
       help='Subject of email')
    options = ap.parse_args()
    logger = logging.getLogger()
    logger.setLevel(logging.DEBUG)
    h = BufferingSMTPHandler(options.host, options.port, options.user,
                             options.password, options.sender,
                             options.to, options.subject, 10)
    logger.addHandler(h)
    for i in range(102):
        logger.info("Info index = %d", i)
    h.flush()
    h.close()

Если запустить этот сценарий и ваш SMTP-сервер настроен правильно, он отправит на указанный вами адрес одиннадцать писем. В каждом из первых десяти писем будет по десять сообщений журнала, а в одиннадцатом — два сообщения. Всего получится 102 сообщения, как указано в сценарии.

Форматирование времени в UTC (GMT) с помощью конфигурации

Иногда требуется форматировать время в UTC. Для этого можно использовать класс, например UTCFormatter, показанный ниже:

import logging
import time

class UTCFormatter(logging.Formatter):
    converter = time.gmtime

Затем в коде можно использовать UTCFormatter вместо Formatter. Если требуется сделать это с помощью конфигурации, можно воспользоваться API dictConfig(), как показано в следующем полном примере:

import logging
import logging.config
import time

class UTCFormatter(logging.Formatter):
    converter = time.gmtime

LOGGING = {
    'version': 1,
    'disable_existing_loggers': False,
    'formatters': {
        'utc': {
            '()': UTCFormatter,
            'format': '%(asctime)s %(message)s',
        },
        'local': {
            'format': '%(asctime)s %(message)s',
        }
    },
    'handlers': {
        'console1': {
            'class': 'logging.StreamHandler',
            'formatter': 'utc',
        },
        'console2': {
            'class': 'logging.StreamHandler',
            'formatter': 'local',
        },
    },
    'root': {
        'handlers': ['console1', 'console2'],
   }
}

if __name__ == '__main__':
    logging.config.dictConfig(LOGGING)
    logging.warning('The local time is %s', time.asctime())

При запуске этого сценария он должен вывести что-то вроде:

2015-10-17 12:53:29,501 The local time is Sat Oct 17 13:53:29 2015
2015-10-17 13:53:29,501 The local time is Sat Oct 17 13:53:29 2015

что демонстрирует форматирование времени как для местного времени, так и для UTC — по одному варианту для каждого обработчика.

Использование менеджера контекста для выборочного журналирования

Иногда бывает полезно временно изменить конфигурацию журналирования, а после выполнения некоторого действия восстановить её. Менеджер контекста — наиболее очевидный способ сохранить и восстановить состояние журналирования. Ниже приведён простой пример такого менеджера контекста, позволяющего при необходимости изменить уровень журналирования и добавить обработчик только в пределах области действия менеджера контекста:

import logging
import sys

class LoggingContext:
    def __init__(self, logger, level=None, handler=None, close=True):
        self.logger = logger
        self.level = level
        self.handler = handler
        self.close = close

    def __enter__(self):
        if self.level is not None:
            self.old_level = self.logger.level
            self.logger.setLevel(self.level)
        if self.handler:
            self.logger.addHandler(self.handler)

    def __exit__(self, et, ev, tb):
        if self.level is not None:
            self.logger.setLevel(self.old_level)
        if self.handler:
            self.logger.removeHandler(self.handler)
        if self.handler and self.close:
            self.handler.close()
        # implicit return of None => don't swallow exceptions

Если указать значение уровня, на время блока with, охваченного менеджером контекста, для регистратора устанавливается этот уровень. Если указать обработчик, при входе в блок он добавляется к регистратору, а при выходе удаляется. Также можно попросить менеджер закрыть обработчик при выходе из блока — это пригодится, если он больше не нужен.

Чтобы показать, как это работает, добавим к приведённому выше коду следующий блок:

if __name__ == '__main__':
    logger = logging.getLogger('foo')
    logger.addHandler(logging.StreamHandler())
    logger.setLevel(logging.INFO)
    logger.info('1. This should appear just once on stderr.')
    logger.debug('2. This should not appear.')
    with LoggingContext(logger, level=logging.DEBUG):
        logger.debug('3. This should appear once on stderr.')
    logger.debug('4. This should not appear.')
    h = logging.StreamHandler(sys.stdout)
    with LoggingContext(logger, level=logging.DEBUG, handler=h, close=True):
        logger.debug('5. This should appear twice - once on stderr and once on stdout.')
    logger.info('6. This should appear just once on stderr.')
    logger.debug('7. This should not appear.')

Изначально мы устанавливаем для регистратора уровень INFO, поэтому сообщение № 1 появляется, а сообщение № 2 — нет. Затем в следующем блоке with мы временно меняем уровень на DEBUG, поэтому появляется сообщение № 3. После выхода из блока уровень регистратора восстанавливается до INFO, поэтому сообщение № 4 не появляется. В следующем блоке with мы снова устанавливаем уровень DEBUG, а также добавляем обработчик, записывающий данные в sys.stdout. Поэтому сообщение № 5 появляется в консоли дважды: один раз через stderr и второй раз через stdout. После завершения инструкции with состояние восстанавливается, поэтому сообщение № 6 появляется (как сообщение № 1), а сообщение № 7 — нет (как сообщение № 2).

Если запустить получившийся сценарий, результат будет таким:

$ python logctx.py
1. This should appear just once on stderr.
3. This should appear once on stderr.
5. This should appear twice - once on stderr and once on stdout.
5. This should appear twice - once on stderr and once on stdout.
6. This should appear just once on stderr.

Если запустить его снова, но передать stderr в /dev/null, мы увидим следующее — это единственное сообщение, записанное в stdout:

$ python logctx.py 2>/dev/null
5. This should appear twice - once on stderr and once on stdout.

Если повторить запуск, передав stdout в /dev/null, получим:

$ python logctx.py >/dev/null
1. This should appear just once on stderr.
3. This should appear once on stderr.
5. This should appear twice - once on stderr and once on stdout.
6. This should appear just once on stderr.

В этом случае сообщение № 5, выведенное в stdout, не появляется, как и ожидалось.

Разумеется, описанный здесь подход можно обобщить, например, чтобы временно подключать фильтры журналирования. Обратите внимание, что приведённый выше код работает как в Python 2, так и в Python 3.

Стартовый шаблон приложения CLI

Ниже приведён пример, демонстрирующий, как можно:

• Использовать уровень журналирования, заданный аргументами командной строки
• Направлять выполнение в разные подкоманды, расположенные в отдельных файлах, чтобы все они вели журналирование на одном уровне единообразным способом
• Использовать простую минимальную конфигурацию

Предположим, у нас есть приложение командной строки, предназначенное для остановки, запуска или перезапуска служб. Для наглядности его можно организовать так: файл app.py — главный сценарий приложения, а отдельные команды реализованы в файлах start.py, stop.py и restart.py. Предположим также, что мы хотим управлять подробностью вывода приложения с помощью аргумента командной строки, по умолчанию задающего уровень logging.INFO. Вот один из способов написания app.py:

import argparse
import importlib
import logging
import os
import sys

def main(args=None):
    scriptname = os.path.basename(__file__)
    parser = argparse.ArgumentParser(scriptname)
    levels = ('DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL')
    parser.add_argument('--log-level', default='INFO', choices=levels)
    subparsers = parser.add_subparsers(dest='command',
                                       help='Available commands:')
    start_cmd = subparsers.add_parser('start', help='Start a service')
    start_cmd.add_argument('name', metavar='NAME',
                           help='Name of service to start')
    stop_cmd = subparsers.add_parser('stop',
                                     help='Stop one or more services')
    stop_cmd.add_argument('names', metavar='NAME', nargs='+',
                          help='Name of service to stop')
    restart_cmd = subparsers.add_parser('restart',
                                        help='Restart one or more services')
    restart_cmd.add_argument('names', metavar='NAME', nargs='+',
                             help='Name of service to restart')
    options = parser.parse_args()
    # the code to dispatch commands could all be in this file. For the purposes
    # of illustration only, we implement each command in a separate module.
    try:
        mod = importlib.import_module(options.command)
        cmd = getattr(mod, 'command')
    except (ImportError, AttributeError):
        print('Unable to find the code for command \'%s\'' % options.command)
        return 1
    # Could get fancy here and load configuration from file or dictionary
    logging.basicConfig(level=options.log_level,
                        format='%(levelname)s %(name)s %(message)s')
    cmd(options)

if __name__ == '__main__':
    sys.exit(main())

Команды start, stop и restart можно реализовать в отдельных модулях. Например, команда запуска может выглядеть так:

# start.py
import logging

logger = logging.getLogger(__name__)

def command(options):
    logger.debug('About to start %s', options.name)
    # actually do the command processing here ...
    logger.info('Started the \'%s\' service.', options.name)

А команда остановки — так:

# stop.py
import logging

logger = logging.getLogger(__name__)

def command(options):
    n = len(options.names)
    if n == 1:
        plural = ''
        services = '\'%s\'' % options.names[0]
    else:
        plural = 's'
        services = ', '.join('\'%s\'' % name for name in options.names)
        i = services.rfind(', ')
        services = services[:i] + ' and ' + services[i + 2:]
    logger.debug('About to stop %s', services)
    # actually do the command processing here ...
    logger.info('Stopped the %s service%s.', services, plural)

Аналогично можно реализовать команду перезапуска:

# restart.py
import logging

logger = logging.getLogger(__name__)

def command(options):
    n = len(options.names)
    if n == 1:
        plural = ''
        services = '\'%s\'' % options.names[0]
    else:
        plural = 's'
        services = ', '.join('\'%s\'' % name for name in options.names)
        i = services.rfind(', ')
        services = services[:i] + ' and ' + services[i + 2:]
    logger.debug('About to restart %s', services)
    # actually do the command processing here ...
    logger.info('Restarted the %s service%s.', services, plural)

Если запустить это приложение с уровнем журналирования по умолчанию, вывод будет примерно таким:

$ python app.py start foo
INFO start Started the 'foo' service.

$ python app.py stop foo bar
INFO stop Stopped the 'foo' and 'bar' services.

$ python app.py restart foo bar baz
INFO restart Restarted the 'foo', 'bar' and 'baz' services.

Первое слово — это уровень журналирования, а второе — имя модуля или пакета, в котором было зарегистрировано событие.

Изменив уровень журналирования, можно изменить объём информации, отправляемой в журнал. Например, если требуется больше информации:

$ python app.py --log-level DEBUG start foo
DEBUG start About to start foo
INFO start Started the 'foo' service.

$ python app.py --log-level DEBUG stop foo bar
DEBUG stop About to stop 'foo' and 'bar'
INFO stop Stopped the 'foo' and 'bar' services.

$ python app.py --log-level DEBUG restart foo bar baz
DEBUG restart About to restart 'foo', 'bar' and 'baz'
INFO restart Restarted the 'foo', 'bar' and 'baz' services.

А если требуется меньше:

$ python app.py --log-level WARNING start foo
$ python app.py --log-level WARNING stop foo bar
$ python app.py --log-level WARNING restart foo bar baz

В этом случае команды ничего не выводят в консоль, поскольку они не регистрируют сообщения уровня WARNING или выше.

Графический интерфейс Qt для журналирования

Время от времени возникает вопрос о том, как вести журналирование в приложении с графическим интерфейсом. Фреймворк Qt — популярный кроссплатформенный фреймворк для пользовательских интерфейсов с привязками к Python в библиотеках PySide2 и PyQt5.

В следующем примере показано, как вести журналирование в графический интерфейс Qt. В нём представлен простой класс QtHandler, принимающий вызываемый объект — слот в главном потоке, который выполняет обновление графического интерфейса. Кроме того, создаётся рабочий поток, чтобы показать, как вести журналирование в интерфейсе как из самого пользовательского интерфейса (с помощью кнопки для ручного журналирования), так и из рабочего потока, выполняющего фоновую работу (в данном случае — регистрирующего сообщения случайных уровней с короткими случайными задержками между ними).

Рабочий поток реализован с помощью класса Qt QThread, а не модуля threading, поскольку в некоторых случаях необходимо использовать QThread, обеспечивающий более тесную интеграцию с другими компонентами Qt.

Код должен работать с недавними выпусками любого из вариантов: PySide6, PyQt6, PySide2 или PyQt5. Подход можно адаптировать и для более ранних версий Qt. Дополнительную информацию см. в комментариях к фрагменту кода.

import datetime
import logging
import random
import sys
import time

# Deal with minor differences between different Qt packages
try:
    from PySide6 import QtCore, QtGui, QtWidgets
    Signal = QtCore.Signal
    Slot = QtCore.Slot
except ImportError:
    try:
        from PyQt6 import QtCore, QtGui, QtWidgets
        Signal = QtCore.pyqtSignal
        Slot = QtCore.pyqtSlot
    except ImportError:
        try:
            from PySide2 import QtCore, QtGui, QtWidgets
            Signal = QtCore.Signal
            Slot = QtCore.Slot
        except ImportError:
            from PyQt5 import QtCore, QtGui, QtWidgets
            Signal = QtCore.pyqtSignal
            Slot = QtCore.pyqtSlot

logger = logging.getLogger(__name__)


#
# Signals need to be contained in a QObject or subclass in order to be correctly
# initialized.
#
class Signaller(QtCore.QObject):
    signal = Signal(str, logging.LogRecord)

#
# Output to a Qt GUI is only supposed to happen on the main thread. So, this
# handler is designed to take a slot function which is set up to run in the main
# thread. In this example, the function takes a string argument which is a
# formatted log message, and the log record which generated it. The formatted
# string is just a convenience - you could format a string for output any way
# you like in the slot function itself.
#
# You specify the slot function to do whatever GUI updates you want. The handler
# doesn't know or care about specific UI elements.
#
class QtHandler(logging.Handler):
    def __init__(self, slotfunc, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.signaller = Signaller()
        self.signaller.signal.connect(slotfunc)

    def emit(self, record):
        s = self.format(record)
        self.signaller.signal.emit(s, record)

#
# This example uses QThreads, which means that the threads at the Python level
# are named something like "Dummy-1". The function below gets the Qt name of the
# current thread.
#
def ctname():
    return QtCore.QThread.currentThread().objectName()


#
# Used to generate random levels for logging.
#
LEVELS = (logging.DEBUG, logging.INFO, logging.WARNING, logging.ERROR,
          logging.CRITICAL)

#
# This worker class represents work that is done in a thread separate to the
# main thread. The way the thread is kicked off to do work is via a button press
# that connects to a slot in the worker.
#
# Because the default threadName value in the LogRecord isn't much use, we add
# a qThreadName which contains the QThread name as computed above, and pass that
# value in an "extra" dictionary which is used to update the LogRecord with the
# QThread name.
#
# This example worker just outputs messages sequentially, interspersed with
# random delays of the order of a few seconds.
#
class Worker(QtCore.QObject):
    @Slot()
    def start(self):
        extra = {'qThreadName': ctname() }
        logger.debug('Started work', extra=extra)
        i = 1
        # Let the thread run until interrupted. This allows reasonably clean
        # thread termination.
        while not QtCore.QThread.currentThread().isInterruptionRequested():
            delay = 0.5 + random.random() * 2
            time.sleep(delay)
            try:
                if random.random() < 0.1:
                    raise ValueError('Exception raised: %d' % i)
                else:
                    level = random.choice(LEVELS)
                    logger.log(level, 'Message after delay of %3.1f: %d', delay, i, extra=extra)
            except ValueError as e:
                logger.exception('Failed: %s', e, extra=extra)
            i += 1

#
# Implement a simple UI for this cookbook example. This contains:
#
# * A read-only text edit window which holds formatted log messages
# * A button to start work and log stuff in a separate thread
# * A button to log something from the main thread
# * A button to clear the log window
#
class Window(QtWidgets.QWidget):

    COLORS = {
        logging.DEBUG: 'black',
        logging.INFO: 'blue',
        logging.WARNING: 'orange',
        logging.ERROR: 'red',
        logging.CRITICAL: 'purple',
    }

    def __init__(self, app):
        super().__init__()
        self.app = app
        self.textedit = te = QtWidgets.QPlainTextEdit(self)
        # Set whatever the default monospace font is for the platform
        f = QtGui.QFont('nosuchfont')
        if hasattr(f, 'Monospace'):
            f.setStyleHint(f.Monospace)
        else:
            f.setStyleHint(f.StyleHint.Monospace)  # for Qt6
        te.setFont(f)
        te.setReadOnly(True)
        PB = QtWidgets.QPushButton
        self.work_button = PB('Start background work', self)
        self.log_button = PB('Log a message at a random level', self)
        self.clear_button = PB('Clear log window', self)
        self.handler = h = QtHandler(self.update_status)
        # Remember to use qThreadName rather than threadName in the format string.
        fs = '%(asctime)s %(qThreadName)-12s %(levelname)-8s %(message)s'
        formatter = logging.Formatter(fs)
        h.setFormatter(formatter)
        logger.addHandler(h)
        # Set up to terminate the QThread when we exit
        app.aboutToQuit.connect(self.force_quit)

        # Lay out all the widgets
        layout = QtWidgets.QVBoxLayout(self)
        layout.addWidget(te)
        layout.addWidget(self.work_button)
        layout.addWidget(self.log_button)
        layout.addWidget(self.clear_button)
        self.setFixedSize(900, 400)

        # Connect the non-worker slots and signals
        self.log_button.clicked.connect(self.manual_update)
        self.clear_button.clicked.connect(self.clear_display)

        # Start a new worker thread and connect the slots for the worker
        self.start_thread()
        self.work_button.clicked.connect(self.worker.start)
        # Once started, the button should be disabled
        self.work_button.clicked.connect(lambda : self.work_button.setEnabled(False))

    def start_thread(self):
        self.worker = Worker()
        self.worker_thread = QtCore.QThread()
        self.worker.setObjectName('Worker')
        self.worker_thread.setObjectName('WorkerThread')  # for qThreadName
        self.worker.moveToThread(self.worker_thread)
        # This will start an event loop in the worker thread
        self.worker_thread.start()

    def kill_thread(self):
        # Just tell the worker to stop, then tell it to quit and wait for that
        # to happen
        self.worker_thread.requestInterruption()
        if self.worker_thread.isRunning():
            self.worker_thread.quit()
            self.worker_thread.wait()
        else:
            print('worker has already exited.')

    def force_quit(self):
        # For use when the window is closed
        if self.worker_thread.isRunning():
            self.kill_thread()

    # The functions below update the UI and run in the main thread because
    # that's where the slots are set up

    @Slot(str, logging.LogRecord)
    def update_status(self, status, record):
        color = self.COLORS.get(record.levelno, 'black')
        s = '<pre><font color="%s">%s</font></pre>' % (color, status)
        self.textedit.appendHtml(s)

    @Slot()
    def manual_update(self):
        # This function uses the formatted message passed in, but also uses
        # information from the record to format the message in an appropriate
        # color according to its severity (level).
        level = random.choice(LEVELS)
        extra = {'qThreadName': ctname() }
        logger.log(level, 'Manually logged!', extra=extra)

    @Slot()
    def clear_display(self):
        self.textedit.clear()


def main():
    QtCore.QThread.currentThread().setObjectName('MainThread')
    logging.getLogger().setLevel(logging.DEBUG)
    app = QtWidgets.QApplication(sys.argv)
    example = Window(app)
    example.show()
    if hasattr(app, 'exec'):
        rc = app.exec()
    else:
        rc = app.exec_()
    sys.exit(rc)

if __name__=='__main__':
    main()

Журналирование в syslog с поддержкой RFC 5424

Хотя RFC 5424 датирован 2009 годом, большинство серверов syslog по умолчанию настроены на использование более старого RFC 3164, появившегося в 2001 году. Когда в 2003 году в Python был добавлен logging, он поддерживал более ранний (и единственный существовавший на тот момент) протокол. После появления RFC 5424 широкого внедрения этого протокола на серверах syslog не произошло, поэтому функциональность SysLogHandler не обновлялась.

В RFC 5424 есть полезные возможности, например поддержка структурированных данных. Если вам нужно вести журналирование на сервер syslog с поддержкой этого стандарта, можно воспользоваться обработчиком-подклассом примерно такого вида:

import datetime
import logging.handlers
import re
import socket
import time

class SysLogHandler5424(logging.handlers.SysLogHandler):

    tz_offset = re.compile(r'([+-]\d{2})(\d{2})$')
    escaped = re.compile(r'([\]"\\])')

    def __init__(self, *args, **kwargs):
        self.msgid = kwargs.pop('msgid', None)
        self.appname = kwargs.pop('appname', None)
        super().__init__(*args, **kwargs)

    def format(self, record):
        version = 1
        asctime = datetime.datetime.fromtimestamp(record.created).isoformat()
        m = self.tz_offset.match(time.strftime('%z'))
        has_offset = False
        if m and time.timezone:
            hrs, mins = m.groups()
            if int(hrs) or int(mins):
                has_offset = True
        if not has_offset:
            asctime += 'Z'
        else:
            asctime += f'{hrs}:{mins}'
        try:
            hostname = socket.gethostname()
        except Exception:
            hostname = '-'
        appname = self.appname or '-'
        procid = record.process
        msgid = '-'
        msg = super().format(record)
        sdata = '-'
        if hasattr(record, 'structured_data'):
            sd = record.structured_data
            # This should be a dict where the keys are SD-ID and the value is a
            # dict mapping PARAM-NAME to PARAM-VALUE (refer to the RFC for what these
            # mean)
            # There's no error checking here - it's purely for illustration, and you
            # can adapt this code for use in production environments
            parts = []

            def replacer(m):
                g = m.groups()
                return '\\' + g[0]

            for sdid, dv in sd.items():
                part = f'[{sdid}'
                for k, v in dv.items():
                    s = str(v)
                    s = self.escaped.sub(replacer, s)
                    part += f' {k}="{s}"'
                part += ']'
                parts.append(part)
            sdata = ''.join(parts)
        return f'{version} {asctime} {hostname} {appname} {procid} {msgid} {sdata} {msg}'

Чтобы полностью понять приведённый выше код, нужно знать RFC 5424. Кроме того, ваши задачи могут несколько отличаться (например, может потребоваться другой способ передачи структурированных данных в журнал). Тем не менее приведённый выше код можно адаптировать под ваши конкретные нужды. С этим обработчиком структурированные данные можно передавать, например, так:

sd = {
    'foo@12345': {'bar': 'baz', 'baz': 'bozz', 'fizz': r'buzz'},
    'foo@54321': {'rab': 'baz', 'zab': 'bozz', 'zzif': r'buzz'}
}
extra = {'structured_data': sd}
i = 1
logger.debug('Message %d', i, extra=extra)

Как использовать регистратор в качестве выходного потока

Иногда требуется взаимодействовать со сторонним API, который ожидает файловоподобный объект для записи данных, но вывод API нужно направить в регистратор. Это можно сделать с помощью класса, который оборачивает регистратор и предоставляет файловый интерфейс. Ниже приведён короткий сценарий, демонстрирующий такой класс:

import logging

class LoggerWriter:
    def __init__(self, logger, level):
        self.logger = logger
        self.level = level

    def write(self, message):
        if message != '\n':  # avoid printing bare newlines, if you like
            self.logger.log(self.level, message)

    def flush(self):
        # doesn't actually do anything, but might be expected of a file-like
        # object - so optional depending on your situation
        pass

    def close(self):
        # doesn't actually do anything, but might be expected of a file-like
        # object - so optional depending on your situation. You might want
        # to set a flag so that later calls to write raise an exception
        pass

def main():
    logging.basicConfig(level=logging.DEBUG)
    logger = logging.getLogger('demo')
    info_fp = LoggerWriter(logger, logging.INFO)
    debug_fp = LoggerWriter(logger, logging.DEBUG)
    print('An INFO message', file=info_fp)
    print('A DEBUG message', file=debug_fp)

if __name__ == "__main__":
    main()

При запуске этого сценария он выводит:

INFO:demo:An INFO message
DEBUG:demo:A DEBUG message

Также можно использовать LoggerWriter для перенаправления sys.stdout и sys.stderr примерно так:

import sys

sys.stdout = LoggerWriter(logger, logging.INFO)
sys.stderr = LoggerWriter(logger, logging.WARNING)

Это следует делать после настройки журналирования. В приведённом выше примере это выполняет вызов basicConfig() (используя значение sys.stderr до того, как оно будет переопределено экземпляром LoggerWriter). Тогда результат будет выглядеть примерно так:

>>> print('Foo')
INFO:demo:Foo
>>> print('Bar', file=sys.stderr)
WARNING:demo:Bar
>>>

Разумеется, в приведённых выше примерах показан вывод в формате, используемом функцией basicConfig(), но при настройке журналирования можно использовать другой форматировщик.

Обратите внимание: при использовании описанного подхода многое зависит от буферизации и последовательности перехватываемых вызовов записи. Например, при приведённом выше определении LoggerWriter, если использовать следующий фрагмент:

sys.stderr = LoggerWriter(logger, logging.WARNING)
1 / 0

то при запуске сценария получится такой результат:

WARNING:demo:Traceback (most recent call last):

WARNING:demo:  File "/home/runner/cookbook-loggerwriter/test.py", line 53, in <module>

WARNING:demo:
WARNING:demo:main()
WARNING:demo:  File "/home/runner/cookbook-loggerwriter/test.py", line 49, in main

WARNING:demo:
WARNING:demo:1 / 0
WARNING:demo:ZeroDivisionError
WARNING:demo::
WARNING:demo:division by zero

Как видите, этот вывод не очень удобен. Дело в том, что нижележащий код, который записывает данные в sys.stderr, выполняет несколько операций записи, каждая из которых приводит к появлению отдельной строки журнала (например, последние три строки выше). Чтобы обойти эту проблему, нужно буферизовать данные и выводить строки журнала только при обнаружении символа новой строки. Создадим немного более удачную реализацию LoggerWriter:

class BufferingLoggerWriter(LoggerWriter):
    def __init__(self, logger, level):
        super().__init__(logger, level)
        self.buffer = ''

    def write(self, message):
        if '\n' not in message:
            self.buffer += message
        else:
            parts = message.split('\n')
            if self.buffer:
                s = self.buffer + parts.pop(0)
                self.logger.log(self.level, s)
            self.buffer = parts.pop()
            for part in parts:
                self.logger.log(self.level, part)

Она просто буферизует данные до появления символа новой строки, а затем регистрирует целые строки. При таком подходе вывод будет лучше:

WARNING:demo:Traceback (most recent call last):
WARNING:demo:  File "/home/runner/cookbook-loggerwriter/main.py", line 55, in <module>
WARNING:demo:    main()
WARNING:demo:  File "/home/runner/cookbook-loggerwriter/main.py", line 52, in main
WARNING:demo:    1/0
WARNING:demo:ZeroDivisionError: division by zero

Единообразная обработка переводов строк в выводе журнала

Обычно регистрируемые сообщения (например, в консоль или файл) состоят из одной текстовой строки. Однако иногда требуется обрабатывать многострочные сообщения — например, если строка формата журналирования или регистрируемые данные содержат символы новой строки. Чтобы обрабатывать такие сообщения единообразно и форматировать каждую строку зарегистрированного сообщения так, как если бы она была зарегистрирована отдельно, можно воспользоваться примесью обработчика, как в следующем фрагменте:

# Assume this is in a module mymixins.py
import copy

class MultilineMixin:
    def emit(self, record):
        s = record.getMessage()
        if '\n' not in s:
            super().emit(record)
        else:
            lines = s.splitlines()
            rec = copy.copy(record)
            rec.args = None
            for line in lines:
                rec.msg = line
                super().emit(rec)

Использовать примесь можно в следующем сценарии:

import logging

from mymixins import MultilineMixin

logger = logging.getLogger(__name__)

class StreamHandler(MultilineMixin, logging.StreamHandler):
    pass

if __name__ == '__main__':
    logging.basicConfig(level=logging.DEBUG, format='%(asctime)s %(levelname)-9s %(message)s',
                        handlers = [StreamHandler()])
    logger.debug('Single line')
    logger.debug('Multiple lines:\nfool me once ...')
    logger.debug('Another single line')
    logger.debug('Multiple lines:\n%s', 'fool me ...\ncan\'t get fooled again')

При запуске сценарий выведет что-то вроде:

2025-07-02 13:54:47,234 DEBUG     Single line
2025-07-02 13:54:47,234 DEBUG     Multiple lines:
2025-07-02 13:54:47,234 DEBUG     fool me once ...
2025-07-02 13:54:47,234 DEBUG     Another single line
2025-07-02 13:54:47,234 DEBUG     Multiple lines:
2025-07-02 13:54:47,234 DEBUG     fool me ...
2025-07-02 13:54:47,234 DEBUG     can't get fooled again

Если же вас беспокоит инъекция в журнал, можно использовать форматировщик, экранирующий символы новой строки, как в следующем примере:

import logging

logger = logging.getLogger(__name__)

class EscapingFormatter(logging.Formatter):
    def format(self, record):
        s = super().format(record)
        return s.replace('\n', r'\n')

if __name__ == '__main__':
    h = logging.StreamHandler()
    h.setFormatter(EscapingFormatter('%(asctime)s %(levelname)-9s %(message)s'))
    logging.basicConfig(level=logging.DEBUG, handlers = [h])
    logger.debug('Single line')
    logger.debug('Multiple lines:\nfool me once ...')
    logger.debug('Another single line')
    logger.debug('Multiple lines:\n%s', 'fool me ...\ncan\'t get fooled again')

Разумеется, можно использовать любой подход к экранированию, который кажется вам наиболее подходящим. При запуске сценарий должен вывести примерно следующее:

2025-07-09 06:47:33,783 DEBUG     Single line
2025-07-09 06:47:33,783 DEBUG     Multiple lines:\nfool me once ...
2025-07-09 06:47:33,783 DEBUG     Another single line
2025-07-09 06:47:33,783 DEBUG     Multiple lines:\nfool me ...\ncan't get fooled again

Поведение с экранированием не может быть вариантом по умолчанию в стандартной библиотеке, поскольку это нарушило бы обратную совместимость.

Шаблоны, которых следует избегать

Хотя в предыдущих разделах описывались способы выполнения задач, с которыми вам, возможно, придётся столкнуться, стоит упомянуть некоторые неудачные шаблоны использования, которых в большинстве случаев следует избегать. Следующие разделы расположены в произвольном порядке.

Другие ресурсы