Пособие по ведению журналов

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

Несколько вызовов 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 и выше в файл, а сообщения с уровнями 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='/temp/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 отображается только в файле. Другие сообщения отправляются в оба места назначения.

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

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

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

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 очень прост: ему передаётся очередь и некоторые обработчики, и он запускает внутренний поток, который прослушивает очередь для записей журнала, отправленных из 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

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

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

Вы также можете написать свой собственный обработчик, который использует класс Lock из модуля multiprocessing, чтобы сериализовать доступ к файлу из ваших процессов. Существующие FileHandler и подклассы в настоящее время не используют multiprocessing, хотя они могут сделать это в будущем. Обратите внимание, что в настоящее время модуль multiprocessing не предоставляет работоспособную функциональность блокировки на всех платформах (см. https://bugs.python.org/issue3770).

В качестве альтернативы, вы можете использовать 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

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

Использование вращения файлов

Иногда требуется, чтобы файл журнала рос до определенного размера, затем открывался новый файл, и в него записывались данные. Можно сохранить определенное количество таких файлов, а при достижении этого количества файлов вращать файлы, чтобы количество файлов и размер файлов оставались ограниченными. Для этой схемы использования пакет регистрации предоставляет 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, потому что внутренне пакет регистрации использует %-форматирование для объединения строки форматирования и переменных аргументов. Это невозможно изменить, сохранив обратную совместимость, так как все вызовы регистрации, которые существуют в существующем коде, будут использовать %-форматирование строк.

Однако есть способ использовать форматирование {} и $ для построения ваших отдельных сообщений журнала. Помните, что для сообщения вы можете использовать произвольный объект как строку форматирования сообщения, и пакет регистрации вызовет 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 __init__(self, logger, extra=None):
        super().__init__(logger, extra or {})

    def log(self, level, msg, /, *args, **kwargs):
        if self.isEnabledFor(level):
            msg, kwargs = self.process(msg, kwargs)
            self.logger._log(level, Message(msg, args), (), **kwargs)

logger = StyleAdapter(logging.getLogger(__name__))

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

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

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

Настройка 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 — пример ZeroMQ

Можно использовать подкласс QueueHandler, чтобы отправлять сообщения в другие типы очередей, например, в сокет ZeroMQ «publish». В примере ниже сокет создаётся отдельно и передаётся обработчику (в качестве его «очереди»):

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

Наследование от QueueListener — пример ZeroMQ

Можно также унаследовать от QueueListener, чтобы получать сообщения из других типов очередей, например, из сокета ZeroMQ «subscribe». Вот пример:

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)

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

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

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

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

Использование rotator и namer для настройки обработки вращения логов

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

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

def rotator(source, dest):
    with open(source, "rb") as sf:
        data = sf.read()
        compressed = zlib.compress(data, 9)
        with open(dest, "wb") as df:
            df.write(compressed)
    os.remove(source)

rh = logging.handlers.RotatingFileHandler(...)
rh.rotator = rotator
rh.namer = namer

Это не «истинные» файлы .gz, так как это необработанные сжатые данные без «контейнера», такого как в реальном файле gzip. Этот фрагмент приведен только для иллюстрации.

Более подробный пример с использованием многопроцессорности

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

В примере главный процесс порождает процесс-слушатель и несколько рабочих процессов. У каждого из главного процесса, процесса-слушателя и рабочих процессов есть три отдельных конфигурации (рабочие процессы все разделяют одну и ту же конфигурацию). Мы можем увидеть ведение журнала в главном процессе, как рабочие процессы записывают в 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'
   

   Код Unicode U+FEFF, при кодировании с помощью UTF-8, будет закодирован как маркер порядка байтов 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, как в следующем полном примере:

from __future__ import unicode_literals

import json
import logging

# This next bit is to ensure the script runs unchanged on 2.x and 3.x
try:
    unicode
except NameError:
    unicode = str

class Encoder(json.JSONEncoder):
    def default(self, o):
        if isinstance(o, set):
            return tuple(o)
        elif isinstance(o, unicode):
            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://..., как описано в Доступ к внешним объектам. Например, вы могли бы использовать текст 'ext://__main__.MyFilter' вместо 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())

При запуске этого скрипта должно прозвучать «Привет» и затем «До свидания» женским голосом.

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

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

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

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

В примере скрипта есть простая функция foo, которая просто циклически проходит через все уровни регистрации, записывая в sys.stderr то, какой уровень она собирается зарегистрировать, а затем фактически регистрирует сообщение на этом уровне. Вы можете передать параметр в foo, который, если он имеет значение true, будет регистрировать на уровнях 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):
    ...

Форматирование времени с использованием 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 — нет. Затем мы временно изменяем уровень на DEBUG в следующем with блоке, и поэтому сообщение #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.

Шаблон приложения командной строки

Вот пример, который показывает, как можно:

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

Предположим, у нас есть приложение командной строки, задачей которого является остановка, запуск или перезапуск некоторых служб. Для целей иллюстрации это может быть организовано как файл 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.

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

import datetime
import logging
import random
import sys
import time

# Deal with minor differences between PySide2 and PyQt5
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)
            level = random.choice(LEVELS)
            logger.log(level, 'Message after delay of %3.1f: %d', delay, i, 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')
        f.setStyleHint(f.Monospace)
        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()
    sys.exit(app.exec_())

if __name__=='__main__':
    main()

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

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