日志操作手冊?
- 作者
Vinay Sajip <vinay_sajip at red-dove dot com>
本頁包含了許多日志記錄相關的概念，這些概念在過去一直被認為很有用。
在多個模塊中記錄日志?
多次調用``logging.getLogger('someLogger')``會返回對同一個日志記錄器對象的引用。不僅在同一個模塊中是這樣的，而且在不同模塊之間，只要是在同一個Python解釋器進程中，也是如此。 這就是對同一個對象的多個引用；此外，應用程序代碼也可以在一個模塊中定義和配置父日志記錄器，在單獨的模塊中創(chuàng)建（但不配置）一個子日志記錄器，并且對子日志記錄器的所有調用都將傳遞給父日志記錄器。 這里是一個主要模塊:
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*的日志處理器。
在編寫和測試應用程序時，能夠創(chuàng)建帶有更高或更低消息等級的過濾器的日志處理器是非常有用的。為了避免過多地使用 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')
處理日志處理器的阻塞?
有時候需要讓日志處理程序在不阻塞當前正在記錄線程的情況下完成工作。 這在Web應用程序中很常見，當然也會在其他場景中出現。
一個常見的緩慢行為是 SMTPHandler: 由于開發(fā)者無法控制的多種原因（例如，性能不佳的郵件或網絡基礎架構），發(fā)送電子郵件可能需要很長時間。 其實幾乎所有基于網絡的處理程序都可能造成阻塞：即便是 SocketHandler 也可能在底層進行 DNS 查詢，這太慢了（這個查詢會深入至套接字代碼，位于 Python 層之下，這是不受開發(fā)者控制的）。
一種解決方案是分成兩部分去處理。第一部分，針對那些對性能有要求的關鍵線程的日志記錄附加一個 QueueHandler。 日志記錄器只需簡單寫入隊列，該隊列可以設置一個足夠大的容量甚至不設置容量上限。通常寫入隊列是一個快速的操作，即使可能需要在代碼中去捕獲例如 queue.Full 等異常。 如果你是一名處理關鍵線程的開發(fā)者，請務必記錄這些信息 (包括建議只為日志處理器附加 QueueHandlers) 以便于其他開發(fā)者使用你的代碼。
解決方案的另一部分是 QueueListener，它被設計用來作為 QueueHandler 的對應。 QueueListener 非常簡單：向其傳入一個隊列和一些處理句柄，它會啟動一個內部線程來監(jiān)聽從 QueueHandlers (或任何其他可用的 LogRecords 源) 發(fā)送過來的 LogRecords 隊列。 LogRecords 會從隊列中被移除，并被傳遞給句柄進行處理。
使用一個單獨的類 QueueListener?優(yōu)點是可以使用同一個實例去服務于多個``QueueHandlers``。這樣會更節(jié)省資源，否則每個處理程序都占用一個線程沒有任何益處。
以下是使用了這樣兩個類的示例(省略了導入語句):
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!
在 3.5 版更改: 在Python 3.5之前，QueueListener?總是把從隊列中接收的每個消息都傳給它初始化的日志處理程序。(這是因為它會假設過濾級別總是在隊列的另一側去設置的。) 從Python 3.5開始，可以通過在監(jiān)聽器構造函數中添加一個參數``respect_handler_level=True``改變這種情況。當這樣設置時，監(jiān)聽器會比較每條消息的等級和日志處理器中設置的等級，只把需要傳遞的消息傳給對應的日志處理器。
通過網絡發(fā)送和接收日志?
如果你想在網絡上發(fā)送日志，并在接收端處理它們。一個簡單的方式是通過附加一個 SocketHandler 的實例在發(fā)送端的根日志處理器中:
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.
請注意，在某些情況下序列化會存在一些安全。如果這影響到你，那么你可以通過覆蓋 makePickle() 方法，使用自己的實現來解決，并調整上述腳本也使用覆蓋后的序列化方法。
在日志記錄中添加上下文信息?
有時，除了傳遞給日志記錄器調用的參數外，我們還希望日志記錄中包含上下文信息。例如，有一個網絡應用，可能需要記錄一些特殊的客戶端信息在日志中（比如客戶端的用戶名、IP地址等）。雖然你可以通過設置額外的參數去達到這個目的，但這種方式不一定方便?；蛘吣憧赡芟氲皆诿總€連接的基礎上創(chuàng)建一個 Logger 的實例，但這些實例是不會被垃圾回收的，這在練習中也許不是問題，但當 Logger 的實例數量取決于你應用程序中想記錄的細致程度時，如果 Logger 的實例數量不受限制的話，將會變得難以管理。
使用日志適配器傳遞上下文信息?
一個傳遞上下文信息和日志事件信息的簡單辦法是使用類 LoggerAdapter。 這個類設計的像 Logger，所以可以直接調用 debug()、info()、 warning()、 error()、exception()、 critical() 和 log()。 這些方法在對應的 Logger 中使用相同的簽名，所以可以交替使用兩種類型的實例。
當你創(chuàng)建一個 LoggerAdapter 的實例時，你會傳入一個 Logger 的實例和一個包含了上下文信息的字典對象。當你調用一個 LoggerAdapter 實例的方法時，它會把調用委托給內部的 Logger 的實例，并為其整理相關的上下文信息。這是 LoggerAdapter 的一個代碼片段:
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)
LoggerAdapter 的 process() 方法是將上下文信息添加到日志的輸出中。 它傳入日志消息和日志調用的關鍵字參數，并傳回（隱式的）這些修改后的內容去調用底層的日志記錄器。此方法的默認參數只是一個消息字段，但留有一個 'extra' 的字段作為關鍵字參數傳給構造器。當然，如果你在調用適配器時傳入了一個 'extra' 字段的參數，它會被靜默覆蓋。
使用 'extra' 的優(yōu)點是這些鍵值對會被傳入 LogRecord 實例的 __dict__ 中，讓你通過 Formatter 的實例直接使用定制的字符串，實例能找到這個字典類對象的鍵。 如果你需要一個其他的方法，比如說，想要在消息字符串前后增加上下文信息，你只需要創(chuàng)建一個 LoggerAdapter 的子類，并覆蓋它的 process() 方法來做你想做的事情，以下是一個簡單的示例:
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})
然后，你記錄在適配器中的任何事件消息前將添加``some_conn_id``的值。
使用除字典之外的其它對象傳遞上下文信息?
你不需要將一個實際的字典傳遞給 LoggerAdapter-你可以傳入一個實現了``__getitem__`` 和``__iter__``的類的實例，這樣它就像是一個字典。這對于你想動態(tài)生成值（而字典中的值往往是常量）將很有幫助。
使用過濾器傳遞上下文信息?
你也可以使用一個用戶定義的類 Filter 在日志輸出中添加上下文信息。Filter 的實例是被允許修改傳入的 LogRecords，包括添加其他的屬性，然后可以使用合適的格式化字符串輸出，或者可以使用一個自定義的類 Formatter。
例如，在一個web應用程序中，正在處理的請求（或者至少是請求的一部分），可以存儲在一個線程本地(threading.local) 變量中，然后從``Filter``中去訪問。請求中的信息，如IP地址和用戶名將被存儲在``LogRecord``中，使用上例``LoggerAdapter``中的 'ip' 和 'user' 屬性名。在這種情況下，可以使用相同的格式化字符串來得到上例中類似的輸出結果。這是一段示例代碼:
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
從多個進程記錄至單個文件?
盡管 logging 是線程安全的，將單個進程中的多個線程日志記錄至單個文件也 是 受支持的，但將 多個進程 中的日志記錄至單個文件則 不是 受支持的，因為在 Python 中并沒有在多個進程中實現對單個文件訪問的序列化的標準方案。 如果你需要將多個進程中的日志記錄至單個文件，有一個方案是讓所有進程都將日志記錄至一個 SocketHandler，然后用一個實現了套接字服務器的單獨進程一邊從套接字中讀取一邊將日志記錄至文件。 （如果愿意的話，你可以在一個現有進程中專門開一個線程來執(zhí)行此項功能。） 這一部分 文檔對此方式有更詳細的介紹，并包含一個可用的套接字接收器，你自己的應用可以在此基礎上進行適配。
如果你使用的是包含了 multiprocessing 模塊的較新版本的 Python，你也可以使用 Lock 來編寫自己的處理程序讓其從多個進程中按順序記錄至文件。 現有的 FileHandler 和它的子類目前沒有使用 multiprocessing，盡管將來可能會這樣做。 請注意目前 multiprocessing 模塊并非在所有平臺上提供可用的鎖功能 (參見 https://bugs.python.org/issue3770)。
或者，你也可以使用 Queue 和 QueueHandler 將所有的日志事件發(fā)送至你的多進程應用的一個進程中。 以下示例腳本演示了如何執(zhí)行此操作。 在示例中，一個單獨的監(jiān)聽進程負責監(jiān)聽其他進程的日志事件，并根據自己的配置記錄。 盡管示例只演示了這種方法（例如你可能希望使用單獨的監(jiān)聽線程而非監(jiān)聽進程 —— 它們的實現是類似的），但你也可以在應用程序的監(jiān)聽進程和其他進程使用不同的配置，它可以作為滿足你特定需求的一個基礎:
# 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 子系統(tǒng)中所有的事件記錄至一個文件 mplog-foo.log。在主進程（即使是在工作進程中產生的日志事件）的日志記錄機制中將直接使用恰當的配置。
concurrent.futures.ProcessPoolExecutor 的用法?
若要利用 concurrent.futures.ProcessPoolExecutor 啟動工作進程，創(chuàng)建隊列的方式應稍有不同。不能是：
queue = multiprocessing.Queue(-1)
而應是：
queue = multiprocessing.Manager().Queue(-1) # also works with the examples above
然后就可以將以下工作進程的創(chuàng)建過程：
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()
改為 (記得要先導入 concurrent.futures):
with concurrent.futures.ProcessPoolExecutor(max_workers=10) as executor:
for i in range(10):
executor.submit(worker_process, queue, worker_configurer)
輪換日志文件?
有時，你希望當日志文件不斷記錄增長至一定大小時，打開一個新的文件接著記錄。 你可能希望只保留一定數量的日志文件，當不斷的創(chuàng)建文件到達該數量時，又覆蓋掉最開始的文件形成循環(huán)。 對于這種使用場景，日志包提供了 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
最新的文件始終是:file:logging_rotatingfile_example.out，每次到達大小限制時，都會使用后綴``.1``重命名。每個現有的備份文件都會被重命名并增加其后綴（例如``.1`` 變?yōu)閌`.2``），而``.6``文件會被刪除掉。
顯然，這個例子將日志長度設置得太小，這是一個極端的例子。 你可能希望將*maxBytes*設置為一個合適的值。
使用其他日志格式化方式?
當日志模塊被添加至 Python 標準庫時，只有一種格式化消息內容的方法即 %-formatting。 在那之后，Python 又增加了兩種格式化方法: string.Template (在 Python 2.4 中新增) 和 str.format() (在 Python 2.6 中新增)。
日志（從 3.2 開始）為這兩種格式化方式提供了更多支持。Formatter 類可以添加一個額外的可選關鍵字參數 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
>>>
請注意最終輸出到日志的消息格式完全獨立于單條日志消息的構造方式。 它仍然可以使用 %-formatting，如下所示:
>>> 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 語法進行日志調用，因為日志包在內部使用 %-f 格式來合并格式串和參數變量。在保持向下兼容性時，這一點不會改變，因為已有代碼中的所有日志調用都會使用%-f 格式串。
還有一種方法可以構建自己的日志信息，就是利用 {}- 和 $- 格式?；叵胍幌拢我鈱ο蠖伎捎脼槿罩拘畔⒌母袷酱?，日志包將會調用該對象的 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)
上述兩個類均可代替格式串，使得能用 {}- 或 $-formatting 構建最終的“日志信息”部分，這些信息將出現在格式化后的日志輸出中，替換 %(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() 的方法來應用。
值得注意的是，上述做法對性能并沒什么影響：格式化過程其實不是在日志記錄調用時發(fā)生的，而是在日志信息即將由 handler 輸出到日志時發(fā)生。因此，唯一可能讓人困惑的稍不尋常的地方，就是包裹在格式串和參數外面的括號，而不是格式串。因為 __ 符號只是對 XXXMessage 類的構造函數調用的語法糖。
只要愿意，上述類似的效果即可用 LoggerAdapter 實現，如下例所示：
import logging
class Message(object):
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(StyleAdapter, self).__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()
在用 Python 3.2 以上版本運行時，上述代碼應該會把 Hello, world! 寫入日志。
自定義 LogRecord?
每條日志事件都由一個 LogRecord 實例表示。當某事件要記入日志并且沒有被某級別過濾掉時，就會創(chuàng)建一個 LogRecord 對象，并將有關事件的信息填入，傳給該日志對象的 handler（及其祖先，直至對象禁止向上傳播為止）。在 Python 3.2 之前，只有兩個地方會進行事件的創(chuàng)建：
Logger.makeRecord()，在事件正常記入日志的過程中調用。這會直接調用LogRecord來創(chuàng)建一個實例。makeLogRecord()，調用時會帶上一個字典參數，其中存放著要加入 LogRecord 的屬性。這通常在通過網絡接收到合適的字典時調用（如通過SocketHandler以 pickle 形式，或通過HTTPHandler以 JSON 形式）。
于是這意味著若要對 LogRecord 進行定制，必須進行下述某種操作。
創(chuàng)建
Logger? 自定義子類，重寫Logger.makeRecord()，并在實例化所需日志對象之前用setLoggerClass()進行設置。為日志對象添加
Filter或 handler，當其filter()方法被調用時，會執(zhí)行必要的定制操作。
比如說在有多個不同庫要完成不同操作的場景下，第一種方式會有點笨拙。 每次都要嘗試設置自己的 Logger 子類，而起作用的是最后一次嘗試。
第二種方式在多數情況下效果都比較良好，但不允許你使用特殊化的 LogRecord 子類。 庫開發(fā)者可以為他們的日志記錄器設置合適的過濾器，但他們應當要記得每次引入新的日志記錄器時都需如此（他們只需通過添加新的包或模塊并執(zhí)行以下操作即可）:
logger = logging.getLogger(__name__)
或許這樣要顧及太多事情。開發(fā)人員還可以將過濾器附加到其頂級日志對象的 NullHandler 中，但如果應用程序開發(fā)人員將 handler 附加到較底層庫的日志對象，則不會調用該過濾器 --- 所以 handler 輸出的內容不會符合庫開發(fā)人員的預期。
在 Python 3.2 以上版本中，LogRecord 的創(chuàng)建是通過工廠對象完成的，工廠對象可以指定。工廠對象只是一個可調用對象，可以用 setLogRecordFactory() 進行設置，并用 getLogRecordFactory() 進行查詢。工廠對象的調用參數與 LogRecord 的構造函數相同，因為 LogRecord 是工廠對象的默認設置。
這種方式可以讓自定義工廠對象完全控制 LogRecord 的創(chuàng)建過程。比如可以返回一個子類，或者在創(chuàng)建的日志對象中加入一些額外的屬性，使用方式如下所示：
old_factory = logging.getLogRecordFactory()
def record_factory(*args, **kwargs):
record = old_factory(*args, **kwargs)
record.custom_attribute = 0xdecafbad
return record
logging.setLogRecordFactory(record_factory)
這種模式允許不同的庫將多個工廠對象鏈在一起，只要不會覆蓋彼此的屬性或標準屬性，就不會出現意外。但應記住，工廠鏈中的每個節(jié)點都會增加日志操作的運行開銷，本技術僅在采用 Filter 無法達到目標時才應使用。
子類化 QueueHandler - ZeroMQ 示例?
你可以使用 QueueHandler 子類將消息發(fā)送給其他類型的隊列 ，比如 ZeroMQ 'publish' 套接字。 在以下示例中，套接字將單獨創(chuàng)建并傳給處理句柄 (作為它的 'queue'):
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)
當然還有其他方案，比如通過 hander 傳入所需數據，以創(chuàng)建 socket：
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)
參見
- 模塊
logging 日志記錄模塊的 API 參考。
- 模塊
logging.config 日志記錄模塊的配置 API 。
- 模塊
logging.handlers 日志記錄模塊附帶的有用處理程序。
基于字典進行日志配置的示例?
Below is an example of a logging configuration dictionary - it's taken from
the documentation on the Django project.
This dictionary is passed to dictConfig() to put the configuration into effect:
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']
}
}
}
For more information about this configuration, you can see the relevant section of the Django documentation.
利用 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 和較為復雜的日志配置，如何將由隊列接收到的事件分發(fā)給配置指定的 handler。請注意，這些配置純粹用于演示，但應該能調整代碼以適用于自己的場景。
以下是代碼——但愿文檔字符串和注釋能有助于理解其工作原理：
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):
logger = logging.getLogger(record.name)
# 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,
'formatters': {
'detailed': {
'class': 'logging.Formatter',
'format': '%(asctime)s %(name)-15s %(levelname)-8s %(processName)-10s %(message)s'
}
},
'handlers': {
'console': {
'class': 'logging.StreamHandler',
'level': 'INFO',
},
},
'root': {
'level': 'DEBUG',
'handlers': ['console']
},
}
# 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': {
'level': 'DEBUG',
'handlers': ['queue']
},
}
# 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',
'level': 'INFO',
'formatter': 'simple',
},
'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']
},
}
# 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()
在發(fā)送給 SysLogHandler 的信息中插入一個 BOM。?
RFC 5424 要求，Unicode 信息應采用字節(jié)流形式發(fā)送到系統(tǒng) syslog 守護程序，字節(jié)流結構如下所示：可選的純 ASCII部分，后跟 UTF-8 字節(jié)序標記（BOM），然后是采用 UTF-8 編碼的 Unicode。（參見 相關規(guī)范 。）
在 Python 3.1 的 SysLogHandler 中，已加入了在日志信息中插入 BOM 的代碼，但不幸的是，代碼并不正確，BOM 出現在了日志信息的開頭，因此在它之前就不允許出現純 ASCII 內容了。
由于無法正常工作， Python 3.2.4 以上版本已刪除了出錯的插入 BOM 代碼。但已有版本的代碼不會被替換，若要生成與 RFC 5424 兼容的日志信息，包括一個 BOM 符，前面有可選的純 ASCII 字節(jié)流，后面為 UTF-8 編碼的任意 Unicode，那么 需要執(zhí)行以下操作：
為
SysLogHandler實例串上一個Formatter實例，格式串可如下：'ASCII section\ufeffUnicode section'
用 UTF-8 編碼時，Unicode 碼位 U+FEFF 將會編碼為 UTF-8 BOM——字節(jié)串
b'\xef\xbb\xbf'。用任意占位符替換 ASCII 部分，但要保證替換之后的數據一定是 ASCII 碼（這樣在 UTF-8 編碼后就會維持不變）。
用任意占位符替換 Unicode 部分；如果替換后的數據包含超出 ASCII 范圍的字符，沒問題——他們將用 UTF-8 進行編碼。
SysLogHandler 將 對格式化后的日志信息進行 UTF-8 編碼。如果遵循上述規(guī)則，應能生成符合 RFC 5424 的日志信息。否則，日志記錄過程可能不會有什么反饋，但日志信息將不與 RFC 5424 兼容，syslog 守護程序可能會有出錯反應。
結構化日志的實現代碼?
大多數日志信息是供人閱讀的，所以機器解析起來并不容易，但某些時候可能希望以結構化的格式輸出，以 能夠 被程序解析（無需用到復雜的正則表達式）。這可以直接用 logging 包實現。實現方式有很多，以下是一種比較簡單的方案，利用 JSON 以機器可解析的方式對事件信息進行序列化：
import json
import logging
class StructuredMessage(object):
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(Encoder, self).default(o)
class StructuredMessage(object):
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() 自定義 handler?
有時需要以特定方式自定義日志 handler，如果采用 dictConfig()，可能無需生成子類就可以做到。比如要設置日志文件的所有權。在 POSIX 上，可以利用 shutil.chown() 輕松完成，但 stdlib 中的文件 handler 并不提供內置支持。于是可以用普通函數自定義 handler 的創(chuàng)建，例如：
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() 的日志配置中指定通過調用此函數來創(chuàng)建日志處理程序:
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() 是從此版本開始出現的。 此方式應當適用于任何支持 dictConfig() 的 Python 版本 —— 例如 Python 2.7, 3.2 或更新的版本。 對于 3.3 之前的版本，你應當使用 os.chown() 之類的函數來實現實際的所有權修改。
實際應用中，handler 的創(chuàng)建函數可能位于項目的工具模塊中。以下配置：
'()': owned_file_handler,
應使用：
'()': 'ext://project.util.owned_file_handler',
這里的 project.util 可以換成函數所在包的實際名稱。 在上述的可用腳本中，應該可以使用 'ext://__main__.owned_file_handler'。 在這里，實際的可調用對象是由 dictConfig() 從 ext:// 說明中解析出來的。
上述示例還指明了其他的文件修改類型的實現方案 —— 比如同樣利用 os.chmod() 設置 POSIX 訪問權限位。
當然，以上做法也可以擴展到 FileHandler 之外的其他類型的 handler ——比如某個輪換文件 handler，或類型完全不同的其他 handler。
生效于整個應用程序的格式化樣式?
在 Python 3.2 中，Formatter 增加了一個 style 關鍵字形參，它默認為 % 以便向下兼容，但是允許采用 { 或 {TX-PL-LABEL}#x60; 來支持 str.format() 和 string.Template 所支持的格式化方式。 請注意此形參控制著用用于最終輸出到日志的日志消息格式，并且與單獨日志消息的構造方式完全無關。
日志函數（debug(), info() 等）只會讀取位置參數獲取日志信息本身，而關鍵字參數僅用于確定日志函數的工作選項（比如關鍵字參數 exc_info 表示應將跟蹤信息記入日志，關鍵字參數 extra 則給出了需加入日志的額外上下文信息）。所以不能直接使用 str.format() 或 string.Template 這種語法進行日志調用，因為日志包在內部使用 %-f 格式來合并格式串和可變參數。因為尚需保持向下兼容，這一點不會改變，已有代碼中的所有日志調用都將采用 %-f 格式串。
有人建議將格式化樣式與特定的日志對象進行關聯，但其實也會遇到向下兼容的問題，因為已有代碼可能用到了某日志對象并采用了 %-f 格式串。
為了讓第三方庫和自編代碼都能夠交互使用日志功能，需要決定在單次日志記錄調用級別采用什么格式。于是就出現了其他幾種格式化樣式方案。
LogRecord 工廠的用法?
在 Python 3.2 中，伴隨著 Formatter 的上述變化，logging 包增加了允許用戶使用 setLogRecordFactory() 函數來。設置自己的 LogRecord 子類的功能。 你可以使用此功能來設置自己的 LogRecord 子類，它會通過重載 getMessage() 方法來完成適當的操作。 msg % args 格式化是在此方法的基類實現中進行的，你可以在那里用你自己的格式化操作來替換；但是，你應當注意要支持全部的格式化樣式并允許將 %-formatting 作為默認樣式，以確保與其他代碼進行配合。 還應當注意調用 str(self.msg)，正如基類實現所做的一樣。
更多信息請參閱 setLogRecordFactory() 和 LogRecord 的參考文檔。
自定義信息對象的使用?
另一種方案可能更為簡單，可以利用 {}- 和 $- 格式構建自己的日志消息。大家或許還記得（來自 使用任意對象作為消息），可以用任意對象作為日志信息的格式串，日志包將調用該對象上 str() 獲取實際的格式串?？聪乱韵聝蓚€類：
class BraceMessage(object):
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(object):
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)
以上兩個類均都可用于替代格式串，以便用 {}- 或 $-formatting 構建實際的“日志信息”部分，此部分將出現在格式化后的日志輸出中，替換 %(message)s 、“{message}”或“$message”。每次要寫入日志時都使用類名，如果覺得使用不便，可以采用 M 或 _ 之類的別名（如果將 _ 用于本地化操作，則可用 __）。
下面給出示例。 首先用 str.format() 進行格式化：
>>> __ = 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)
然后，用 string.Template 格式化：
>>> __ = DollarMessage
>>> print(__('Message with $num $what', num=2, what='placeholders'))
Message with 2 placeholders
>>>
值得注意的是，上述做法對性能并沒什么影響：格式化過程其實不是在日志調用時發(fā)生的，而是在日志信息即將由 handler 輸出到日志時發(fā)生。因此，唯一可能讓人困惑的稍不尋常的地方，就是包裹在格式串和參數外面的括號，而不是格式串。因為 __ 符號只是對 XXXMessage 類的構造函數調用的語法糖。
利用 dictConfig() 定義過濾器?
用 dictConfig() 可以 對日志過濾器進行設置，盡管乍一看做法并不明顯（所以才需要本秘籍）。 由于 Filter 是標準庫中唯一的日志過濾器類，不太可能滿足眾多的要求（它只是作為基類存在），通常需要定義自己的 Filter 子類，并重寫 filter() 方法。為此，請在過濾器的配置字典中設置 () 鍵，指定要用于創(chuàng)建過濾器的可調用對象（最明顯可用的就是給出一個類，但也可以提供任何一個可調用對象，只要能返回 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對象。與過濾器一樣，上述技術還可用于配置自定義 handler 和格式化對象。有關如何在日志配置中使用用戶自定義對象的更多信息，請參閱 用戶定義對象，以及上述 利用 dictConfig() 自定義 handler 的其他指南。
異常信息的自定義格式化?
有時可能需要設置自定義的異常信息格式——考慮到會用到參數，假定要讓每條日志事件只占一行，即便存在異常信息也一樣。這可以用自定義格式化類來實現，如下所示：
import logging
class OneLineExceptionFormatter(logging.Formatter):
def formatException(self, exc_info):
"""
Format an exception so that it prints on a single line.
"""
result = super(OneLineExceptionFormatter, self).formatException(exc_info)
return repr(result) # or format into one line however you want to
def format(self, record):
s = super(OneLineExceptionFormatter, self).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 模塊能滿足更專門的需求。
語音播報日志信息?
有時可能需要以聲音的形式呈現日志消息。如果系統(tǒng)自帶了文本轉語音 （TTS）功能，即便沒與 Python 關聯也很容易做到。大多數 TTS 系統(tǒng)都有一個可運行的命令行程序，在 handler 中可以用 subprocess 進行調用。這里假定 TTS 命令行程序不會與用戶交互，或需要很長時間才會執(zhí)行完畢，寫入日志的信息也不會多到影響用戶查看，并且可以接受每次播報一條信息，以下示例實現了等一條信息播完再處理下一條，可能會導致其他 handler 的等待。這個簡短示例僅供演示，假定 espeak TTS 包已就緒：
import logging
import subprocess
import sys
class TTSHandler(logging.Handler):
def emit(self, record):
msg = self.format(record)
# Speak slowly in a female English voice
cmd = ['espeak', '-s150', '-ven+f3', msg]
p = subprocess.Popen(cmd, stdout=subprocess.PIPE,
stderr=subprocess.STDOUT)
# wait for the program to finish
p.communicate()
def configure_logging():
h = TTSHandler()
root = logging.getLogger()
root.addHandler(h)
# the default formatter just returns the message
root.setLevel(logging.DEBUG)
def main():
logging.info('Hello')
logging.debug('Goodbye')
if __name__ == '__main__':
configure_logging()
sys.exit(main())
運行后將會以女聲播報“Hello”和“Goodbye”。
當然，上述方案也適用于其他 TTS 系統(tǒng)，甚至可以通過利用命令行運行的外部程序來處理消息。
緩沖日志消息并有條件地輸出它們?
在某些情況下，你可能希望在臨時區(qū)域中記錄日志消息，并且只在發(fā)生某種特定的情況下才輸出它們。 例如，你可能希望起始在函數中記錄調試事件，如果函數執(zhí)行完成且沒有錯誤，你不希望輸出收集的調試信息以避免造成日志混亂，但如果出現錯誤，那么你希望所有調試以及錯誤消息被輸出。
下面是一個示例，展示如何在你的日志記錄函數上使用裝飾器以實現這一功能。該示例使用 logging.handlers.MemoryHandler ，它允許緩沖已記錄的事件直到某些條件發(fā)生，緩沖的事件才會被刷新（flushed） - 傳遞給另一個處理程序（ target handler）進行處理。 默認情況下， MemoryHandler 在其緩沖區(qū)被填滿時被刷新，或者看到一個級別大于或等于指定閾值的事件。 如果想要自定義刷新行為，你可以通過更專業(yè)的 MemoryHandler 子類來使用這個秘訣。
這個示例腳本有一個簡單的函數 foo ，它只是在所有的日志級別中循環(huán)運行，寫到 sys.stderr ，說明它要記錄在哪個級別上，然后在這個級別上實際記錄一個消息。你可以給 foo 傳遞一個參數，如果為 true ，它將在ERROR和CRITICAL級別記錄，否則，它只在DEBUG、INFO和WARNING級別記錄。
腳本只是使用了一個裝飾器來裝飾 foo，這個裝飾器將記錄執(zhí)行所需的條件。裝飾器使用一個記錄器作為參數，并在調用被裝飾的函數期間附加一個內存處理程序。裝飾器可以使用目標處理程序、記錄級別和緩沖區(qū)的容量（緩沖記錄的數量）來附加參數。這些參數分別默認為寫入``sys.stderr`` 的 StreamHandler ， 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或更高的事件時發(fā)生，但在這種情況下，任何之前較低消息等級的事件還會被記錄。
你當然可以使用傳統(tǒng)的裝飾方法:
@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。 如果你想通過配置來實現這一功能，你可以使用 dictConfig() API 來完成，該方法在以下完整示例中展示:
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兩種形式，其中每種形式對應一個日志處理器 。
使用上下文管理器的可選的日志記錄?
有時候，我們需要暫時更改日志配置，并在執(zhí)行某些操作后將其還原。為此，上下文管理器是實現保存和恢復日志上下文的最明顯的方式。這是一個關于上下文管理器的簡單例子，它允許你在上下文管理器的作用域內更改日志記錄等級以及增加日志處理器：
import logging
import sys
class LoggingContext(object):
def __init__(self, logger, level=None, handler=None, close=True):
self.logger = logger
self.level = level
self.handler = handler
self.close = close
def __enter__(self):
if self.level is not None:
self.old_level = self.logger.level
self.logger.setLevel(self.level)
if self.handler:
self.logger.addHandler(self.handler)
def __exit__(self, et, ev, tb):
if self.level is not None:
self.logger.setLevel(self.old_level)
if self.handler:
self.logger.removeHandler(self.handler)
if self.handler and self.close:
self.handler.close()
# implicit return of None => don't swallow exceptions
如果指定上下文管理器的日志記錄等級屬性，則在上下文管理器的with語句所涵蓋的代碼中，日志記錄器的記錄等級將臨時設置為上下文管理器所配置的日志記錄等級。 如果指定上下文管理的日志處理器屬性，則該句柄在進入上下文管理器的上下文時添加到記錄器中，并在退出時被刪除。 如果你再也不需要該日志處理器時，你可以讓上下文管理器在退出上下文管理器的上下文時關閉它。
為了說明它是如何工作的，我們可以在上面添加以下代碼塊:
if __name__ == '__main__':
logger = logging.getLogger('foo')
logger.addHandler(logging.StreamHandler())
logger.setLevel(logging.INFO)
logger.info('1. This should appear just once on stderr.')
logger.debug('2. This should not appear.')
with LoggingContext(logger, level=logging.DEBUG):
logger.debug('3. This should appear once on stderr.')
logger.debug('4. This should not appear.')
h = logging.StreamHandler(sys.stdout)
with LoggingContext(logger, level=logging.DEBUG, handler=h, close=True):
logger.debug('5. This should appear twice - once on stderr and once on stdout.')
logger.info('6. This should appear just once on stderr.')
logger.debug('7. This should not appear.')
我們最初設置日志記錄器的消息等級為``INFO``，因此消息#1出現，消息#2沒有出現。在接下來的``with``代碼塊中我們暫時將消息等級變更為``DEBUG``，從而消息#3出現。在這一代碼塊退出后，日志記錄器的消息等級恢復為``INFO``，從而消息#4沒有出現。在下一個``with``代碼塊中，我們再一次將設置消息等級設置為``DEBUG``，同時添加一個將消息寫入``sys.stdout``的日志處理器。因此，消息#5在控制臺出現兩次（分別通過``stderr``和``stdout``）。在``with``語句完成后，狀態(tài)與之前一樣，因此消息#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.
在這種情況下，與預期一致，打印到``stdout``標準輸出的消息＃5不會出現。
當然，這里描述的方法可以被推廣，例如臨時附加日志記錄過濾器。 請注意，上面的代碼適用于Python 2以及Python 3。
命令行日志應用起步?
下面的示例提供了如下功能：
根據命令行參數確定日志級別
在單獨的文件中分發(fā)多條子命令，同一級別的子命令均以一致的方式記錄。
最簡單的配置用法
假定有一個命令行應用程序，用于停止、啟動或重新啟動某些服務。為了便于演示，不妨將 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.
第一個單詞是日志級別，第二個單詞是日志事件所在的模塊或包的名稱。
如果修改了日志級別，發(fā)送給日志的信息就能得以改變。如要顯示更多信息，則可：
$ 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 GUI 日志示例?
GUI 應用程序如何記錄日志，這是個常見的問題。 Qt 框架是一個流行的跨平臺 UI 框架，采用的是 PySide2 或 PyQt5 庫。
下面的例子演示了將日志寫入 Qt GUI 程序的過程。這里引入了一個簡單的 QtHandler 類，參數是一個可調用對象，其應為嵌入主線程某個“槽位”中運行的，因為GUI 的更新由主線程完成。這里還創(chuàng)建了一個工作線程，以便演示由 UI（通過人工點擊日志按鈕）和后臺工作線程（此處只是記錄級別和時間間隔均隨機生成的日志信息）將日志寫入 GUI 的過程。
該工作線程是用 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(QtHandler, self).__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(Window, self).__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()