Сборник рецептов по логированию¶Logging Cookbook

import logging
import auxiliary_module

# создать логгер с именем 'spam_application'
logger = logging.getLogger('spam_application')
logger.setLevel(logging.DEBUG)
# создать файловый обработчик, записывающий даже отладочные сообщения
fh = logging.FileHandler('spam.log')
fh.setLevel(logging.DEBUG)
# создать консольный обработчик с более высоким уровнем логирования
ch = logging.StreamHandler()
ch.setLevel(logging.ERROR)
# создать форматтер и добавить его к обработчикам
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
fh.setFormatter(formatter)
ch.setFormatter(formatter)
# добавить обработчики к логгеру
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

# создать логгер
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"')

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

import logging

logger = logging.getLogger('simple_example')
logger.setLevel(logging.DEBUG)
# создать файловый обработчик, записывающий даже отладочные сообщения
fh = logging.FileHandler('spam.log')
fh.setLevel(logging.DEBUG)
# создать консольный обработчик с более высоким уровнем логирования
ch = logging.StreamHandler()
ch.setLevel(logging.ERROR)
# создать форматтер и добавить его к обработчикам
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
ch.setFormatter(formatter)
fh.setFormatter(formatter)
# добавить обработчики в логгер
logger.addHandler(ch)
logger.addHandler(fh)

# код 'приложения'
logger.debug('debug message')
logger.info('info message')
logger.warning('warn message')
logger.error('error message')
logger.critical('critical message')

import logging

# настроить логирование в файл – подробнее см. предыдущий раздел
logging.basicConfig(level=logging.DEBUG,
                    format='%(asctime)s %(name)-12s %(levelname)-8s %(message)s',
                    datefmt='%m-%d %H:%M',
                    filename='/tmp/myapp.log',
                    filemode='w')
# Определяет обработчик, который записывает сообщения уровня INFO и выше в sys.stderr.
console = logging.StreamHandler()
console.setLevel(logging.INFO)
# Задает формат, упрощенный для вывода в консоль.
formatter = logging.Formatter('%(name)-12s: %(levelname)-8s %(message)s')
# Указывает обработчику использовать этот формат.
console.setFormatter(formatter)
# Добавляет обработчик в корневой логгер.
logging.getLogger().addHandler(console)

# Теперь можно выполнять запись в корневой логгер или любой другой. Сначала корневой...
logging.info('Jackdaws love my big sphinx of quartz.')

# Теперь определите несколько других логгеров, которые могут представлять разные части приложения:
# приложение:

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.')

import logging
import logging.config
import time
import os

# Прочитать начальный конфигурационный файл.
logging.config.fileConfig('logging.conf')

# Создать и запустить слушатель на порту 9999.
t = logging.config.listen(9999)
t.start()

logger = logging.getLogger('simpleExample')

try:
    # Прокрутить вызовы логирования, чтобы увидеть разницу.
    # Создавать новые конфигурации до нажатия Ctrl+C.
    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:
    # очистка
    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')

que = queue.Queue(-1)  # Без ограничения размера.
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()
# Вывод лога будет отображать поток, создавший событие (главный поток), а не внутренний поток,
# отслеживающий внутреннюю очередь.
# Именно это и должно происходить.
# то, что должно произойти.
root.warning('Look out!')
listener.stop()

import logging, logging.handlers

rootLogger = logging.getLogger()
rootLogger.setLevel(logging.DEBUG)
socketHandler = logging.handlers.SocketHandler('localhost',
                    logging.handlers.DEFAULT_TCP_LOGGING_PORT)
# Не нужно использовать форматтер, так как сокет-обработчик отправляет событие в виде неформатированного пикла.
# неформатированный pickle
rootLogger.addHandler(socketHandler)

# Теперь можно выполнять запись в корневой логгер или любой другой. Сначала корневой...
logging.info('Jackdaws love my big sphinx of quartz.')

# Теперь определите несколько других логгеров, которые могут представлять разные части приложения:
# приложение:

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.')

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


class LogRecordStreamHandler(socketserver.StreamRequestHandler):
    """Обработчик для потокового запроса логирования.

    Это по сути записывает запись, используя ту политику логирования, которая настроена локально.
    настроено локально.
    """

    def handle(self):
        """
        Обрабатывает несколько запросов – каждый ожидается в формате: 4-байтовая длина, затем запись журнала в формате пикл.
        Записывает запись в соответствии с политикой, настроенной локально.
        согласно локально настроенной политике.
        """
        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 self.server.logname is not None:
            name = self.server.logname
        else:
            name = record.name
        logger = logging.getLogger(name)
        # Примечание: КАЖДАЯ запись попадает в журнал. Это связано с тем, что Logger.handle
        # обычно вызывается ПОСЛЕ фильтрации на уровне регистратора. Если требуется
        # выполнять фильтрацию, делайте это на стороне клиента, чтобы не тратить
        # циклы процессора и сетевую пропускную способность!
        logger.handle(record)

class LogRecordSocketReceiver(socketserver.ThreadingTCPServer):
    """
    Простой приёмник журнала на основе TCP-сокетов, подходящий для тестирования.
    """

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

import copy
import logging

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

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

    logger.info('A log message')

# Вам понадобятся эти импорты в вашем коде.
import logging
import logging.handlers
import multiprocessing

# Следующие две строки импорта только для этой демонстрации.
from random import choice, random
import time

#
# Поскольку нужно задать настройки логирования для слушателя и рабочих процессов, функции
# слушателя и рабочего процесса принимают параметр configurer – вызываемый объект для
# настройки логирования этого процесса. Эти функции также получают очередь, которую
# используют для взаимодействия.
#
# На практике слушателя можно настроить как угодно, но заметьте: в этом простом
# примере слушатель не применяет логику уровней или фильтров к полученным записям.
# На практике эту логику, вероятно, стоит вынести в рабочие процессы, чтобы не
# пересылать между процессами события, которые всё равно будут отфильтрованы.
#
# Размер файлов при ротации намеренно сделан небольшим, чтобы результат был хорошо виден.
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)

# Это главный цикл процесса-слушателя: ждём события логирования
# (LogRecords) в очереди и обрабатываем их; завершаем работу, когда получаем None вместо
# LogRecord.
def listener_process(queue, configurer):
    configurer()
    while True:
        try:
            record = queue.get()
            if record is None:  # Отправляем это как сигнал завершения, чтобы слушатель остановился.
                break
            logger = logging.getLogger(record.name)
            logger.handle(record)  # Никакой логики уровней и фильтров – просто делаем дело!
        except Exception:
            import sys, traceback
            print('Whoops! Problem:', file=sys.stderr)
            traceback.print_exc(file=sys.stderr)

# Массивы для случайного выбора в этой демонстрации.

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',
]

# Настройка рабочего процесса выполняется в начале его работы.
# Обратите внимание: в Windows нельзя полагаться на семантику fork, поэтому каждый процесс
# будет выполнять код настройки логирования при запуске.
def worker_configurer(queue):
    h = logging.handlers.QueueHandler(queue)  # Нужен только один обработчик.
    root = logging.getLogger()
    root.addHandler(h)
    # Отправляем все сообщения для демонстрации; никакой другой логики уровней или фильтров.
    root.setLevel(logging.DEBUG)

# Это главный цикл рабочего процесса, который просто логирует десять событий с
# случайные задержки перед завершением.
# Сообщения print нужны лишь для того, чтобы было видно, что программа работает.
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)

# Здесь организуется демонстрация. Создать очередь, создать и запустить
# слушатель, создать десять рабочих процессов и запустить их, дождаться их завершения,
# затем отправить None в очередь, чтобы сообщить слушателю о завершении.
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()
    # На этом этапе главный процесс может заняться своей полезной работой
    # Когда он с этим закончит, он может дождаться завершения рабочих процессов...
    for wp in workers:
        wp.join()
    # А теперь указать потоку журналирования завершиться
    q.put(None)
    lp.join()

import glob
import logging
import logging.handlers

LOG_FILENAME = 'logging_rotatingfile_example.out'

# Настроить конкретный регистратор с нужным уровнем вывода
my_logger = logging.getLogger('MyLogger')
my_logger.setLevel(logging.DEBUG)

# Добавить обработчик сообщений журнала в регистратор
handler = logging.handlers.RotatingFileHandler(
              LOG_FILENAME, maxBytes=20, backupCount=5)

my_logger.addHandler(handler)

# Записать несколько сообщений
for i in range(20):
    my_logger.debug('i = %d' % i)

# Посмотреть, какие файлы созданы
logfiles = glob.glob('%s*' % LOG_FILENAME)

for filename in logfiles:
    print(filename)

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

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)

import logging

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

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

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

logger = StyleAdapter(logging.getLogger(__name__))

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

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

logger = logging.getLogger(__name__)

old_factory = logging.getLogRecordFactory()

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

logging.setLogRecordFactory(record_factory)

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

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

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

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


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

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

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

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

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

        if logger.isEnabledFor(record.levelno):
            # Имя процесса преобразуется только для того, чтобы показать, что это слушатель
            # выполняет логирование в файлы и консоль.
            record.processName = '%s (for %s)' % (current_process().name, record.processName)
            logger.handle(record)

def listener_process(q, stop_event, config):
    """
    Это можно было бы сделать в главном процессе, но сделано в отдельном
    процессе для наглядности.

    Это инициализирует логирование согласно указанной конфигурации
    запускает слушатель и ожидает сигнала от главного процесса о завершении
    через событие. Затем слушатель останавливается, и процесс завершается.
    """
    logging.config.dictConfig(config)
    listener = logging.handlers.QueueListener(q, MyHandler())
    listener.start()
    if os.name == 'posix':
        # В POSIX логгер setup уже был настроен в
        # родительском процессе, но должен был быть отключён после вызова
        # dictConfig.
        # В Windows, поскольку fork не используется, логгер setup не будет
        # существовать в дочернем процессе, поэтому он будет создан, и сообщение
        # появится – отсюда и конструкция "if posix".
        logger = logging.getLogger('setup')
        logger.critical('Should not appear, because of disabled logger ...')
    stop_event.wait()
    listener.stop()

def worker_process(config):
    """
    Некоторое количество таких процессов порождается для иллюстрации. На
    практике это могла бы быть разнородная группа процессов, а не
    идентичные друг другу.

    Это инициализирует логирование согласно указанной конфигурации
    и записывает сотню сообщений со случайными уровнями в случайно выбранные
    логгеры.

    Добавлена небольшая задержка, чтобы дать другим процессам шанс выполниться. Это
    не является строго необходимым, но позволяет немного перемешать вывод от разных
    процессов больше, чем если бы её не было.
    """
    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':
        # В POSIX логгер setup уже был настроен в
        # родительском процессе, но должен был быть отключён после вызова
        # dictConfig.
        # В Windows, поскольку fork не используется, логгер setup не будет
        # существовать в дочернем процессе, поэтому он будет создан, и сообщение
        # появится – отсюда и конструкция "if posix".
        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()
    # Главный процесс получает простую конфигурацию, которая выводит информацию в консоль.
    config_initial = {
        'version': 1,
        'handlers': {
            'console': {
                'class': 'logging.StreamHandler',
                'level': 'INFO'
            }
        },
        'root': {
            'handlers': ['console'],
            'level': 'DEBUG'
        }
    }
    # Конфигурация рабочего процесса представляет собой QueueHandler, прикреплённый к
    # корневому логгеру, что позволяет отправлять все сообщения в очередь.
    # Мы отключаем существующие логгеры, чтобы отключить логгер "setup", используемый в
    # родительском процессе. Это необходимо в POSIX, потому что логгер будет
    # присутствовать в дочернем процессе после вызова fork().
    config_worker = {
        'version': 1,
        'disable_existing_loggers': True,
        'handlers': {
            'queue': {
                'class': 'logging.handlers.QueueHandler',
                'queue': q
            }
        },
        'root': {
            'handlers': ['queue'],
            'level': 'DEBUG'
        }
    }
    # Конфигурация процесса-слушателя показывает, что вся гибкость
    # конфигурации логирования доступна для отправки событий обработчикам так, как
    # требуется.
    # Мы отключаем существующие логгеры, чтобы отключить логгер "setup", используемый в
    # родительском процессе. Это необходимо в POSIX, потому что логгер будет
    # присутствовать в дочернем процессе после вызова 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'
        }
    }
    # Записываем несколько начальных событий, просто чтобы показать, что логирование в родительском процессе работает
    # нормально.
    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')
    # Теперь ожидаем завершения работы рабочих процессов.
    for wp in workers:
        wp.join()
    # Все рабочие процессы завершены, теперь можно остановить прослушивание.
    # Логирование в родительском процессе всё ещё работает нормально.
    logger.info('Telling listener to stop ...')
    stop_event.set()
    lp.join()
    logger.info('All done.')

if __name__ == '__main__':
    main()

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   # необязательно, для улучшения читаемости

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

import json
import logging


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

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

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

_ = StructuredMessage   # необязательно, для улучшения читаемости

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

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)