前回はBoundLoggerのコンテキスト管理とログメソッドについて解説しました。
今回はログメソッドが呼ばれた後の処理についてみていきます。
_proxy_to_logger()
まずはログメソッドがどうなっていたかのおさらいです。
# FilteringBoundLogger の方 (_native.py) def meth(self: Any, event: str, *args: Any, **kw: Any) -> Any: if not args: return self._proxy_to_logger(name, event, **kw) return self._proxy_to_logger(name, event % args, **kw) # stdlib.BoundLogger の方 (stdlib.py) def debug(self, event: str | None = None, *args: Any, **kw: Any) -> Any: return self._proxy_to_logger("debug", event, *args, **kw) def _proxy_to_logger( self, method_name: str, event: str | None = None, *event_args: str, **event_kw: Any, ) -> Any: if event_args: event_kw["positional_args"] = event_args return super()._proxy_to_logger(method_name, event=event, **event_kw)
BoundLoggerBase._proxy_to_logger() の第一引数にメソッド名 ("debug"), 第二引数は event という名前でログ文字列、そしてキーワード引数を渡しています。
stdlib側はloggingのLoggerと互換性を持たせるために _proxy_to_logger() をオーバーライドして、位置引数のタプルをキーワード引数 positional_args に格納しています。これはその後logging互換用のprocessorが処理に利用します。
BoundLoggerBase._proxy_to_logger() の実装は次のようになっています。
def _proxy_to_logger( self, method_name: str, event: str | None = None, **event_kw: Any ) -> Any: try: args, kw = self._process_event(method_name, event, event_kw) return getattr(self._logger, method_name)(*args, **kw) except DropEvent: return None
self._process_event() が processors を呼び出すなどの処理を行い、 args, kw を返します。
getattr(self._logger, method_name) の部分は、 self._logger が wrapped logger、 method_name がログメソッド名なので、例えば BoundLogger.debug() が呼ばれたのだとしたら self._logger.debug(*args, **kw) を呼び出すことになります。
最後の except DropEvent: ですが、 processors で動的なログのフィルタリングを行うことができて、そのログをスキップしたい場合に DropEvent を発生させることになっています。この except 句はそのスキップを行う部分になります。
_process_event()
次に self._process_event(method_name, event, event_kw) の中を見ていきましょう。
この関数の役割は method_name ("debug"など), event (ログ文字列), event_kw (キーワード引数) から、 wrapped logger に渡すための args, kw を作ることでした。 少し行数があるので、コード内にコメントで解説を書いていきます。
def _process_event( self, method_name: str, event: str | None, event_kw: dict[str, Any] ) -> tuple[Sequence[Any], Mapping[str, Any]]: # context.copy()はdictを返すが、後でprocessorがdict以外を返す可能性があるので typing としては Any を指定する。 event_dict: Any = self._context.copy() # contextから作った辞書とキーワード引数辞書をマージする event_dict.update(**event_kw) # イベント（ログ文字列）を "event" という名前で event_dict に格納する。 # processorはイベント文字列を独立した引数としてではなく、 event_dict に入った状態で受け取ることになる。 if event is not None: event_dict["event"] = event # processors をループで適用していく。 for proc in self._processors: event_dict = proc(self._logger, method_name, event_dict) # 最後の processors の出力が文字列かバイト列なら、それを単独の引数として wrapped logger に渡す if isinstance(event_dict, (str, bytes, bytearray)): return (event_dict,), {} # タプルであれば、それは (*args, **kw) の2要素タプルとして wrapped logger に渡す if isinstance(event_dict, tuple): return event_dict # dictであれば、それをキーワード引数とし、位置引数はなしとして wrapped logger に渡す if isinstance(event_dict, dict): return (), event_dict # それ以外の形であればエラーにする msg = ( "Last processor didn't return an appropriate value. " "Valid return values are a dict, a tuple of (args, kwargs), bytes, or a str." ) raise ValueError(msg)
processor の書き方
Processorを正しく設定したり独自のProcessorを作る場合のためにドキュメントがあります。 _proxy_to_logger() と _process_event() の実装を見てきたので、このドキュメントを完全に理解できるはずです。
- processorは wrapped_logger, method_name, event_dict を受け取り、 event_dict を返す
- 最初の processor に渡される event_dict は dict 型になっている
- ログメソッドに渡されたログメッセージ文字列は、（あれば） event_dict["event"] に入っている
- 途中の processor に渡される event_dict は、その前に処理された processor の戻り値である
- 最後の processor は wrapped logger の引数を組み立てる役割を持つ
- ログのフィルタリングをするなら、 DropEvent を raise する
- processororsの中のなるべく早い場所でフィルタリングすれば、後続のprocessorsの処理をまるごとスキップできる
