さらにいくつかのアプローチ...
例: ロギング モジュール
おそらく必要なのは、stderr からlogging、強力なパブリッシュ/サブスクライブ モデルを持つ Python のモジュールに切り替えることだけです。
ログ レコードの作成を開始するのは簡単です。
# producer
import logging
log = logging.getLogger("myjobs") # that's all the setup you need
class MyJob(object):
def run(self):
log.info("starting job")
n = 10
for i in range(n):
log.info("%.1f%% done" % (100.0 * i / n))
log.info("work complete")
消費者側では、もう少し作業があります。残念ながら、ロガー出力の構成には、7 行のコードが必要です。;)
# consumer
import myjobs, sys, logging
if user_wants_log_output:
ch = logging.StreamHandler(sys.stderr)
ch.setLevel(logging.INFO)
formatter = logging.Formatter(
"%(asctime)s - %(name)s - %(levelname)s - %(message)s")
ch.setFormatter(formatter)
myjobs.log.addHandler(ch)
myjobs.log.setLevel(logging.INFO)
myjobs.MyJob().run()
一方、logging パッケージには驚くほど多くのものがあります。ログ データを一連のローテーション ファイル、電子メール アドレス、および Windows イベント ログに送信する必要がある場合は、問題ありません。
例: 最も単純なオブザーバー
ただし、ライブラリをまったく使用する必要はありません。オブザーバーをサポートする非常に簡単な方法は、何もしないメソッドを呼び出すことです。
# producer
class MyJob(object):
def on_progress(self, pct):
"""Called when progress is made. pct is the percent complete.
By default this does nothing. The user may override this method
or even just assign to it."""
pass
def run(self):
n = 10
for i in range(n):
self.on_progress(100.0 * i / n)
self.on_progress(100.0)
# consumer
import sys, myjobs
job = myjobs.MyJob()
job.on_progress = lambda pct: sys.stdout.write("%.1f%% done\n" % pct)
job.run()
時々、ラムダを書く代わりに、単に と言うことができますjob.on_progress = progressBar.update。これは素晴らしいことです。
これは、それが得られるのと同じくらい簡単です。1 つの欠点は、同じイベントにサブスクライブする複数のリスナーを自然にサポートしないことです。
例: C# ライクなイベント
少しのサポート コードを使用すると、Python で C# のようなイベントを取得できます。コードは次のとおりです。
# glue code
class event(object):
def __init__(self, func):
self.__doc__ = func.__doc__
self._key = ' ' + func.__name__
def __get__(self, obj, cls):
try:
return obj.__dict__[self._key]
except KeyError, exc:
be = obj.__dict__[self._key] = boundevent()
return be
class boundevent(object):
def __init__(self):
self._fns = []
def __iadd__(self, fn):
self._fns.append(fn)
return self
def __isub__(self, fn):
self._fns.remove(fn)
return self
def __call__(self, *args, **kwargs):
for f in self._fns[:]:
f(*args, **kwargs)
プロデューサーは、デコレーターを使用してイベントを宣言します。
# producer
class MyJob(object):
@event
def progress(pct):
"""Called when progress is made. pct is the percent complete."""
def run(self):
n = 10
for i in range(n+1):
self.progress(100.0 * i / n)
#consumer
import sys, myjobs
job = myjobs.MyJob()
job.progress += lambda pct: sys.stdout.write("%.1f%% done\n" % pct)
job.run()
これは上記の「単純なオブザーバー」コードとまったく同じように機能しますが、 を使用して好きなだけリスナーを追加できます+=。(C# とは異なり、イベント ハンドラーの型はありませんnew EventHandler(foo.bar)。イベントをサブスクライブするときに必要がなく、イベントを発生させる前に null をチェックする必要もありません。C# と同様に、イベントは例外をスケルチしません。)
選び方
logging必要なものがすべて揃っている場合は、それを使用してください。それ以外の場合は、自分にとって最も簡単なことを行ってください。注意すべき重要なことは、大きな外部依存を引き受ける必要がないということです。