Logging クックブック¶
このページは、過去に有用であるとされていた、logging に関連するいくつものレシピを含んでいます。チュートリアルやリファレンス情報へのリンクについては その他のリソース を参照してください。
複数のモジュールで logging を使う¶
logging.getLogger('someLogger') の複数回の呼び出しは同じ logger への参照を返します。これは同じ Python インタプリタプロセス上で動いている限り、一つのモジュールの中からに限らず、モジュールをまたいでも当てはまります。同じオブジェクトへの参照という点でも正しいです。さらに、一つのモジュールの中で親 logger を定義して設定し、別のモジュールで子 logger を定義する (ただし設定はしない) ことが可能で、すべての子 logger への呼び出しは親にまで渡されます。まずはメインのモジュールです:
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()')
そして補助モジュール (auxiliary module) がこちらです:
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"')
出力はこのようになります:
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()
複数のスレッドからのロギング¶
複数スレッドからのロギングでは特別に何かをする必要はありません。 次の例はmain (初期) スレッドとそれ以外のスレッドからのロギングの例です:
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
予想した通りかもしれませんが、ログ出力が散らばっているのが分かります。 もちろん、この手法はより多くのスレッドでも上手くいきます。
複数の handler と formatter¶
logger は通常の Python オブジェクトです。 addHandler() メソッドは追加されるハンドラの個数について最小値も最大値も定めていません。時にアプリケーションがすべての深刻度のすべてのメッセージをテキストファイルに記録しつつ、同時にエラーやそれ以上のものをコンソールに出力することが役に立ちます。これを実現する方法は、単に適切なハンドラを設定するだけです。アプリケーションコードの中のログ記録の呼び出しは変更されずに残ります。少し前に取り上げた単純なモジュール式の例を少し変えるとこうなります:
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')
'application' 部分のコードは複数の handler について何も気にしていないことに注目してください。変更した箇所は新しい fh という名の handler を追加して設定したところがすべてです。
新しい handler を、異なる深刻度に対する filter と共に生成できることは、アプリケーションを書いてテストを行うときとても助けになります。デバッグ用にたくさんの print 文を使う代わりに logger.debug を使いましょう。あとで消したりコメントアウトしたりしなければならない print 文と違って、logger.debug 命令はソースコードの中にそのまま残しておいて再び必要になるまで休眠させておけます。その時必要になるのはただ logger および/または handler の深刻度の設定をいじることだけです。
複数の出力先にログを出力する¶
コンソールとファイルに、それぞれ異なる書式や条件でログを出力したいとします。 例えば DEBUG 以上のレベルのメッセージをファイルに、INFO 以上のメッセージをコンソールに出力するとします。 また、ファイルにはタイムスタンプを含め、コンソールには含めないものとします。 次のように実現できます。
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.')
これを実行すると、コンソールには以下のように出力されます:
root : INFO Jackdaws love my big sphinx of quartz.
myapp.area1 : INFO How quickly daft jumping zebras vex.
myapp.area2 : WARNING Jail zesty vixen who grabbed pay from quack.
myapp.area2 : ERROR The five boxing wizards jump quickly.
そして、ファイルには以下のように出力されるはずです:
10-22 22:19 root INFO Jackdaws love my big sphinx of quartz.
10-22 22:19 myapp.area1 DEBUG Quick zephyrs blow, vexing daft Jim.
10-22 22:19 myapp.area1 INFO How quickly daft jumping zebras vex.
10-22 22:19 myapp.area2 WARNING Jail zesty vixen who grabbed pay from quack.
10-22 22:19 myapp.area2 ERROR The five boxing wizards jump quickly.
これを見て分かる通り、DEBUG メッセージはファイルだけに出力され、その他のメッセージは両方に出力されます。
この例ではコンソールとファイルのハンドラだけを使っていますが、実際には任意の数のハンドラや組み合わせを使えます。
上のログファイル名 /tmp/myapp.log は、POSIX システムの一時ファイルの標準的な場所を使っています。
Windows では、別のディレクトリを選ぶ必要があるかもしれません。
いずれの場合も、ディレクトリが存在し、その中でファイルを作成・更新する権限があることを確認してください。
ログレベルのカスタム処理¶
しきい値以上のログレベル全てがハンドラによって処理される標準的な処理に対して、ときにはわずかに異なる振る舞いを必要とすることもあるでしょう。そのような場合はフィルタを使う必要があります。以下のような処理が必要なシナリオについて取り上げてみましょう:
深刻度
INFOとWARNINGのメッセージをsys.stdoutに送る深刻度
ERROR以上のメッセージをsys.stderrに送る深刻度
DEBUG以上のメッセージをファイルapp.logに送る
logging モジュールを下記の JSON によって構成したとしましょう:
{
"version": 1,
"disable_existing_loggers": false,
"formatters": {
"simple": {
"format": "%(levelname)-8s - %(message)s"
}
},
"handlers": {
"stdout": {
"class": "logging.StreamHandler",
"level": "INFO",
"formatter": "simple",
"stream": "ext://sys.stdout"
},
"stderr": {
"class": "logging.StreamHandler",
"level": "ERROR",
"formatter": "simple",
"stream": "ext://sys.stderr"
},
"file": {
"class": "logging.FileHandler",
"formatter": "simple",
"filename": "app.log",
"mode": "w"
}
},
"root": {
"level": "DEBUG",
"handlers": [
"stderr",
"stdout",
"file"
]
}
}
This configuration does almost what we want, except that sys.stdout would show messages
of severity ERROR and only events of this severity and higher will be tracked
as well as INFO and WARNING messages. To prevent this, we can set up a filter which
excludes those messages and add it to the relevant handler. This can be configured by
adding a filters section parallel to formatters and handlers:
{
"filters": {
"warnings_and_below": {
"()" : "__main__.filter_maker",
"level": "WARNING"
}
}
}
そして stdout のハンドラに関するセクションにフィルターを追加します:
{
"stdout": {
"class": "logging.StreamHandler",
"level": "INFO",
"formatter": "simple",
"stream": "ext://sys.stdout",
"filters": ["warnings_and_below"]
}
}
フィルターは単なる関数なので、 filter_maker (ファクトリ関数) は以下のように定義することができます:
def filter_maker(level):
level = getattr(logging, level)
def filter(record):
return record.levelno <= level
return filter
この関数は引数として与えられた文字列をログレベルの数値に変換し、ログレコード内のログレベルが指定されたレベル以下の場合に True を返す関数を返します。この例では filter_maker をコマンドラインから実行するテストスクリプト main.py で関数が定義されているため、フィルター設定上のモジュール名は __main__ になります。そのためフィルター設定の中では __main__.filter_maker となります。異なるモジュールで関数を定義した場合、この部分を変更する必要があります。
フィルターを追加した main.py の全体像は以下のようになり、これで実行できる状態になりました:
import json
import logging
import logging.config
CONFIG = '''
{
"version": 1,
"disable_existing_loggers": false,
"formatters": {
"simple": {
"format": "%(levelname)-8s - %(message)s"
}
},
"filters": {
"warnings_and_below": {
"()" : "__main__.filter_maker",
"level": "WARNING"
}
},
"handlers": {
"stdout": {
"class": "logging.StreamHandler",
"level": "INFO",
"formatter": "simple",
"stream": "ext://sys.stdout",
"filters": ["warnings_and_below"]
},
"stderr": {
"class": "logging.StreamHandler",
"level": "ERROR",
"formatter": "simple",
"stream": "ext://sys.stderr"
},
"file": {
"class": "logging.FileHandler",
"formatter": "simple",
"filename": "app.log",
"mode": "w"
}
},
"root": {
"level": "DEBUG",
"handlers": [
"stderr",
"stdout",
"file"
]
}
}
'''
def filter_maker(level):
level = getattr(logging, level)
def filter(record):
return record.levelno <= level
return filter
logging.config.dictConfig(json.loads(CONFIG))
logging.debug('A DEBUG message')
logging.info('An INFO message')
logging.warning('A WARNING message')
logging.error('An ERROR message')
logging.critical('A CRITICAL message')
これを以下のように実行します:
python main.py 2>stderr.log >stdout.log
すると期待通りの結果を得ることができます:
$ more *.log
::::::::::::::
app.log
::::::::::::::
DEBUG - A DEBUG message
INFO - An INFO message
WARNING - A WARNING message
ERROR - An ERROR message
CRITICAL - A CRITICAL message
::::::::::::::
stderr.log
::::::::::::::
ERROR - An ERROR message
CRITICAL - A CRITICAL message
::::::::::::::
stdout.log
::::::::::::::
INFO - An INFO message
WARNING - A WARNING message
設定サーバの例¶
ログ記録設定サーバを使うモジュールの例です:
import logging
import logging.config
import time
import os
# 初期設定ファイルを読む
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')
ブロックする handler を扱う¶
ときどき、logging を行っているスレッドをブロックせずに、handler が動くようにしないといけないときがあります。これは Web アプリケーションではよくあることですし、もちろん他のシナリオでも起きる話です。
動作が鈍くなるときの元凶はたいてい、開発者のコントロール外にあるいくつもの理由で (例えば、残念なパフォーマンスのメールやネットワークのインフラ) 、 SMTPHandler: が電子メールを送るのに時間がかかることです。しかし、ほとんどのネットワークをまたぐ handler はブロックする可能性があります: SocketHandler による処理でさえ、裏で DNS への問い合わせというとても遅い処理を行うことがあります (そしてこの問い合わせ処理は、 Python の層より下のあなたの手の届かない、ソケットライブラリの深いところにある可能性もあります)。
解決策の1つは、2パートに分離したアプローチを用いることです。最初のパートは、パフォーマンスが重要なスレッドからアクセスされる、 QueueHandler だけをアタッチした logger です。この logger は単に、十分大きい、あるいは無制限の容量を持ったキューに書き込むだけです。キューへの書き込みは通常すぐに完了しますが、念の為に queue.Full 例外をキャッチする必要があるかもしれません。もしパフォーマンスクリティカルなスレッドを持つライブラリの開発者であるなら、このことを (QueueHandler だけをアタッチした logger についての言及を添えて) ドキュメントに書いておきましょう。
2つ目のパートは QueueHandler の対向として作られた QueueListener です。 QueueListener はとてもシンプルで、キューと handler を渡され、内部で QueueHandler (もしくは他の LogRecord の出力元) から送られた LogRecord をキューから受け取るスレッドを起動します。 LogRecord をキューから取り出して、 handler に渡して処理させます。
分離した QueueListener クラスを持つメリットは、複数の QueueHandler に対して1つのインスタンスで logging できることです。既存の handler のスレッド利用版を使って handler ごとにスレッドを持つよりはずっとリソースにやさしくなります。
この2つのクラスを利用する例です (import は省略):
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()
実行すると次のように出力します:
MainThread: Look out!
注釈
ここまでは特に非同期コードを対象とせず、処理の遅いロギングハンドラについて説明しました。
ただし、非同期コードからログを出力する場合、ネットワークハンドラだけでなくファイルハンドラも、イベントループをブロックして問題を起こす可能性があります。
asyncio の内部でもログが出力されるためです。
アプリケーションで非同期コードを使う場合は、ブロックする処理が QueueListener スレッド内だけで実行されるよう、上記の方法を使うのがよいでしょう。
バージョン 3.5 で変更: Python 3.5 より前の QueueListener は、キューから受け取ったすべてのメッセージを、初期化時に渡されたすべてのハンドラに渡していました。
レベルによるフィルタリングは、キューへ書き込む側ですべて行うと想定していたためです。
3.5 以降では、リスナーのコンストラクタにキーワード引数 respect_handler_level=True を渡すと、この動作を変更できます。
この場合、リスナーは各メッセージのレベルとハンドラのレベルを比較し、条件を満たすメッセージだけをハンドラに渡します。
バージョン 3.14 で変更: QueueListener は with 文で開始 (および停止) できます。
例えば:
with QueueListener(que, handler) as listener:
# with ブロックに入ると、
# キューリスナーが自動的に開始する。
pass
# with ブロックを抜けると、
# キューリスナーが自動的に停止する。
ネットワーク越しの logging イベントの送受信¶
ログイベントをネットワーク越しに送信し、受信端でそれを処理したいとしましょう。 SocketHandler インスタンスを送信端の root logger にアタッチすれば、簡単に実現できます:
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.')
受信端では socketserver モジュールを使って受信プログラムを作成しておきます。簡単な実用プログラムを以下に示します:
import pickle
import logging
import logging.handlers
import socketserver
import struct
class LogRecordStreamHandler(socketserver.StreamRequestHandler):
"""ストリームで届くログ要求のハンドラー。
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 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):
"""
Simple TCP socket-based logging receiver suitable for testing.
"""
allow_reuse_address = True
def __init__(self, host='localhost',
port=logging.handlers.DEFAULT_TCP_LOGGING_PORT,
handler=LogRecordStreamHandler):
socketserver.ThreadingTCPServer.__init__(self, (host, port), handler)
self.abort = 0
self.timeout = 1
self.logname = None
def serve_until_stopped(self):
import select
abort = 0
while not abort:
rd, wr, ex = select.select([self.socket.fileno()],
[], [],
self.timeout)
if rd:
self.handle_request()
abort = self.abort
def main():
logging.basicConfig(
format='%(relativeCreated)5d %(name)-15s %(levelname)-8s %(message)s')
tcpserver = LogRecordSocketReceiver()
print('About to start TCP server...')
tcpserver.serve_until_stopped()
if __name__ == '__main__':
main()
先にサーバを起動しておき、次にクライアントを起動します。クライアント側では、コンソールには何も出力されません; サーバ側では以下のようなメッセージを目にするはずです:
About to start TCP server...
59 root INFO Jackdaws love my big sphinx of quartz.
59 myapp.area1 DEBUG Quick zephyrs blow, vexing daft Jim.
69 myapp.area1 INFO How quickly daft jumping zebras vex.
69 myapp.area2 WARNING Jail zesty vixen who grabbed pay from quack.
69 myapp.area2 ERROR The five boxing wizards jump quickly.
注意： pickle にはいくつかのシナリオでセキュリティ上の問題があります。これらが影響する場合は、 makePickle() メソッドをオーバーライドし代替シリアライズ手法を実装して、それを使うように上記のスクリプトを修正することもできます。
本番環境でロギングソケットのリスナーを実行する¶
本番環境でロギングリスナーを実行するには、Supervisor などのプロセス管理ツールが必要になる場合があります。 こちらの Gist には、上記の機能を Supervisor で実行するための最小限のファイルが含まれています。 構成は次のとおりです:
ファイル |
目的 |
|---|---|
|
試験用の環境を準備する Bash スクリプト |
|
リスナーとマルチプロセスの web アプリケーションのための設定を含む Supervisor の設定ファイル |
|
Supervisor が上記の設定で実行されていることを保証するための Bash スクリプト |
|
ログイベントを受信してファイルに記録するソケットリスナープログラム |
|
リスナーに接続されたソケットを通じてロギングを実行する簡単な web アプリケーション |
|
web アプリケーションのための JSON 設定ファイル |
|
Web アプリケーションの動作を試す Python スクリプト |
この web アプリケーションは、リクエストを処理する複数のワーカープロセスを起動する web アプリケーションサーバーである Gunicorn を使っています。ここに例として挙げた設定は、複数のワーカーがいかにして互いに衝突することなく同じログファイルに書き込みを行うことができるかを示しています --- ワーカーは全てソケットリスナーを通じてログを書き込むのです。
これらのファイルを試すためには、 POSIX 環境において以下を行なってください:
Download ZIP ボタンを押して Gist を ZIP アーカイブとしてダウンロードしてください。
アーカイブファイルをスクラッチディレクトリに展開してください。
準備のために、スクラッチディレクトリにおいて
bash prepare.shを実行してください。これにより Supervisor 関連のファイルおよびログファイルのためのrunサブディレクトリ、およびbottle,gunicornそしてsupervisorがインストールされる仮想環境を含むvenvサブディレクトリが生成されます。bash ensure_app.shを実行して Supervisor が上記の設定で実行されていることを確認してください。venv/bin/python client.pyを実行して Web アプリケーションの動作を試してください。 これによりログにレコードが書き込まれます。runサブディレクトリにあるログファイルを調べてください。最新のログは、パターンapp.log*に一致する名前のファイルにあるはずです。ログは異なるワーカープロセスによって非決定論的な形で並行に処理されるため、特定の決まった順番にはなりません。リスナーと web アプリケーションは
venv/bin/supervisorctl -c supervisor.conf shutdownを実行することでシャットダウンできます。
ありそうもないことですが、テスト環境で設定したポートが別の設定と衝突してしまった場合、設定ファイルを修正する必要があるかもしれません。
デフォルトの設定はポート 9020 の TCP ソケットを使います。 次の変更で、TCP ソケットの代わりに Unix ドメインソケットを使えます:
listener.jsonにsocketキーを追加し、使うドメインソケットのパスを指定します。 このキーがある場合、リスナーは TCP ソケットではなく、指定されたドメインソケットで待ち受けます (portキーは無視されます)。webapp.jsonのソケットハンドラの設定辞書で、hostの値をドメインソケットのパスに、portの値をnullに変更します。
コンテキスト情報をログ記録出力に付加する¶
ログ呼び出しに渡す引数に加えて、コンテキスト情報もログに含めたい場合があります。
例えばネットワークアプリケーションで、リモートクライアントのユーザー名や IP アドレスなど、クライアント固有の情報を記録したい場合です。
extra パラメータで実現できますが、この方法で毎回情報を渡すのは便利とは限りません。
接続ごとに Logger インスタンスを作りたくなるかもしれませんが、これらはガベージコレクションで回収されないため、よい方法ではありません。
Logger インスタンスの数がアプリケーションのログ記録の粒度で決まるなら、実用上は問題ありません。
しかし Logger インスタンスの数が事実上無制限になると、管理が難しくなる可能性があります。
LoggerAdapter を使ったコンテキスト情報の伝達¶
logging イベントの情報と一緒に出力されるコンテキスト情報を渡す簡単な方法は、 LoggerAdapter を使うことです。このクラスは Logger のように見えるように設計されていて、 debug(), info(), warning(), error(), exception(), critical(), log() の各メソッドを呼び出せるようになっています。これらのメソッドは対応する Logger のメソッドと同じ引数を取るので、二つの型を取り替えて使うことができます。
LoggerAdapter のインスタンスを生成する際には、 Logger インスタンスとコンテキスト情報を収めた辞書風 (dict-like) のオブジェクトを渡します。 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" を用いる利点は辞書風オブジェクトの中の値が LogRecord インスタンスの __dict__ にマージされることで、辞書風オブジェクトのキーを知っている Formatter を用意して文字列をカスタマイズするようにできることです。それ以外のメソッドが必要なとき、たとえばコンテキスト情報をメッセージの前や後ろにつなげたい場合には、 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})
これで、この adapter 経由でログした全てのイベントに対して、some_conn_id の値がログメッセージの前に追加されます。
コンテキスト情報を渡すために dict 以外のオブジェクトを使う¶
LoggerAdapter に渡すのは本物の dict でなくても構いません。 __getitem__ と __iter__ を実装していて logging が辞書のように扱えるクラスのインスタンスを利用することができます。これは (dict の値が固定されるのに対して) 値を動的に生成できるので便利です。
Filter を使ったコンテキスト情報の伝達¶
ユーザ定義の Filter を使ってログ出力にコンテキスト情報を加えることもできます。 Filter インスタンスは、渡された LogRecords を修正することができます。これにより、適切なフォーマット文字列や必要なら Formatter を使って、出力となる属性を新しく追加することも出来ます。
例えば、web アプリケーションで、処理されるリクエスト (または、少なくともその重要な部分) は、スレッドローカル (threading.local) な変数に保存して、 Filter からアクセスすることで、 LogRecord にリクエストの情報を追加できます。例えば、リモート IP アドレスやリモートユーザのユーザ名にアクセスしたいなら、上述の 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
contextvars の利用¶
Python 3.7 以降の contextvars モジュールは、threading と asyncio の両方で使えるコンテキストローカルな記憶領域を提供します。
このため、一般にスレッドローカルな記憶領域より適している場合があります。
次の例では、マルチスレッド環境で Web アプリケーションが処理するリクエストの属性など、コンテキスト情報をログに追加する方法を示します。
説明のため、同じ Python プロセス上で共通のライブラリを使っている独立した複数の web アプリケーションがあるとします。これらのアプリケーションが、クライアントの IP アドレスや HTTP リクエストメソッド、およびクライアントのユーザー名のようなコンテキスト依存の情報を追加情報として含んだ自身のログを、共通のライブラリ (と他のリクエスト処理のためのコード) から各アプリケーションのログファイルへ適切に振り向けるためにはどのようにしたら良いでしょうか？
ここで、ライブラリは以下のコードで模することができるとします:
# webapplib.py
import logging
import time
logger = logging.getLogger(__name__)
def useful():
# ライブラリから記録する代表的なイベント
logger.debug('Hello from webapplib!')
# 他のスレッドが実行されるよう少し待つ
time.sleep(0.01)
複数の web アプリケーションは2つの簡単なクラス, Request と WebApp, によって模倣することができます。これらは実際のスレッド化された web アプリケーションがどのように動作するかを模擬的にあらわしています - すなわち各リクエストはスレッドで処理される状況です:
# main.py
import argparse
from contextvars import ContextVar
import logging
import os
from random import choice
import threading
import webapplib
logger = logging.getLogger(__name__)
root = logging.getLogger()
root.setLevel(logging.DEBUG)
class Request:
"""
A simple dummy request class which just holds dummy HTTP request method,
client IP address and client username
"""
def __init__(self, method, ip, user):
self.method = method
self.ip = ip
self.user = user
# A dummy set of requests which will be used in the simulation - we'll just pick
# from this list randomly. Note that all GET requests are from 192.168.2.XXX
# addresses, whereas POST requests are from 192.16.3.XXX addresses. Three users
# are represented in the sample requests.
REQUESTS = [
Request('GET', '192.168.2.20', 'jim'),
Request('POST', '192.168.3.20', 'fred'),
Request('GET', '192.168.2.21', 'sheila'),
Request('POST', '192.168.3.21', 'jim'),
Request('GET', '192.168.2.22', 'fred'),
Request('POST', '192.168.3.22', 'sheila'),
]
# Note that the format string includes references to request context information
# such as HTTP method, client IP and username
formatter = logging.Formatter('%(threadName)-11s %(appName)s %(name)-9s %(user)-6s %(ip)s %(method)-4s %(message)s')
# Create our context variables. These will be filled at the start of request
# processing, and used in the logging that happens during that processing
ctx_request = ContextVar('request')
ctx_appname = ContextVar('appname')
class InjectingFilter(logging.Filter):
"""
A filter which injects context-specific information into logs and ensures
that only information for a specific webapp is included in its log
"""
def __init__(self, app):
self.app = app
def filter(self, record):
request = ctx_request.get()
record.method = request.method
record.ip = request.ip
record.user = request.user
record.appName = appName = ctx_appname.get()
return appName == self.app.name
class WebApp:
"""
A dummy web application class which has its own handler and filter for a
webapp-specific log.
"""
def __init__(self, name):
self.name = name
handler = logging.FileHandler(name + '.log', 'w')
f = InjectingFilter(self)
handler.setFormatter(formatter)
handler.addFilter(f)
root.addHandler(handler)
self.num_requests = 0
def process_request(self, request):
"""
This is the dummy method for processing a request. It's called on a
different thread for every request. We store the context information into
the context vars before doing anything else.
"""
ctx_request.set(request)
ctx_appname.set(self.name)
self.num_requests += 1
logger.debug('Request processing started')
webapplib.useful()
logger.debug('Request processing finished')
def main():
fn = os.path.splitext(os.path.basename(__file__))[0]
adhf = argparse.ArgumentDefaultsHelpFormatter
ap = argparse.ArgumentParser(formatter_class=adhf, prog=fn,
description='Simulate a couple of web '
'applications handling some '
'requests, showing how request '
'context can be used to '
'populate logs')
aa = ap.add_argument
aa('--count', '-c', type=int, default=100, help='How many requests to simulate')
options = ap.parse_args()
# Create the dummy webapps and put them in a list which we can use to select
# from randomly
app1 = WebApp('app1')
app2 = WebApp('app2')
apps = [app1, app2]
threads = []
# Add a common handler which will capture all events
handler = logging.FileHandler('app.log', 'w')
handler.setFormatter(formatter)
root.addHandler(handler)
# Generate calls to process requests
for i in range(options.count):
try:
# Pick an app at random and a request for it to process
app = choice(apps)
request = choice(REQUESTS)
# Process the request in its own thread
t = threading.Thread(target=app.process_request, args=(request,))
threads.append(t)
t.start()
except KeyboardInterrupt:
break
# Wait for the threads to terminate
for t in threads:
t.join()
for app in apps:
print('%s processed %s requests' % (app.name, app.num_requests))
if __name__ == '__main__':
main()
上のコードを実行すると、リクエストのおよそ半分が app1.log に、残りが app2.log に記録され、すべてのリクエストが app.log にも記録されます。
各 Web アプリケーション専用のログには、そのアプリケーションのログエントリだけが含まれます。
また、リクエストの情報は一貫して表示されます。
つまり、各ダミーリクエストの情報は常に同じログ行にまとまって現れます。
次のシェル出力で確認できます:
~/logging-contextual-webapp$ python main.py
app1 processed 51 requests
app2 processed 49 requests
~/logging-contextual-webapp$ wc -l *.log
153 app1.log
147 app2.log
300 app.log
600 total
~/logging-contextual-webapp$ head -3 app1.log
Thread-3 (process_request) app1 __main__ jim 192.168.3.21 POST Request processing started
Thread-3 (process_request) app1 webapplib jim 192.168.3.21 POST Hello from webapplib!
Thread-5 (process_request) app1 __main__ jim 192.168.3.21 POST Request processing started
~/logging-contextual-webapp$ head -3 app2.log
Thread-1 (process_request) app2 __main__ sheila 192.168.2.21 GET Request processing started
Thread-1 (process_request) app2 webapplib sheila 192.168.2.21 GET Hello from webapplib!
Thread-2 (process_request) app2 __main__ jim 192.168.2.20 GET Request processing started
~/logging-contextual-webapp$ head app.log
Thread-1 (process_request) app2 __main__ sheila 192.168.2.21 GET Request processing started
Thread-1 (process_request) app2 webapplib sheila 192.168.2.21 GET Hello from webapplib!
Thread-2 (process_request) app2 __main__ jim 192.168.2.20 GET Request processing started
Thread-3 (process_request) app1 __main__ jim 192.168.3.21 POST Request processing started
Thread-2 (process_request) app2 webapplib jim 192.168.2.20 GET Hello from webapplib!
Thread-3 (process_request) app1 webapplib jim 192.168.3.21 POST Hello from webapplib!
Thread-4 (process_request) app2 __main__ fred 192.168.2.22 GET Request processing started
Thread-5 (process_request) app1 __main__ jim 192.168.3.21 POST Request processing started
Thread-4 (process_request) app2 webapplib fred 192.168.2.22 GET Hello from webapplib!
Thread-6 (process_request) app1 __main__ jim 192.168.3.21 POST Request processing started
~/logging-contextual-webapp$ grep app1 app1.log | wc -l
153
~/logging-contextual-webapp$ grep app2 app2.log | wc -l
147
~/logging-contextual-webapp$ grep app1 app.log | wc -l
153
~/logging-contextual-webapp$ grep app2 app.log | wc -l
147
ハンドラ内でのコンテキスト情報の付与¶
Handler はそれぞれ一連のフィルタを独自に持っています。コンテキスト情報を、他のハンドラに漏らすことなく LogRecord に付加したい場合、以下に示すスクリプトのように、直接レコードを編集する代わりに新しい LogRecord を返すフィルタを使うことができます:
import copy
import logging
def filter(record: logging.LogRecord):
record = copy.copy(record)
record.user = 'jim'
return record
if __name__ == '__main__':
logger = logging.getLogger()
logger.setLevel(logging.INFO)
handler = logging.StreamHandler()
formatter = logging.Formatter('%(message)s from %(user)-8s')
handler.setFormatter(formatter)
handler.addFilter(filter)
logger.addHandler(handler)
logger.info('A log message')
複数のプロセスからの単一ファイルへのログ記録¶
logging はスレッドセーフで、一つのプロセスの複数のスレッドから単一のファイルにログを記録することはサポート されています。
一方、複数のプロセス から単一のファイルに記録することはサポート されません。
Python には、複数のプロセスによる単一ファイルへのアクセスを直列化する標準的な方法がないためです。
必要な場合は、すべてのプロセスで SocketHandler にログを送り、別のプロセスにソケットサーバーを実装して、受信したログをファイルに書き込む方法があります。
既存のプロセスの一つのスレッドを、この処理専用にすることもできます。
この節 ではこの方法を詳しく説明し、動作するソケット受信プログラムも示しています。
自分のアプリケーションに合わせて変更する際の出発点として使えます。
multiprocessing モジュールの Lock クラスを使い、複数のプロセスからファイルへのアクセスを直列化する独自のハンドラを作ることもできます。
標準ライブラリの FileHandler とそのサブクラスは multiprocessing を使っていません。
別の方法として、 Queue と QueueHandler を使って、マルチプロセスアプリケーションの1つのプロセスに全ての logging イベントを送る事ができます。次の例はこれを行う方法を示します。この例では独立した listener プロセスが他のプロセスから送られた event を受け取り、それを独自の logging 設定にしたがって保存します。この例は実装の方法の1つを示しているだけ (例えば、 listener プロセスを分離する代わりに listener スレッドを使うこともできます) ですが、これは listener とアプリケーション内の他のプロセスで完全に異なる設定を使う例になっているので、各自の要求に応じたコードを書く叩き台になるでしょう:
# 自分のコードでもこれらのインポートが必要
import logging
import logging.handlers
import multiprocessing
# 次の 2 行のインポートはこの実演用
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)
# リスナープロセスの最上位のループ。キューからログイベント
# (LogRecord) を待ち受けて処理し、
# LogRecord の代わりに None を受け取ったら終了する。
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)
# ワーカープロセスの最上位のループ。ランダムな間隔を空けて
# 10 個のイベントを記録してから終了する。
# 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)
# 実演全体を制御する。キューを作成し、リスナーを作成して開始する。
# 10 個のワーカーを作成して開始し、終了を待つ。
# 最後にキューに 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()
上のスクリプトの亜種で、logging をメインプロセスの別スレッドで行う例:
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()
こちらは特定の logger に設定を適用する例になっています。foo logger は foo サブシステムの中の全てのイベントを mplog-foo.log に保存する特別な handler を持っています。これはメインプロセス側の log 処理で使われ、(logging イベントは worker プロセス側で生成されますが) 各メッセージを適切な出力先に出力します。
concurrent.futures.ProcessPoolExecutorの利用¶
もし、ワーカープロセスを起動するために concurrent.futures.ProcessPoolExecutor を使いたいのであれば、少し違う方法でキューを作る必要があります。次のような方法の代わりに
queue = multiprocessing.Queue(-1)
次のコードを利用する必要があります
queue = multiprocessing.Manager().Queue(-1) # also works with the examples above
そして、次のようなワーカー作成コードがあったとすると:
workers = []
for i in range(10):
worker = multiprocessing.Process(target=worker_process,
args=(queue, worker_configurer))
workers.append(worker)
worker.start()
for w in workers:
w.join()
次のように変更します（最初に concurrent.futures をインポートするのを忘れないようにしましょう）:
with concurrent.futures.ProcessPoolExecutor(max_workers=10) as executor:
for i in range(10):
executor.submit(worker_process, queue, worker_configurer)
Gunicorn や uWSGI を使った Web アプリケーションのデプロイ¶
Gunicorn や uWSGI などで Web アプリケーションをデプロイすると、クライアントのリクエストを処理するために複数のワーカープロセスが作成されます。
このような環境では、Web アプリケーション内でファイルに書き込むハンドラを直接作成することは避けてください。
代わりに SocketHandler を使い、Web アプリケーションから別のプロセスのリスナーにログを送ります。
Supervisor などのプロセス管理ツールで構成できます。
詳細は 本番環境でロギングソケットのリスナーを実行する を参照してください。
ファイルをローテートする¶
ログファイルがある大きさに達したら、新しいファイルを開いてそこにログを取りたいことがあります。そのファイルをある数だけ残し、その数のファイルが生成されたらファイルを循環し、ファイルの数も大きさも制限したいこともあるでしょう。この利用パターンのために、 logging パッケージは RotatingFileHandler を提供しています:
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)
この結果として、アプリケーションのログ履歴の一部である、6 つに別れたファイルが得られます:
logging_rotatingfile_example.out
logging_rotatingfile_example.out.1
logging_rotatingfile_example.out.2
logging_rotatingfile_example.out.3
logging_rotatingfile_example.out.4
logging_rotatingfile_example.out.5
最新のファイルはいつでも logging_rotatingfile_example.out で、サイズの上限に達するたびに拡張子 .1 を付けた名前に改名されます。既にあるバックアップファイルはその拡張子がインクリメントされ (.1 が .2 になるなど)、 .6 ファイルは消去されます。
明らかに、ここでは極端な例示のためにファイルの大きさをかなり小さな値に設定しています。実際に使うときは maxBytes を適切な値に設定してください。
別の format スタイルを利用する¶
logging が Python 標準ライブラリに追加された時、メッセージを動的な内容でフォーマットする唯一の方法は % を使ったフォーマットでした。その後、 Python には新しい文字列フォーマット機構として string.Template (Python 2.4 で追加) と str.format() (Python 2.6 で追加) が加わりました。
logging は (3.2 から) この2つの追加されたフォーマット方法をサポートしています。 Formatter クラスが style という名前のオプションのキーワード引数を受け取ります。このデフォルト値は '%' ですが、他に '{' と '$' が指定可能で、それぞれのフォーマット方法に対応しています。後方互換性はデフォルト値によって維持されていますが、明示的に style 引数を指定することで、 str.format() か string.Template を使った format を指定する事ができます。次の例はこの機能を使ってみています:
>>> 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
>>>
最終的に出力されるログメッセージの format と、各メッセージを生成する部分は完全に独立していることに注意してください。次の例でわかるように、メッセージを生成する部分では % を使い続けています:
>>> 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
>>>
logging の呼び出し (logger.debug() や logger.info() など) は、ログメッセージのために位置引数しか受け取らず、キーワード引数はそれを処理するときのオプションを指定するためだけに使われます。 (例えば、 exc_info キーワード引数を使ってログを取るトレースバック情報を指定したり、 extra キーワード引数を使ってログに付与する追加のコンテキスト情報を指定します。) logging パッケージは内部で % を使って format 文字列と引数をマージしているので、 str.format() や string.Template を使って logging を呼び出す事はできません。既存の logging 呼び出しは %-format を使っているので、後方互換性のためにこの部分を変更することはできません。
しかし、{} や $ を使ってログメッセージをフォーマットする方法はあります。ログメッセージには任意のオブジェクトを format 文字列として渡すことができ、logging パッケージはそのオブジェクトに対して str() を使って実際の format 文字列を生成します。次の2つのクラスを見てください:
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)
どちらのクラスも format 文字列の代わりに利用して、 {} や $ を使って実際のログの "%(message)s", "{message}", "$message" で指定された "message" 部分を生成することができます。これは何かログを取りたいときに常に使うには使いにくいクラス名ですが、 __ (アンダースコア2つ --- gettext.gettext() やその仲間によくエイリアスとして使われるアンダースコア1つと混同しないように) などの使いやすいエイリアスを使うことができます。
上のクラスは 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() などを使います。
注意点として、この方法には大きなパフォーマンスのペナルティはありません。実際のフォーマット操作は logging の呼び出し時ではなくて、メッセージが実際に handler によって出力されるときに起こります。(出力されないならフォーマットもされません) そのため、この方法で注意しないといけないのは、追加の括弧が書式文字列だけではなく引数も囲わないといけないことだけです。これは __ が XXXMessage クラスのコンストラクタ呼び出しのシンタックスシュガーでしか無いからです。
次の例のように、 LoggerAdapter を利用して上と似たような効果を実現する方法もあります:
import logging
class Message:
def __init__(self, fmt, args):
self.fmt = fmt
self.args = args
def __str__(self):
return self.fmt.format(*self.args)
class StyleAdapter(logging.LoggerAdapter):
def log(self, level, msg, /, *args, stacklevel=1, **kwargs):
if self.isEnabledFor(level):
msg, kwargs = self.process(msg, kwargs)
self.logger.log(level, Message(msg, args), **kwargs,
stacklevel=stacklevel+1)
logger = StyleAdapter(logging.getLogger(__name__))
def main():
logger.debug('Hello, {}', 'world!')
if __name__ == '__main__':
logging.basicConfig(level=logging.DEBUG)
main()
上のスクリプトは Python 3.8 以降では Hello, world! というメッセージをログするはずです。
LogRecord のカスタマイズ¶
すべてのログイベントは LogRecord インスタンスで表されます。
ログ呼び出しが行われ、ロガーのレベルで除外されなければ、イベントの情報を格納した LogRecord が作成されます。
これはそのロガーのハンドラと、階層の上方向への伝播が無効になっているロガーまでの、祖先ロガーのハンドラに渡されます (伝播が無効なロガー自身も含みます)。
Python 3.2 より前では、レコードを作成する箇所は次の二つだけでした:
Logger.makeRecord()は、通常のログイベントの処理中に呼び出されます。 このメソッドはLogRecordを直接呼び出してインスタンスを作成していました。makeLogRecord(): LogRecord に追加される属性を含む辞書を渡して呼び出されます。これはネットワーク越しに (pickle 形式でSocketHandler経由で、あるいは JSON 形式でHTTPHandler経由で) 辞書を受け取った場合などに利用されます。
そのために LogRecord で何か特別なことをしたい場合は、次のどちらかをしなければなりません。
Logger.makeRecord()をオーバーライドする独自のLoggerサブクラスを作り、対象とするロガーが一つでもインスタンス化される前に、setLoggerClass()で登録する。
最初の方法は、複数の異なるライブラリが別々のことをしようとした場合にうまく行きません。各ライブラリが独自の Logger のサブクラスを登録しようとして、最後に登録されたライブラリが生き残ります。
2つ目の方法はほとんどのケースでうまくいきますが、たとえば LogRecord を特殊化したサブクラスを使うことなどはできません。ライブラリの開発者は利用している logger に適切な filter を設定できますが、新しい logger を作るたびに忘れずに設定しないといけなくなります。 (新しいパッケージやモジュールを追加し、モジュールレベルで次の式を実行することで、新しい logger が作られます)
logger = logging.getLogger(__name__)
これでは考えることが余計に1つ増えてしまうでしょう。開発者は、最も高いレベルのロガーに取り付けられた NullHandler に、フィルタを追加することもできますが、アプリケーション開発者が、より低いレベルに対するライブラリのロガーにハンドラを取り付けた場合、フィルタは呼び出されません --- 従って、そのハンドラからの出力はライブラリ開発者の意図を反映したものにはなりません。
Python 3.2 以降では、 LogRecord の生成は、指定できるファクトリ経由になります。ファクトリは callable で、 setLogRecordFactory() で登録でき、 getLogRecordFactory() で現在のファクトリを取得できます。ファクトリは LogRecord コンストラクタと同じシグネチャで呼び出され、 LogRecord がデフォルトのファクトリとして設定されています。
このアプローチはカスタムのファクトリが LogRecord の生成のすべての面を制御できるようにしています。たとえば、サブクラスを返したり、次のようなパターンを使って単に属性を追加することができます:
old_factory = logging.getLogRecordFactory()
def record_factory(*args, **kwargs):
record = old_factory(*args, **kwargs)
record.custom_attribute = 0xdecafbad
return record
logging.setLogRecordFactory(record_factory)
このアプローチでは複数の異なるライブラリがファクトリをチェインさせて、お互いに同じ属性を上書きしたり、標準で提供されている属性を意図せず上書きしない限りはうまくいきます。しかし、チェインのつながり全てが、全ての logging の操作に対しての実行時のオーバーヘッドになることを念頭に置き、このテクニックは Filter を利用するだけでは望む結果が得られない場合にのみ使うべきです。
QueueHandler と QueueListener のサブクラス化: ZeroMQ の例¶
QueueHandler のサブクラス化¶
QueueHandler のサブクラスを作ってメッセージを他のキュー、例えば ZeroMQ の 'publish' ソケットに送信することができます。下の例では、ソケットを別に作ってそれを handler に ('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)
もちろん同じことを別の設計でもできます。socket を作るのに必要な情報を handler に渡す例です:
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 のサブクラス化¶
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)
QueueHandler と QueueListener のサブクラス化: pynng の例¶
前の節と同様の方法で、pynng を使ったリスナーとハンドラを実装できます。
これは、ZeroMQ の精神を受け継ぐ後継とされる NNG の Python バインディングです。
以下のコードは pynng をインストールした環境で試せます。
今回は順序を変えて、リスナーから示します。
QueueListener のサブクラス化¶
# listener.py
import json
import logging
import logging.handlers
import pynng
DEFAULT_ADDR = "tcp://localhost:13232"
interrupted = False
class NNGSocketListener(logging.handlers.QueueListener):
def __init__(self, uri, /, *handlers, **kwargs):
# 中断できるようタイムアウトを設定し、
# 購読用のソケットを開く
socket = pynng.Sub0(listen=uri, recv_timeout=500)
# b'' の購読設定はすべてのトピックに一致する
topics = kwargs.pop('topics', None) or b''
socket.subscribe(topics)
# ソケットをキューとして扱う
super().__init__(socket, *handlers, **kwargs)
def dequeue(self, block):
data = None
# 中断されず、ソケットからデータを受信していない間は
# ループを続ける
while not interrupted:
try:
data = self.queue.recv(block=block)
break
except pynng.Timeout:
pass
except pynng.Closed: # Ctrl-C を押すと起こることがある
break
if data is None:
return None
# 発行者から送られたログイベントを取得する
event = json.loads(data.decode('utf-8'))
return logging.makeLogRecord(event)
def enqueue_sentinel(self):
# ソケットは実際にはキューではないため、
# この実装では使わない
pass
logging.getLogger('pynng').propagate = False
listener = NNGSocketListener(DEFAULT_ADDR, logging.StreamHandler(), topics=b'')
listener.start()
print('Press Ctrl-C to stop.')
try:
while True:
pass
except KeyboardInterrupt:
interrupted = True
finally:
listener.stop()
QueueHandler のサブクラス化¶
# sender.py
import json
import logging
import logging.handlers
import time
import random
import pynng
DEFAULT_ADDR = "tcp://localhost:13232"
class NNGSocketHandler(logging.handlers.QueueHandler):
def __init__(self, uri):
socket = pynng.Pub0(dial=uri, send_timeout=500)
super().__init__(socket)
def enqueue(self, record):
# レコードを UTF-8 でエンコードした JSON として送る
d = dict(record.__dict__)
data = json.dumps(d)
self.queue.send(data.encode('utf-8'))
def close(self):
self.queue.close()
logging.getLogger('pynng').propagate = False
handler = NNGSocketHandler(DEFAULT_ADDR)
# 出力にプロセス ID を含める
logging.basicConfig(level=logging.DEBUG,
handlers=[logging.StreamHandler(), handler],
format='%(levelname)-8s %(name)10s %(process)6s %(message)s')
levels = (logging.DEBUG, logging.INFO, logging.WARNING, logging.ERROR,
logging.CRITICAL)
logger_names = ('myapp', 'myapp.lib1', 'myapp.lib2')
msgno = 1
while True:
# ロガーとレベルをランダムに選んで記録する
level = random.choice(levels)
logger = logging.getLogger(random.choice(logger_names))
logger.log(level, 'Message no. %5d' % msgno)
msgno += 1
delay = random.random() * 2 + 0.5
time.sleep(delay)
上の二つのコードは別々のシェルで実行できます。 一つのシェルでリスナーを、別の二つのシェルで送信側を実行すると、次のような出力になります。 一つ目の送信側のシェルでは:
$ python sender.py
DEBUG myapp 613 Message no. 1
WARNING myapp.lib2 613 Message no. 2
CRITICAL myapp.lib2 613 Message no. 3
WARNING myapp.lib2 613 Message no. 4
CRITICAL myapp.lib1 613 Message no. 5
DEBUG myapp 613 Message no. 6
CRITICAL myapp.lib1 613 Message no. 7
INFO myapp.lib1 613 Message no. 8
(and so on)
二つ目の送信側のシェルでは:
$ python sender.py
INFO myapp.lib2 657 Message no. 1
CRITICAL myapp.lib2 657 Message no. 2
CRITICAL myapp 657 Message no. 3
CRITICAL myapp.lib1 657 Message no. 4
INFO myapp.lib1 657 Message no. 5
WARNING myapp.lib2 657 Message no. 6
CRITICAL myapp 657 Message no. 7
DEBUG myapp.lib1 657 Message no. 8
(and so on)
リスナーのシェルでは:
$ python listener.py
Press Ctrl-C to stop.
DEBUG myapp 613 Message no. 1
WARNING myapp.lib2 613 Message no. 2
INFO myapp.lib2 657 Message no. 1
CRITICAL myapp.lib2 613 Message no. 3
CRITICAL myapp.lib2 657 Message no. 2
CRITICAL myapp 657 Message no. 3
WARNING myapp.lib2 613 Message no. 4
CRITICAL myapp.lib1 613 Message no. 5
CRITICAL myapp.lib1 657 Message no. 4
INFO myapp.lib1 657 Message no. 5
DEBUG myapp 613 Message no. 6
WARNING myapp.lib2 657 Message no. 6
CRITICAL myapp 657 Message no. 7
CRITICAL myapp.lib1 613 Message no. 7
INFO myapp.lib1 613 Message no. 8
DEBUG myapp.lib1 657 Message no. 8
(and so on)
このように、リスナーの出力には二つの送信プロセスのログが入り交じっています。
辞書ベースで構成する例¶
次の例は辞書を使った logging の構成です。この例は Django プロジェクトのドキュメント から持ってきました。この辞書を dictConfig() に渡して設定を有効にします:
LOGGING = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'verbose': {
'format': '{levelname} {asctime} {module} {process:d} {thread:d} {message}',
'style': '{',
},
'simple': {
'format': '{levelname} {message}',
'style': '{',
},
},
'filters': {
'special': {
'()': 'project.logging.SpecialFilter',
'foo': 'bar',
},
},
'handlers': {
'console': {
'level': 'INFO',
'class': 'logging.StreamHandler',
'formatter': 'simple',
},
'mail_admins': {
'level': 'ERROR',
'class': 'django.utils.log.AdminEmailHandler',
'filters': ['special']
}
},
'loggers': {
'django': {
'handlers': ['console'],
'propagate': True,
},
'django.request': {
'handlers': ['mail_admins'],
'level': 'ERROR',
'propagate': False,
},
'myproject.custom': {
'handlers': ['console', 'mail_admins'],
'level': 'INFO',
'filters': ['special']
}
}
}
この構成方法についてのより詳しい情報は、 Django のドキュメントの 該当のセクション で見ることができます。
rotator と namer を使ってログローテートをカスタマイズする¶
namer と rotator を定義する方法の例は以下の実行可能なスクリプトに示されています。ここではログファイルを gzip により圧縮する例を示しています:
import gzip
import logging
import logging.handlers
import os
import shutil
def namer(name):
return name + ".gz"
def rotator(source, dest):
with open(source, 'rb') as f_in:
with gzip.open(dest, 'wb') as f_out:
shutil.copyfileobj(f_in, f_out)
os.remove(source)
rh = logging.handlers.RotatingFileHandler('rotated.log', maxBytes=128, backupCount=5)
rh.rotator = rotator
rh.namer = namer
root = logging.getLogger()
root.setLevel(logging.INFO)
root.addHandler(rh)
f = logging.Formatter('%(asctime)s %(message)s')
rh.setFormatter(f)
for i in range(1000):
root.info(f'Message no. {i + 1}')
このスクリプトを実行すると、6つの新しいファイルが生成され、そのうち5つは圧縮されています:
$ ls rotated.log*
rotated.log rotated.log.2.gz rotated.log.4.gz
rotated.log.1.gz rotated.log.3.gz rotated.log.5.gz
$ zcat rotated.log.1.gz
2023-01-20 02:28:17,767 Message no. 996
2023-01-20 02:28:17,767 Message no. 997
2023-01-20 02:28:17,767 Message no. 998
より手の込んだ multiprocessing の例¶
次の実際に動作する例は、logging を multiprocessing と設定ファイルを使って利用する方法を示しています。設定内容はかなりシンプルですが、より複雑な構成を実際の multiprocessing を利用するシナリオで実装する方法を示しています。
この例では、メインプロセスは listener プロセスといくつかのワーカープロセスを起動します。メイン、listener、ワーカープロセスはそれぞれ分離した設定を持っています (ワーカープロセス群は同じ設定を共有します)。この例から、メインプロセスの logging, ワーカーが QueueHandler でログを送っているところ、listener が利用する QueueListener の実装、複雑な設定、キューから受け取ったイベントを設定された handler に分配する部分を見ることができます。この設定は説明用のものですが、この例を自分のシナリオに適応させることができるでしょう。
これがそのスクリプトです。docstring とコメントで動作を説明しています:
import logging
import logging.config
import logging.handlers
from multiprocessing import Process, Queue, Event, current_process
import os
import random
import time
class MyHandler:
"""
A simple handler for logging events. It runs in the listener process and
dispatches events to loggers based on the name in the received record,
which then get dispatched, by the logging system, to the handlers
configured for those loggers.
"""
def handle(self, record):
if record.name == "root":
logger = logging.getLogger()
else:
logger = logging.getLogger(record.name)
if logger.isEnabledFor(record.levelno):
# The process name is transformed just to show that it's the listener
# doing the logging to files and console
record.processName = '%s (for %s)' % (current_process().name, record.processName)
logger.handle(record)
def listener_process(q, stop_event, config):
"""
This could be done in the main process, but is just done in a separate
process for illustrative purposes.
This initialises logging according to the specified configuration,
starts the listener and waits for the main process to signal completion
via the event. The listener is then stopped, and the process exits.
"""
logging.config.dictConfig(config)
listener = logging.handlers.QueueListener(q, MyHandler())
listener.start()
if os.name == 'posix':
# On POSIX, the setup logger will have been configured in the
# parent process, but should have been disabled following the
# dictConfig call.
# On Windows, since fork isn't used, the setup logger won't
# exist in the child, so it would be created and the message
# would appear - hence the "if posix" clause.
logger = logging.getLogger('setup')
logger.critical('Should not appear, because of disabled logger ...')
stop_event.wait()
listener.stop()
def worker_process(config):
"""
A number of these are spawned for the purpose of illustration. In
practice, they could be a heterogeneous bunch of processes rather than
ones which are identical to each other.
This initialises logging according to the specified configuration,
and logs a hundred messages with random levels to randomly selected
loggers.
A small sleep is added to allow other processes a chance to run. This
is not strictly needed, but it mixes the output from the different
processes a bit more than if it's left out.
"""
logging.config.dictConfig(config)
levels = [logging.DEBUG, logging.INFO, logging.WARNING, logging.ERROR,
logging.CRITICAL]
loggers = ['foo', 'foo.bar', 'foo.bar.baz',
'spam', 'spam.ham', 'spam.ham.eggs']
if os.name == 'posix':
# On POSIX, the setup logger will have been configured in the
# parent process, but should have been disabled following the
# dictConfig call.
# On Windows, since fork isn't used, the setup logger won't
# exist in the child, so it would be created and the message
# would appear - hence the "if posix" clause.
logger = logging.getLogger('setup')
logger.critical('Should not appear, because of disabled logger ...')
for i in range(100):
lvl = random.choice(levels)
logger = logging.getLogger(random.choice(loggers))
logger.log(lvl, 'Message no. %d', i)
time.sleep(0.01)
def main():
q = Queue()
# The main process gets a simple configuration which prints to the console.
config_initial = {
'version': 1,
'handlers': {
'console': {
'class': 'logging.StreamHandler',
'level': 'INFO'
}
},
'root': {
'handlers': ['console'],
'level': 'DEBUG'
}
}
# The worker process configuration is just a QueueHandler attached to the
# root logger, which allows all messages to be sent to the queue.
# We disable existing loggers to disable the "setup" logger used in the
# parent process. This is needed on POSIX because the logger will
# be there in the child following a fork().
config_worker = {
'version': 1,
'disable_existing_loggers': True,
'handlers': {
'queue': {
'class': 'logging.handlers.QueueHandler',
'queue': q
}
},
'root': {
'handlers': ['queue'],
'level': 'DEBUG'
}
}
# The listener process configuration shows that the full flexibility of
# logging configuration is available to dispatch events to handlers however
# you want.
# We disable existing loggers to disable the "setup" logger used in the
# parent process. This is needed on POSIX because the logger will
# be there in the child following a fork().
config_listener = {
'version': 1,
'disable_existing_loggers': True,
'formatters': {
'detailed': {
'class': 'logging.Formatter',
'format': '%(asctime)s %(name)-15s %(levelname)-8s %(processName)-10s %(message)s'
},
'simple': {
'class': 'logging.Formatter',
'format': '%(name)-15s %(levelname)-8s %(processName)-10s %(message)s'
}
},
'handlers': {
'console': {
'class': 'logging.StreamHandler',
'formatter': 'simple',
'level': 'INFO'
},
'file': {
'class': 'logging.FileHandler',
'filename': 'mplog.log',
'mode': 'w',
'formatter': 'detailed'
},
'foofile': {
'class': 'logging.FileHandler',
'filename': 'mplog-foo.log',
'mode': 'w',
'formatter': 'detailed'
},
'errors': {
'class': 'logging.FileHandler',
'filename': 'mplog-errors.log',
'mode': 'w',
'formatter': 'detailed',
'level': 'ERROR'
}
},
'loggers': {
'foo': {
'handlers': ['foofile']
}
},
'root': {
'handlers': ['console', 'file', 'errors'],
'level': 'DEBUG'
}
}
# Log some initial events, just to show that logging in the parent works
# normally.
logging.config.dictConfig(config_initial)
logger = logging.getLogger('setup')
logger.info('About to create workers ...')
workers = []
for i in range(5):
wp = Process(target=worker_process, name='worker %d' % (i + 1),
args=(config_worker,))
workers.append(wp)
wp.start()
logger.info('Started worker: %s', wp.name)
logger.info('About to create listener ...')
stop_event = Event()
lp = Process(target=listener_process, name='listener',
args=(q, stop_event, config_listener))
lp.start()
logger.info('Started listener')
# We now hang around for the workers to finish their work.
for wp in workers:
wp.join()
# Workers all done, listening can now stop.
# Logging in the parent still works normally.
logger.info('Telling listener to stop ...')
stop_event.set()
lp.join()
logger.info('All done.')
if __name__ == '__main__':
main()
SysLogHandler に送るメッセージに BOM を挿入する¶
RFC 5424 は syslog デーモンに Unicode メッセージを送る時、次の構造を要求しています: オプショナルなピュア ASCII部分、続けて UTF-8 の Byte Order Mark (BOM)、続けて UTF-8 でエンコードされた Unicode. ( 仕様の該当セクション を参照)
Python 3.1 で SysLogHandler に、 message に BOM を挿入するコードが追加されました。しかし、そのときの実装が悪くて、 message の先頭に BOM をつけてしまうのでピュアASCII 部分をその前に書くことができませんでした。
この動作は壊れているので、Python 3.2.4 以降では BOM を挿入するコードが、削除されました。書き直されるのではなく削除されたので、 RFC 5424 準拠の、BOM と、オプションのピュア ASCII部分をBOMの前に、任意の Unicode を BOM の後ろに持つ UTF-8 でエンコードされた message を生成したい場合、次の手順に従う必要があります:
SysLogHandlerのインスタンスに、次のような format 文字列を持ったFormatterインスタンスをアタッチする:'ASCII section\ufeffUnicode section'
Unicode のコードポイント U+FEFF は、UTF-8 でエンコードすると BOM --
b'\xef\xbb\xbf'-- になります。ASCII セクションを好きなプレースホルダに変更する。ただしその部分の置換結果が常に ASCII になるように注意する(UTF-8 でエンコードされてもその部分が変化しないようにする)。
Unicode セクションを任意のプレースホルダに置き換える。この部分を置換したデータに ASCII 外の文字が含まれていても、それは単に UTF-8 でエンコードされるだけです。
フォーマットされた message は SysLogHandler によって UTF-8 でエンコードされます。上のルールに従えば、RFC 5424 準拠のメッセージを生成できます。上のルールに従わない場合、logging は何もエラーを起こしませんが、message は RFC 5424 に準拠しない形で送られるので、syslog デーモン側で何かエラーが起こる可能性があります。
構造化ログを実装する¶
多くのログメッセージは人間が読むために書かれるため、機械的に処理しにくくなっていますが、場合によっては (複雑な正規表現を使ってログメッセージをパースしなくても) プログラムがパース できる 構造化されたフォーマットでメッセージを出力したい場合があります。 logging パッケージを使うと、これを簡単に実現できます。 実現する方法は幾つもありますが、次の例は JSON を使ってイベントを、機械でパースできる形にシリアライズする単純な方法です:
import json
import logging
class StructuredMessage:
def __init__(self, message, /, **kwargs):
self.message = message
self.kwargs = kwargs
def __str__(self):
return '%s >>> %s' % (self.message, json.dumps(self.kwargs))
_ = StructuredMessage # optional, to improve readability
logging.basicConfig(level=logging.INFO, format='%(message)s')
logging.info(_('message 1', foo='bar', bar='baz', num=123, fnum=123.456))
上のスクリプトを実行すると次のように出力されます:
message 1 >>> {"fnum": 123.456, "num": 123, "bar": "baz", "foo": "bar"}
要素の順序は Python のバージョンによって異なることに注意してください。
より特殊な処理が必要な場合、次の例のように、カスタムの JSON エンコーダを作ることができます:
import json
import logging
class Encoder(json.JSONEncoder):
def default(self, o):
if isinstance(o, set):
return tuple(o)
elif isinstance(o, str):
return o.encode('unicode_escape').decode('ascii')
return super().default(o)
class StructuredMessage:
def __init__(self, message, /, **kwargs):
self.message = message
self.kwargs = kwargs
def __str__(self):
s = Encoder().encode(self.kwargs)
return '%s >>> %s' % (self.message, s)
_ = StructuredMessage # optional, to improve readability
def main():
logging.basicConfig(level=logging.INFO, format='%(message)s')
logging.info(_('message 1', set_value={1, 2, 3}, snowman='\u2603'))
if __name__ == '__main__':
main()
上のスクリプトを実行すると次のように出力されます:
message 1 >>> {"snowman": "\u2603", "set_value": [1, 2, 3]}
要素の順序は Python のバージョンによって異なることに注意してください。
handler を dictConfig() を使ってカスタマイズする¶
ロギングハンドラを特定の方法でカスタマイズしたい場合、dictConfig() を使えばサブクラスを作らずに実現できることがあります。
例えば、ログファイルの所有者を設定したいとします。
POSIX では shutil.chown() で簡単に設定できますが、標準ライブラリのファイルハンドラにはその機能が組み込まれていません。
次のような通常の関数で、ハンドラの作成をカスタマイズできます。
def owned_file_handler(filename, mode='a', encoding=None, owner=None):
if owner:
if not os.path.exists(filename):
open(filename, 'a').close()
shutil.chown(filename, *owner)
return logging.FileHandler(filename, mode, encoding)
そして、 dictConfig() に渡される構成設定の中で、この関数を使って logging handler を生成するように指定することができます:
LOGGING = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'default': {
'format': '%(asctime)s %(levelname)s %(name)s %(message)s'
},
},
'handlers': {
'file':{
# 以下の値はこの辞書から取り出して削除し、
# ハンドラーの作成、レベルの設定、
# フォーマッターの設定に使う。
'()': owned_file_handler,
'level':'DEBUG',
'formatter': 'default',
# 以下の値は、ハンドラーを作成する呼び出し可能オブジェクトに
# キーワード引数として渡す。
'owner': ['pulse', 'pulse'],
'filename': 'chowntest.log',
'mode': 'w',
'encoding': 'utf-8',
},
},
'root': {
'handlers': ['file'],
'level': 'DEBUG',
},
}
この例は説明用のものですが、owner の user と group を 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':{
# 以下の値はこの辞書から取り出して削除し、
# ハンドラーの作成、レベルの設定、
# フォーマッターの設定に使う。
'()': owned_file_handler,
'level':'DEBUG',
'formatter': 'default',
# 以下の値は、ハンドラーを作成する呼び出し可能オブジェクトに
# キーワード引数として渡す。
'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 2.7、3.2 以降で使えます。
3.3 より前では、例えば os.chown() を使って所有者の変更処理を実装する必要があります。
実際には、handler を生成する関数はプロジェクト内のどこかにあるユーティリティモジュールに置くことになるでしょう。設定の中で直接関数を参照する代わりに:
'()': owned_file_handler,
次のように書くこともできます:
'()': 'ext://project.util.owned_file_handler',
project.util は関数がある実際の場所に置き換えてください。上のスクリプトでは 'ext://__main__.owned_file_handler' で動くはずです。 dictConfig() は ext:// から実際の callable を見つけます。
この例は他のファイルに対する変更を実装する例にもなっています。例えば os.chmod() を使って、同じ方法で POSIX パーミッションを設定できるでしょう。
もちろん、このアプローチは FileHandler 以外の handler 、ローテートする file handler のいずれかやその他の handler にも適用できます。
固有の書式化スタイルをアプリケーション全体で使う¶
Python 3.2 では、Formatter クラスが style という名前のオプションのキーワード引数を受け取ります。このデフォルト値は後方互換性を維持するために % となっていますが、 { か $ を指定することで、str.format() か string.Template でサポートされているのと同じ書式化のアプローチを採れます。これは最終的に出力されるログメッセージの書式化に影響を与えますが、個々のログメッセージが構築される方法とは完全に直交していることに注意してください。
ロギングの呼び出し (debug(), info() など) はログメッセージのためには位置引数しか取らず、ロギングの呼び出しを処理する方法を決定するためだけにキーワード引数を指定できます (たとえば exc_info キーワード引数はトレースバックの情報をログに含めるかどうかを指定し、 extra キーワード引数はログに付加する追加のコンテキスト情報を指定します)。ロギングパッケージは内部で %-format を使って書式文字列に可変長引数を埋め込んでいるため、 str.format() や string.Template の構文を使って直接ロギングの呼び出しを行うことはできません。既存のコードにおける全てのロギングの呼び出しは %-format 文字列を使っているため、後方互換性を維持する限りこの部分が変更されることはないでしょう。
特定のロガーに関連付ける書式スタイルへの提案がなされてきましたが、そのアプローチは同時に後方互換性の問題にぶち当たります。あらゆる既存のコードはロガーの名前を使っているでしょうし、 % 形式書式化を使っているでしょう。
あらゆるサードパーティのライブラリ、あらゆるあなたのコードの間で相互運用可能なようにロギングを行うには、書式化についての決定は、個々のログ呼び出しのレベルで行う必要があります。これは受け容れ可能な代替書式化スタイルに様々な手段の可能性を広げます。
LogRecord ファクトリを使う¶
Python 3.2 において、上述した Formatter の変更とともに、 setLogRecordFactory() 関数を使って LogRecord のサブクラスをユーザに指定することを可能にするロギングパッケージの機能拡張がありました。これにより、 getMessage() をオーバライドして た だ し い ことをする、あなた自身の手による LogRecord のサブクラスをセットすることが出来ます。このメソッドの実装は基底クラスでは msg % args 書式化をし、あなたの代替の書式化の置換が出来る場所ですが、他のコードとの相互運用性を保障するために、全ての書式化スタイルをサポートするよう注意深く行うべきであり、また、%-書式化をデフォルトで認めるべきです。基底クラスの実装がそうしているように、 str(self.msg) 呼び出しもしてください。
さらに詳しい情報は、リファレンスの setLogRecordFactory(), LogRecord を参照してください。
カスタムなメッセージオブジェクトを使う¶
あなた独自のログメッセージを構築するのに {}- および $- 書式化を使えるようにするための、もうひとつの、おそらくもっと簡単な方法があります。ロギングの際には、あなたはメッセージ書式文字列として、任意のオブジェクトを使えることを(任意のオブジェクトをメッセージに使用する より)思い出してみましょう、そしてロギングパッケージはそのオブジェクトに対して実際の書式文字列を得るために str() を呼び出すことも。以下 2 つのクラスを検討してみましょう:
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)
どちらのクラスも format 文字列の代わりに利用して、 {} や $ を使って実際のログの “%(message)s”, “{message}”, “$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)
2つめは string.Template でフォーマットする例です:
>>> __ = DollarMessage
>>> print(__('Message with $num $what', num=2, what='placeholders'))
Message with 2 placeholders
>>>
注意点として、この方法には大きなパフォーマンスのペナルティはありません。実際のフォーマット操作は logging の呼び出し時ではなくて、メッセージが実際に handler によって出力されるときに起こります。(出力されないならフォーマットもされません) そのため、この方法で注意しないといけないのは、追加の括弧が書式文字列だけではなく引数も囲わないといけないことだけです。これは上記のとおり __ が XXXMessage クラスのコンストラクタ呼び出しのシンタックスシュガーでしか無いからです。
filter を dictConfig() を使ってカスタマイズする¶
dictConfig() でフィルタを設定 できます。
ただし、一見しただけでは方法が分かりにくいため、このレシピで説明します。
標準ライブラリに含まれるフィルタクラスは Filter だけで、基底クラスとして用意されているため、多くの要件にはそのまま対応できないでしょう。
通常は Filter のサブクラスを定義し、filter() をオーバーライドします。
設定辞書のフィルタの項目では () キーに、フィルタを作成する呼び出し可能オブジェクトを指定します。
クラスを指定するのが分かりやすい方法ですが、Filter インスタンスを返すものであれば指定できます。
完全な例を示します。
import logging
import logging.config
import sys
class MyFilter(logging.Filter):
def __init__(self, param=None):
self.param = param
def filter(self, record):
if self.param is None:
allow = True
else:
allow = self.param not in record.msg
if allow:
record.msg = 'changed: ' + record.msg
return allow
LOGGING = {
'version': 1,
'filters': {
'myfilter': {
'()': MyFilter,
'param': 'noshow',
}
},
'handlers': {
'console': {
'class': 'logging.StreamHandler',
'filters': ['myfilter']
}
},
'root': {
'level': 'DEBUG',
'handlers': ['console']
},
}
if __name__ == '__main__':
logging.config.dictConfig(LOGGING)
logging.debug('hello')
logging.debug('hello - noshow')
この例は、インスタンスを構築する呼び出し可能オブジェクトに、設定データをキーワード引数として渡す方法を示しています。 上のスクリプトを実行すると、次のように表示されます:
changed: hello
設定した通りに動いていますね。
ほかにもいくつか特筆すべき点があります:
設定内で直接その callable を参照出来ない場合(例えばそれが異なるモジュール内にあり、設定辞書のある場所からそれを直接インポート出来ない、など)には、 外部オブジェクトへのアクセス に記述されている
ext://...形式を使えます。例えば、上記例のようにMyFilterと指定する代わりに、'ext://__main__.MyFilter'と記述することが出来ます。フィルタについてとともに、このテクニックは、カスタムハンドラ、カスタムフォーマッタに対しても同様に使えます。ロギングが設定において、どのようにユーザ定義のオブジェクトをサポートするのかについてのさらなる詳細については、 ユーザ定義オブジェクト と、本クックブックの上の方のレシピ handler を dictConfig() を使ってカスタマイズする を参照してください。
例外の書式化をカスタマイズする¶
例外の書式化をカスタマイズしたいことがあるでしょう - わかりやすさのために、例外情報がある場合でもログイベントごとに一行に収まることを死守したいと望むとしましょう。フォーマッタのクラスをカスタマイズして、このように出来ます:
import logging
class OneLineExceptionFormatter(logging.Formatter):
def formatException(self, exc_info):
"""
Format an exception so that it prints on a single line.
"""
result = super().formatException(exc_info)
return repr(result) # or format into one line however you want to
def format(self, record):
s = super().format(record)
if record.exc_text:
s = s.replace('\n', '') + '|'
return s
def configure_logging():
fh = logging.FileHandler('output.txt', 'w')
f = OneLineExceptionFormatter('%(asctime)s|%(levelname)s|%(message)s|',
'%d/%m/%Y %H:%M:%S')
fh.setFormatter(f)
root = logging.getLogger()
root.setLevel(logging.DEBUG)
root.addHandler(fh)
def main():
configure_logging()
logging.info('Sample message')
try:
x = 1 / 0
except ZeroDivisionError as e:
logging.exception('ZeroDivisionError: %s', e)
if __name__ == '__main__':
main()
実行してみましょう。このように正確に2行の出力を生成します:
28/01/2015 07:21:23|INFO|Sample message|
28/01/2015 07:21:23|ERROR|ZeroDivisionError: division by zero|'Traceback (most recent call last):\n File "logtest7.py", line 30, in main\n x = 1 / 0\nZeroDivisionError: division by zero'|
これは扱いとしては単純過ぎますが、例外情報をどのようにしてあなた好みの書式化出来るかを示しています。さらに特殊なニーズが必要な場合には traceback モジュールが有用です。
ロギングメッセージを喋る¶
ログメッセージを画面ではなく音声で出力したい場合もあります。
システムで音声合成 (TTS) が使えれば、Python バインディングがなくても簡単に実現できます。
多くの TTS システムにはコマンドラインプログラムがあり、ハンドラから subprocess で呼び出せます。
ここでは、TTS プログラムがユーザーとの対話を必要とせず、完了に長い時間がかからないことを前提とします。
また、ログの頻度がユーザーを圧倒するほど高くなく、メッセージを同時にではなく一つずつ読み上げてもよいものとします。
次の実装例は、一つのメッセージを読み上げ終わるまで次を処理しないため、他のハンドラを待たせる可能性があります。
espeak TTS パッケージが利用できる場合の短い例を示します。
import logging
import subprocess
import sys
class TTSHandler(logging.Handler):
def emit(self, record):
msg = self.format(record)
# 英語の女性の声でゆっくり話す
cmd = ['espeak', '-s150', '-ven+f3', msg]
p = subprocess.Popen(cmd, stdout=subprocess.PIPE,
stderr=subprocess.STDOUT)
# プログラムの終了を待つ
p.communicate()
def configure_logging():
h = TTSHandler()
root = logging.getLogger()
root.addHandler(h)
# デフォルトのフォーマッターはメッセージをそのまま返す
root.setLevel(logging.DEBUG)
def main():
logging.info('Hello')
logging.debug('Goodbye')
if __name__ == '__main__':
configure_logging()
sys.exit(main())
実行すれば、女性の声で "Hello" に続き "Goodbye" と喋るはずです。
このアプローチは、もちろんほかの TTS システムにも採用出来ますし、メッセージをコマンドライン経由で外部プログラムに渡せるようなものであれば、ほかのシステムであっても全く同じです。
ロギングメッセージをバッファリングし、条件に従って出力する¶
ログメッセージを一時的な領域に記録し、特定の条件が成立した場合にだけ出力したいことがあります。 例えば、関数内でデバッグイベントの記録を開始し、エラーなく終了した場合は、収集した情報でログを埋めたくないとします。 一方、エラーが起きた場合は、エラーとともにデバッグ情報をすべて出力したいとします。
このような振る舞いをするロギングをしたい関数に対して、デコレータを用いてこれを行う例をお見せします。それには logging.handlers.MemoryHandler を使います。これにより何か条件を満たすまでロギングイベントを溜め込むことが出来、条件を満たせば溜め込まれたイベントが flushed として他のハンドラ (target のハンドラ)に渡されます。デフォルトでは、 MemoryHandler はそのバッファが一杯になるか、指定された閾値のレベル以上のイベントが起こるとフラッシュされます。何か特別なフラッシュの振る舞いをしたければ、このレシピはさらに特殊化した MemoryHandler とともに利用出来ます。
スクリプトの例には、ログレベルを順にたどる単純な関数 foo があります。
これから記録するレベルを sys.stderr に表示してから、そのレベルで実際にログを記録します。
foo の引数に真を渡すと ERROR と CRITICAL のレベルでも記録し、それ以外では DEBUG、INFO、WARNING だけを記録します。
スクリプトが行うことは単に、 foo を必要とされている特定の条件でのロギングを行うようにするデコレータで修飾することだけです。このデコレータはパラメータとしてロガーを取り、修飾された関数が呼ばれている間だけメモリハンドラをアタッチします。追加のパラメータとして、ターゲットのハンドラ、フラッシュが発生すべきレベル、バッファの容量(バッファされたレコード数)も受け取れます。これらのデフォルトは順に 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 かそれより大きい場合にのみ行っていますが、この場合はそれよりも重要度の低い ERROR よりも前に発生したイベントも出力されます。
当然のことですが、デコレーションはいつものやり方でどうぞ:
@log_if_errors(logger)
def foo(fail=False):
...
バッファリングしながらロギングメッセージを email で送信する¶
ログメッセージをメールで、特に1つのメールにつき複数のログメッセージを、送信する方法を例示するため、 BufferingHandler を継承します。以下の例は、必要に応じて改変することもできますが、 SMTP 経由でログを送信するのに必要な情報をコマンドライン引数で指定してスクリプトを実行できるように簡単なテストハーネスも提供しています (必須の引数およびオプション引数の詳細を見るためには、ダウンロードしたスクリプトを -h 引数をつけて実行してください)。
import logging
import logging.handlers
import smtplib
class BufferingSMTPHandler(logging.handlers.BufferingHandler):
def __init__(self, mailhost, port, username, password, fromaddr, toaddrs,
subject, capacity):
logging.handlers.BufferingHandler.__init__(self, capacity)
self.mailhost = mailhost
self.mailport = port
self.username = username
self.password = password
self.fromaddr = fromaddr
if isinstance(toaddrs, str):
toaddrs = [toaddrs]
self.toaddrs = toaddrs
self.subject = subject
self.setFormatter(logging.Formatter("%(asctime)s %(levelname)-5s %(message)s"))
def flush(self):
if len(self.buffer) > 0:
try:
smtp = smtplib.SMTP(self.mailhost, self.mailport)
smtp.starttls()
smtp.login(self.username, self.password)
msg = "From: %s\r\nTo: %s\r\nSubject: %s\r\n\r\n" % (self.fromaddr, ','.join(self.toaddrs), self.subject)
for record in self.buffer:
s = self.format(record)
msg = msg + s + "\r\n"
smtp.sendmail(self.fromaddr, self.toaddrs, msg)
smtp.quit()
except Exception:
if logging.raiseExceptions:
raise
self.buffer = []
if __name__ == '__main__':
import argparse
ap = argparse.ArgumentParser()
aa = ap.add_argument
aa('host', metavar='HOST', help='SMTP server')
aa('--port', '-p', type=int, default=587, help='SMTP port')
aa('user', metavar='USER', help='SMTP username')
aa('password', metavar='PASSWORD', help='SMTP password')
aa('to', metavar='TO', help='Addressee for emails')
aa('sender', metavar='SENDER', help='Sender email address')
aa('--subject', '-s',
default='Test Logging email from Python logging module (buffering)',
help='Subject of email')
options = ap.parse_args()
logger = logging.getLogger()
logger.setLevel(logging.DEBUG)
h = BufferingSMTPHandler(options.host, options.port, options.user,
options.password, options.sender,
options.to, options.subject, 10)
logger.addHandler(h)
for i in range(102):
logger.info("Info index = %d", i)
h.flush()
h.close()
SMTP サーバーを正しく設定した上でスクリプトを実行すると、指定したアドレス宛てに11通のメールを受け取るでしょう。最初の10通のメールはそれぞれ10個のログメッセージを含み、11通目のメールは2つのログメッセージを含むはずです。これらのログメッセージはスクリプト内で指定された102個のログメッセージから構成されています。
設定によって時刻を 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 の両方に書式化するのに、それぞれのハンドラにそれぞれフォーマッタを与えています。
ロギングの選択にコンテキストマネージャを使う¶
ロギングの設定を一時的に変更し、処理後に元に戻せると便利な場合があります。 設定を保存・復元するにはコンテキストマネージャが分かりやすい方法です。 次の簡単な例では、必要に応じてロギングレベルを変更したり、ハンドラを追加したりできます。 変更はコンテキストマネージャの有効期間だけ適用されます。
import logging
import sys
class LoggingContext:
def __init__(self, logger, level=None, handler=None, close=True):
self.logger = logger
self.level = level
self.handler = handler
self.close = close
def __enter__(self):
if self.level is not None:
self.old_level = self.logger.level
self.logger.setLevel(self.level)
if self.handler:
self.logger.addHandler(self.handler)
def __exit__(self, et, ev, tb):
if self.level is not None:
self.logger.setLevel(self.old_level)
if self.handler:
self.logger.removeHandler(self.handler)
if self.handler and self.close:
self.handler.close()
# 暗黙に None を返すので、例外を握りつぶさない
レベル値を指定した場合、コンテキストマネージャがカバーする 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 が 2 回 (1回は stderr を通して、もう1回は stdout を通して) コンソールに出力されます。
with 文が完了すると、前の状態になるので (メッセージ #1 のように) メッセージ #6 が現れ、(まさにメッセージ #2 のように) メッセージ #7 は現れません。
出来上がったスクリプトを実行すると、結果は次のようになります:
$ 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 でも動くことに注意してください。
CLIアプリケーションスターターテンプレート¶
ここのサンプルでは次のことを説明します:
コマンドライン引数に応じてログレベルを使用する
複数のファイルに分割されたサブコマンドにディスパッチする。すべて一貫して同じレベルでログ出力を行う
シンプルで最小限の設定で行えるようにする
サービスを停止したり、開始したり、再起動する役割を持ったコマンドラインアプリケーションがあるとします。説明のために、アプリケーションのメインスクリプトが app.py 、個々のコマンドが start.py 、 stop.py 、 restart.py に実装されているものとします。デフォルトは logging.INFO ですが、コマンドライン引数を使ってアプリケーションのログの冗長性を制御したいとします。 app.py は次のコードのようになるでしょう:
import argparse
import importlib
import logging
import os
import sys
def main(args=None):
scriptname = os.path.basename(__file__)
parser = argparse.ArgumentParser(scriptname)
levels = ('DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL')
parser.add_argument('--log-level', default='INFO', choices=levels)
subparsers = parser.add_subparsers(dest='command',
help='Available commands:')
start_cmd = subparsers.add_parser('start', help='Start a service')
start_cmd.add_argument('name', metavar='NAME',
help='Name of service to start')
stop_cmd = subparsers.add_parser('stop',
help='Stop one or more services')
stop_cmd.add_argument('names', metavar='NAME', nargs='+',
help='Name of service to stop')
restart_cmd = subparsers.add_parser('restart',
help='Restart one or more services')
restart_cmd.add_argument('names', metavar='NAME', nargs='+',
help='Name of service to restart')
options = parser.parse_args()
# コマンドを振り分けるコードをすべてこのファイルに置くこともできるが、
# ここでは説明のため、各コマンドを別のモジュールに実装する。
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
# ここを工夫して、ファイルや辞書から設定を読み込んでもよい
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)
# ここで実際のコマンド処理を行う ...
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)
# ここで実際のコマンド処理を行う ...
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)
# ここで実際のコマンド処理を行う ...
logger.info('Restarted the %s service%s.', services, plural)
このアプリケーションをデフォルトのログレベルで実行すると、次のような出力が得られます:
$ python app.py start foo
INFO start Started the 'foo' service.
$ python app.py stop foo bar
INFO stop Stopped the 'foo' and 'bar' services.
$ python app.py restart foo bar baz
INFO restart Restarted the 'foo', 'bar' and 'baz' services.
最初のワードはログレベルで、次のワードはイベントのログ出力が行われたモジュールまたはパッケージ名です。
ログレベルを変更し、ログに出力する情報を変更できるようにしましょう。もし、より詳細な情報が必要だとしましょう:
$ python app.py --log-level DEBUG start foo
DEBUG start About to start foo
INFO start Started the 'foo' service.
$ python app.py --log-level DEBUG stop foo bar
DEBUG stop About to stop 'foo' and 'bar'
INFO stop Stopped the 'foo' and 'bar' services.
$ python app.py --log-level DEBUG restart foo bar baz
DEBUG restart About to restart 'foo', 'bar' and 'baz'
INFO restart Restarted the 'foo', 'bar' and 'baz' services.
あるいは情報を減らしたい場合もあるでしょう:
$ python app.py --log-level WARNING start foo
$ python app.py --log-level WARNING stop foo bar
$ python app.py --log-level WARNING restart foo bar baz
この場合、コマンドは WARNING 以上のレベルのログを記録しないため、コンソールには何も表示されません。
Qt GUIのログ出力¶
GUI アプリケーションにログを表示する方法は、ときどき寄せられる質問です。 Qt は広く使われているクロスプラットフォームの UI フレームワークで、PySide2 や PyQt5 ライブラリによる Python バインディングがあります。
次のサンプルはQt GUIでログ出力を行うサンプルです。ここではシンプルな QtHandler クラスを作成しています。これは呼び出し可能オブジェクトを受け取ります。これはGUI更新を行うメインスレッドの中で利用されるスロットです。ワーカースレッドも作成し、UI自身からボタンを使ってログを出力したり、バックグラウンドのタスクを行うワーカースレッドからログ出力を行います（ここではランダムな期間にランダムなレベルでメッセージを出しています）。
ワーカースレッドは threading モジュールではなく Qt の QThread クラスで実装しています。
他の Qt コンポーネントとの連携に優れた QThread を使う必要がある場合もあるためです。
このコードは PySide6、PyQt6、PySide2、PyQt5 のいずれも最近のリリースで動作するはずです。
以前の Qt に合わせて変更することもできるでしょう。
詳しくはコード内のコメントを参照してください。
import logging
import random
import sys
import time
# Qt パッケージ間の細かな違いに対応する
try:
from PySide6 import QtCore, QtGui, QtWidgets
Signal = QtCore.Signal
Slot = QtCore.Slot
except ImportError:
try:
from PyQt6 import QtCore, QtGui, QtWidgets
Signal = QtCore.pyqtSignal
Slot = QtCore.pyqtSlot
except ImportError:
try:
from PySide2 import QtCore, QtGui, QtWidgets
Signal = QtCore.Signal
Slot = QtCore.Slot
except ImportError:
from PyQt5 import QtCore, QtGui, QtWidgets
Signal = QtCore.pyqtSignal
Slot = QtCore.pyqtSlot
logger = logging.getLogger(__name__)
#
# シグナルを正しく初期化するには、QObject またはそのサブクラスの中に
# 定義する必要がある。
#
class Signaller(QtCore.QObject):
signal = Signal(str, logging.LogRecord)
#
# Qt GUI への出力はメインスレッドでのみ行う。
# そのため、このハンドラーはメインスレッドで実行するよう設定した
# スロット関数を受け取る設計になっている。この例の関数は、
# 書式化済みのログメッセージ文字列と、その生成元のログレコードを受け取る。
# 書式化済みの文字列は便宜上渡しているだけで、
# スロット関数の中で自由に出力文字列を書式化してもよい。
#
# 目的の GUI 更新を行うスロット関数を指定する。
# ハンドラーは具体的な UI 要素を知らず、意識する必要もない。
#
class QtHandler(logging.Handler):
def __init__(self, slotfunc, *args, **kwargs):
super().__init__(*args, **kwargs)
self.signaller = Signaller()
self.signaller.signal.connect(slotfunc)
def emit(self, record):
s = self.format(record)
self.signaller.signal.emit(s, record)
#
# この例は QThread を使うため、Python 側のスレッド名は
# "Dummy-1" のようになる。以下の関数は、
# 現在のスレッドの Qt 側の名前を取得する。
#
def ctname():
return QtCore.QThread.currentThread().objectName()
#
# ランダムなログレベルの生成に使う。
#
LEVELS = (logging.DEBUG, logging.INFO, logging.WARNING, logging.ERROR,
logging.CRITICAL)
#
# このワーカークラスは、メインスレッドとは別のスレッドで行う仕事を表す。
# ボタンを押すと、接続先のワーカーのスロットを介して
# スレッドに仕事を開始させる。
#
# LogRecord のデフォルトの threadName はあまり役立たないため、
# 上の方法で取得した QThread 名を qThreadName として追加する。
# この値を "extra" 辞書に入れて渡し、
# LogRecord に QThread 名を設定する。
#
# この例のワーカーは、数秒程度のランダムな間隔を空けて
# メッセージを順番に出力するだけ。
#
class Worker(QtCore.QObject):
@Slot()
def start(self):
extra = {'qThreadName': ctname() }
logger.debug('Started work', extra=extra)
i = 1
# 中断されるまでスレッドを実行する。
# これにより、スレッドを比較的きれいに終了できる。
while not QtCore.QThread.currentThread().isInterruptionRequested():
delay = 0.5 + random.random() * 2
time.sleep(delay)
try:
if random.random() < 0.1:
raise ValueError('Exception raised: %d' % i)
else:
level = random.choice(LEVELS)
logger.log(level, 'Message after delay of %3.1f: %d', delay, i, extra=extra)
except ValueError as e:
logger.exception('Failed: %s', e, extra=extra)
i += 1
#
# このクックブックの例のための、次の要素を持つ単純な UI を実装する。
#
# * 書式化済みログメッセージを表示する、読み取り専用のテキスト編集ウィンドウ
# * 別のスレッドで仕事とログ出力を開始するボタン
# * メインスレッドからログを出力するボタン
# * ログウィンドウを消去するボタン
#
class Window(QtWidgets.QWidget):
COLORS = {
logging.DEBUG: 'black',
logging.INFO: 'blue',
logging.WARNING: 'orange',
logging.ERROR: 'red',
logging.CRITICAL: 'purple',
}
def __init__(self, app):
super().__init__()
self.app = app
self.textedit = te = QtWidgets.QPlainTextEdit(self)
# プラットフォームのデフォルトの等幅フォントを設定する
f = QtGui.QFont('nosuchfont')
if hasattr(f, 'Monospace'):
f.setStyleHint(f.Monospace)
else:
f.setStyleHint(f.StyleHint.Monospace) # Qt6 用
te.setFont(f)
te.setReadOnly(True)
PB = QtWidgets.QPushButton
self.work_button = PB('Start background work', self)
self.log_button = PB('Log a message at a random level', self)
self.clear_button = PB('Clear log window', self)
self.handler = h = QtHandler(self.update_status)
# 書式文字列では threadName ではなく qThreadName を使う。
fs = '%(asctime)s %(qThreadName)-12s %(levelname)-8s %(message)s'
formatter = logging.Formatter(fs)
h.setFormatter(formatter)
logger.addHandler(h)
# 終了時に QThread も終了するよう設定する
app.aboutToQuit.connect(self.force_quit)
# すべてのウィジェットを配置する
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)
# ワーカー以外のスロットとシグナルを接続する
self.log_button.clicked.connect(self.manual_update)
self.clear_button.clicked.connect(self.clear_display)
# 新しいワーカースレッドを開始し、ワーカー用のスロットを接続する
self.start_thread()
self.work_button.clicked.connect(self.worker.start)
# 開始後はボタンを無効にする
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') # qThreadName 用
self.worker.moveToThread(self.worker_thread)
# ワーカースレッドでイベントループを開始する
self.worker_thread.start()
def kill_thread(self):
# ワーカーに停止を伝え、次に終了を指示して、
# 終了するまで待つ
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):
# ウィンドウが閉じられたときに使う
if self.worker_thread.isRunning():
self.kill_thread()
# 以下の関数は UI を更新する。スロットがメインスレッドに設定されているため、
# そこで実行される
@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):
# この関数は、渡された書式化済みメッセージに加え、
# レコード内の情報も使い、重要度 (レベル) に応じた
# 適切な色でメッセージを表示する。
level = random.choice(LEVELS)
extra = {'qThreadName': ctname() }
logger.log(level, 'Manually logged!', extra=extra)
@Slot()
def clear_display(self):
self.textedit.clear()
def main():
QtCore.QThread.currentThread().setObjectName('MainThread')
logging.getLogger().setLevel(logging.DEBUG)
app = QtWidgets.QApplication(sys.argv)
example = Window(app)
example.show()
if hasattr(app, 'exec'):
rc = app.exec()
else:
rc = app.exec_()
sys.exit(rc)
if __name__=='__main__':
main()
RFC5424 をサポートする syslog へのロギング¶
RFC 5424 は 2009 年に公開されましたが、多くの syslog サーバーはデフォルトで 2001 年の古い RFC 3164 を使う設定になっています。
2003 年に logging が Python に追加された時点では、当時唯一のプロトコルであった古い方に対応していました。
RFC 5424 の公開後も syslog サーバーへの普及が進まなかったため、SysLogHandler の機能は更新されていません。
RFC 5424 は構造化データのサポートなど、いくつかの有用な機能を持っています。その機能をサポートする syslog サーバーへのロギングを可能にする必要がある場合、以下のような派生ハンドラクラスを使うことで実現することができます:
import datetime as dt
import logging.handlers
import re
import socket
import time
class SysLogHandler5424(logging.handlers.SysLogHandler):
tz_offset = re.compile(r'([+-]\d{2})(\d{2})$')
escaped = re.compile(r'([\]"\\])')
def __init__(self, *args, **kwargs):
self.msgid = kwargs.pop('msgid', None)
self.appname = kwargs.pop('appname', None)
super().__init__(*args, **kwargs)
def format(self, record):
version = 1
asctime = dt.datetime.fromtimestamp(record.created).isoformat()
m = self.tz_offset.prefixmatch(time.strftime('%z'))
has_offset = False
if m and time.timezone:
hrs, mins = m.groups()
if int(hrs) or int(mins):
has_offset = True
if not has_offset:
asctime += 'Z'
else:
asctime += f'{hrs}:{mins}'
try:
hostname = socket.gethostname()
except Exception:
hostname = '-'
appname = self.appname or '-'
procid = record.process
msgid = '-'
msg = super().format(record)
sdata = '-'
if hasattr(record, 'structured_data'):
sd = record.structured_data
# キーが SD-ID、値が PARAM-NAME から PARAM-VALUE への辞書となる
# 辞書を指定する (これらの意味は RFC を参照)。
# ここでは説明のためエラーチェックを省略している。
# 本番環境で使う際は適切に変更できる
parts = []
def replacer(m):
g = m.groups()
return '\\' + g[0]
for sdid, dv in sd.items():
part = f'[{sdid}'
for k, v in dv.items():
s = str(v)
s = self.escaped.sub(replacer, s)
part += f' {k}="{s}"'
part += ']'
parts.append(part)
sdata = ''.join(parts)
return f'{version} {asctime} {hostname} {appname} {procid} {msgid} {sdata} {msg}'
上のコードを完全に理解するには RFC 5424 の知識が必要です。 構造化データをログに渡す方法など、要件が多少異なる場合もあるでしょう。 それでも、個別の要件に合わせて変更できるはずです。 このハンドラには、例えば次のように構造化データを渡します。
sd = {
'foo@12345': {'bar': 'baz', 'baz': 'bozz', 'fizz': r'buzz'},
'foo@54321': {'rab': 'baz', 'zab': 'bozz', 'zzif': r'buzz'}
}
extra = {'structured_data': sd}
i = 1
logger.debug('Message %d', i, extra=extra)
ロガーを出力ストリームのように取り扱う方法¶
書き込み先として file-like オブジェクトを期待するサードパーティの API に接続する必要がある一方で、その API の出力を直接ロガーに送りたいということがときどきあります。これは file-like な API でロガーをラップするクラスを使うことで実現できます。以下はそのようなクラスを例解する短いスクリプトです:
import logging
class LoggerWriter:
def __init__(self, logger, level):
self.logger = logger
self.level = level
def write(self, message):
if message != '\n': # 必要なら、改行だけの出力を避ける
self.logger.log(self.level, message)
def flush(self):
# 実際には何もしないが、ファイルライクオブジェクトに期待される場合がある。
# 必要に応じて実装する
pass
def close(self):
# 実際には何もしないが、ファイルライクオブジェクトに期待される場合がある。
# 必要に応じて実装する。フラグを設定し、
# 以降の write の呼び出しで例外を送出するようにしてもよい
pass
def main():
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger('demo')
info_fp = LoggerWriter(logger, logging.INFO)
debug_fp = LoggerWriter(logger, logging.DEBUG)
print('An INFO message', file=info_fp)
print('A DEBUG message', file=debug_fp)
if __name__ == "__main__":
main()
このスクリプトを実行すると、次のように出力されます。
INFO:demo:An INFO message
DEBUG:demo:A DEBUG message
また、 sys.stdout や sys.stderr をリダイレクトするには LoggerWriter を使って以下のようにします:
import sys
sys.stdout = LoggerWriter(logger, logging.INFO)
sys.stderr = LoggerWriter(logger, logging.WARNING)
上記の操作は、必要に応じてロギングを設定した 後に 行うべきです。上記の例では、 basicConfig() の呼び出しが (LoggerWriter インスタンスで上書きされる 前の sys.stderr を使って) 設定を行います。そして、以下のような結果を得るでしょう:
>>> print('Foo')
INFO:demo:Foo
>>> print('Bar', file=sys.stderr)
WARNING:demo:Bar
>>>
言うまでもなく上記の例は basicConfig() で使われている書式にもとづく出力を示していますが、ロギングの設定で異なるフォーマッタを使うことができます。
上記の例では、バッファリングや奪取した書き込み呼び出しのシーケンスの扱いについてはなすがままになっています。たとえば、上記の LoggerWriter の定義で、次のようなコードの断片があったとします。
sys.stderr = LoggerWriter(logger, logging.WARNING)
1 / 0
このスクリプトを実行すると以下のような結果が得られます。
WARNING:demo:Traceback (most recent call last):
WARNING:demo: File "/home/runner/cookbook-loggerwriter/test.py", line 53, in <module>
WARNING:demo:
WARNING:demo:main()
WARNING:demo: File "/home/runner/cookbook-loggerwriter/test.py", line 49, in main
WARNING:demo:
WARNING:demo:1 / 0
WARNING:demo:ZeroDivisionError
WARNING:demo::
WARNING:demo:division by zero
この出力は望ましいものではありません。
内部のコードが sys.stderr に複数回の書き込みを行い、そのたびに別のログ行になるためです (例えば上の最後の三行)。
この問題を避けるには、書き込まれた内容をバッファにため、改行が現れたときだけログ行を出力します。
LoggerWriter を少し改善してみましょう:
class BufferingLoggerWriter(LoggerWriter):
def __init__(self, logger, level):
super().__init__(logger, level)
self.buffer = ''
def write(self, message):
if '\n' not in message:
self.buffer += message
else:
parts = message.split('\n')
if self.buffer:
s = self.buffer + parts.pop(0)
self.logger.log(self.level, s)
self.buffer = parts.pop()
for part in parts:
self.logger.log(self.level, part)
この実装は改行があらわれるまでログをバッファリングし、行全体をログに出力するだけです。このアプローチにより、より適切な出力が得られます:
WARNING:demo:Traceback (most recent call last):
WARNING:demo: File "/home/runner/cookbook-loggerwriter/main.py", line 55, in <module>
WARNING:demo: main()
WARNING:demo: File "/home/runner/cookbook-loggerwriter/main.py", line 52, in main
WARNING:demo: 1/0
WARNING:demo:ZeroDivisionError: division by zero
ログ出力の改行を統一的に扱う方法¶
Usually, messages that are logged (say to console or file) consist of a single line of text. However, sometimes there is a need to handle messages with multiple lines - whether because a logging format string contains newlines, or logged data contains newlines. If you want to handle such messages uniformly, so that each line in the logged message appears uniformly formatted as if it was logged separately, you can do this using a handler mixin, as in the following snippet:
# mymixins.py モジュールにあるとする
import copy
class MultilineMixin:
def emit(self, record):
s = record.getMessage()
if '\n' not in s:
super().emit(record)
else:
lines = s.splitlines()
rec = copy.copy(record)
rec.args = None
for line in lines:
rec.msg = line
super().emit(rec)
このミックスインは次のように使えます:
import logging
from mymixins import MultilineMixin
logger = logging.getLogger(__name__)
class StreamHandler(MultilineMixin, logging.StreamHandler):
pass
if __name__ == '__main__':
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s %(levelname)-9s %(message)s',
handlers = [StreamHandler()])
logger.debug('Single line')
logger.debug('Multiple lines:\nfool me once ...')
logger.debug('Another single line')
logger.debug('Multiple lines:\n%s', 'fool me ...\ncan\'t get fooled again')
実行すると、次のように表示されます:
2025-07-02 13:54:47,234 DEBUG Single line
2025-07-02 13:54:47,234 DEBUG Multiple lines:
2025-07-02 13:54:47,234 DEBUG fool me once ...
2025-07-02 13:54:47,234 DEBUG Another single line
2025-07-02 13:54:47,234 DEBUG Multiple lines:
2025-07-02 13:54:47,234 DEBUG fool me ...
2025-07-02 13:54:47,234 DEBUG can't get fooled again
一方、ログインジェクション が気になる場合は、次の例のように改行をエスケープするフォーマッタを使えます:
import logging
logger = logging.getLogger(__name__)
class EscapingFormatter(logging.Formatter):
def format(self, record):
s = super().format(record)
return s.replace('\n', r'\n')
if __name__ == '__main__':
h = logging.StreamHandler()
h.setFormatter(EscapingFormatter('%(asctime)s %(levelname)-9s %(message)s'))
logging.basicConfig(level=logging.DEBUG, handlers = [h])
logger.debug('Single line')
logger.debug('Multiple lines:\nfool me once ...')
logger.debug('Another single line')
logger.debug('Multiple lines:\n%s', 'fool me ...\ncan\'t get fooled again')
もちろん、用途に合ったエスケープ方法を使えます。 このスクリプトを実行すると、次のように表示されます:
2025-07-09 06:47:33,783 DEBUG Single line
2025-07-09 06:47:33,783 DEBUG Multiple lines:\nfool me once ...
2025-07-09 06:47:33,783 DEBUG Another single line
2025-07-09 06:47:33,783 DEBUG Multiple lines:\nfool me ...\ncan't get fooled again
このエスケープ処理は後方互換性を損なうため、標準ライブラリのデフォルトにはできません。
避けるべきパターン¶
これまでのセクションではログ出力を行うときに必要なこと、考慮すべきことなどを説明してきました。このセクションでは、 役に立たない 利用パターンについて触れます。これは多くの場合避けるべきことです。これらの説明はどこから読んでも構いません。
同じログファイルを何度も開く¶
Windowsでは同じファイルを何度も開くことができず「このファイルは他のプロセスから利用されています」というエラーが表示されます。しかし、POSIXプラットフォームではエラーがおきることなく、同じファイルを何度も開けます。これは次のように間違って使われる可能性があります。
同じファイルを参照するファイルハンドラを複数回追加する (例えばコピー＆ペースト後の変更忘れによって)。
異なる名前を持つ、一見異なる2つのファイルを開くが、片方が他方へのシンボリックリンクとなっている。
プロセスをフォークするが、その後親プロセスと子プロセスが同じファイルへの参照を維持する。これは例えば、
multiprocessingモジュールなどを使うと発生する可能性があります。
ファイルを複数回開くことは、 一見 動作しているように見えますが、さまざまな問題を引き起こす可能性があります:
複数のスレッドやプロセスが同じファイルに書き込もうとすると、ログ出力が混ざって壊れる可能性があります。 logging は複数のスレッドによる同じハンドラインスタンスの同時使用を保護します。 しかし、同じファイルを指す別々のハンドラインスタンスを使い、二つのスレッドが同時に書き込む場合には、そのような保護はありません。
(たとえばファイルのローテーションの間に) ファイルを削除しようとすると、そのファイルを指す別の参照が残っているために、何のエラーも発しないまま失敗します。これは混乱や不要なデバッグの時間のもととなる可能性があります - ログが思いもしない場所に記録されたり、完全に失われたりします。もしくは移動したと思われていたファイルが残っていたり、ファイルサイズにもとづくローテーションが行われているにもかかわらずファイルサイズが予想外に増加したりすることもあります。
この問題を回避するには 複数のプロセスからの単一ファイルへのログ記録 で紹介したテクニックを使用してください。
ロガーをクラスの属性にするか、パラメータで渡す¶
ロガーはシングルトンであるため、一般的には意味がなく、これを行う必要が出てくることは滅多にありません。コードからは名前を使って logging.getLogger(name) 経由でロガーインスタンスにアクセスできるため、インスタンスを持って回って、インスタンス属性として保持することは意味がありません。JavaやC#といった他の言語ではよく静的クラス属性にしてます。しかし、Pythonにおいてはこのパターンはクラスではなくモジュールがソフトウェア分解の単位となっているため、無意味です。
ライブラリ内でロガーに NullHandler 以外のハンドラーを追加する¶
ハンドラ、フォーマッタ、フィルタを追加してロギングを設定するのは、アプリケーション開発者の責任です。
ライブラリ開発者の責任ではありません。
ライブラリを保守している場合、そのロガーに追加するハンドラは NullHandler インスタンスだけにしてください。
大量のロガーを作成する¶
ロガーはシングルトンであり、スクリプトの実行中に解放されることがないため、大量のロガーを作成すると、メモリが解放されることなく消費されます。ファイルの処理単位やネットワーク接続単位でロガーを作るのではなく、 既存のメカニズム を使ってコンテキスト依存の情報をログに渡し、ロガーはアプリケーション内の説明の単位（通常はモジュールだが、場合によってはそれよりも小さい可能性もある）で作るように制限してください。
その他のリソース¶
参考
loggingモジュールlogging モジュールの API リファレンス。
logging.configモジュールlogging モジュールの環境設定 API です。
logging.handlersモジュールlogging モジュールに含まれる、便利なハンドラです。