structlogのwrapped loggerとrendererについて

前回までの記事でstructlogの中心となるBoundLoggerの役割とコードについて解説しました。

ざっくりとシーケンス図にまとめると次のようになっています。

sequenceDiagram
  participant U as App
  box Bound Logger
  participant B as BoundLogger
  participant P as _processors
  participant W as wrapped logger
  end
  U->>B: logger.debug()
  loop
    B->>+P: event_dict
    P-->>-B: event_dict
  end
  B->>+P: event_dict
  P-->>-B: (args, kw)
  B->>W: _logger.debug(*args, **kw)

processorの出力が次のprocessorの入力になり、最後のprocessorの出力がwrapped loggerの入力になっているのが重要なポイントです。この最後のprocessorのことをrendererとも呼びます。

今回はwrapped loggerとrendererの関係を見ていきます。

WriteLoggerとLogFmtRenderer

wrapped logger とそのための最終processorのシンプルな組み合わせとして WriteLogger と LogFmtRenderer を見ていきます。まずはWriteLoggerから。

(WriteLogger)

class WriteLogger:

    def msg(self, message: str) -> None:
        with self._lock:
            self._write(message + "\n")
            self._flush()

    log = debug = info = warn = warning = msg
    fatal = failure = err = error = critical = exception = msg

.debug() 等のログメソッドを用意していますが、それらは全部単一の実装を共有しています。その単一の実装は、受け取ったメッセージに改行をつけてファイルに書き出すだけのシンプルなものです。

メッセージ本体の中にログレベルを入れる等の処理はprocessorsの中で全て終わっているので、WriteLoggerの実装は単に書くだけになっています。そのため、rendererの役割は event_dict を1つのテキストメッセージに変換することです。次にLogFmtRendererを見ていきましょう。

(LogFmtRenderer)

    def __call__(
        self, _: WrappedLogger, __: str, event_dict: EventDict
    ) -> str:
        elements: list[str] = []
        for key, value in self._ordered_items(event_dict):
            # ...
            # bool を true/false にしたり必要ならクォート＆エスケープしたりの処理
            # ...
            elements.append(f"{key}={value}")

        return " ".join(elements)

event_dict から key=value をスペース区切りで繋げたテキストを作って返していますね。このprocessorはevent_dictの文字列化だけに特化していて、ログレベルやタイムスタンプをevent_dictに入れるのはTimeStamperやadd_log_levelといった他のprocessorに任せることで、再利用性を高めています。

logging.Logger と stdlib.render_to_log_kwargs

BoundLoggerからの出力先になるwrapped loggerとして、標準ライブラリの logging.Logger を使う事ができます。 logging.Logger の debug() メソッドは次のような実装になっています。

logging.Logger.debug

    def debug(self, msg, *args, **kwargs):
        """
        Log 'msg % args' with severity 'DEBUG'.
        """
        if self.isEnabledFor(DEBUG):
            self._log(DEBUG, msg, args, **kwargs)

このメソッドの引数を作るための renderer は structlog.stdlib.render_to_log_kwargs になります。

structlog.stdlib.render_to_log_kwargs

def render_to_log_kwargs(
    _: logging.Logger, __: str, event_dict: EventDict
) -> EventDict:

    return {
        "msg": event_dict.pop("event"),
        "extra": event_dict,
        **{
            kw: event_dict.pop(kw)
            for kw in ("exc_info", "stack_info", "stacklevel")
            if kw in event_dict
        },
    }

(args, kw) のペアではなくてキーワード引数のための辞書だけを返していますね。この場合はBoundLoggerは (), **kw を引数として wrapped logger を呼び出しますが、 "msg" というキーワード引数が Logger.debug() の第一引数にマッピングされるのでちゃんとログメッセージが伝わっています。しかし、標準ライブラリの logging.Logger では logger.debug("hello, %s", "world") みたいに追加の位置引数でログ文字列を作れていたのに、この呼び出し方では第二引数以降の位置引数がありませんね？

PositionalArgumentsFormatter

既存の logging を使ったコードを structlog に置き換えたい時には、 structlog.stdlib.BoundLogger を使います。このBoundLoggerの実装は以前の記事で紹介しましたが、改めて書いておきます。

    def debug(self, event: str | None = None, *args: Any, **kw: Any) -> Any:
        """
        Process event and call `logging.Logger.debug` with the result.
        """
        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:
        """
        Propagate a method call to the wrapped logger.

        This is the same as the superclass implementation, except that
        it also preserves positional arguments in the ``event_dict`` so
        that the stdlib's support for format strings can be used.
        """
        if event_args:
            event_kw["positional_args"] = event_args

        return super()._proxy_to_logger(method_name, event=event, **event_kw)

stdlib.BoundLogger は logging.Logger との互換性のために第二引数以降の位置引数 *event_args を受け取ります。そしてそれを event_kw["positional_args"] に格納しています。 event_kw["positional_args"] を処理するためのProcessorとして、 PositionalArgumentsFormatter があります。

    def __call__(
        self, _: WrappedLogger, __: str, event_dict: EventDict
    ) -> EventDict:
        args = event_dict.get("positional_args")  # !!!

        # Mimic the formatting behaviour of the stdlib's logging module, which
        # accepts both positional arguments and a single dict argument. The
        # "single dict" check is the same one as the stdlib's logging module
        # performs in LogRecord.__init__().
        if args:
            if len(args) == 1 and isinstance(args[0], dict) and args[0]:
                args = args[0]

            event_dict["event"] %= args  # !!!

        if self.remove_positional_args and args is not None:
            del event_dict["positional_args"]

        return event_dict

重要な部分に !!! というコメントをつけておきました。 event_dict から positional_args を取り出して、 event_dict["event"] %= args しています。

PositionalArgumentsFormatterを使えば、既存のlogging.Loggerを使った logger.debug("hello, %s", "world") のようなログを含むコードのloggerをstructlog.stdlib.BoundLoggerに置き換えたとき、出力先(wrapped logger)が logging.Logger であれ structlog の WriterLogger であれ位置引数を使った文字列フォーマットに対応できます。

ProcessorFormatter

wrapped logger に logging.Logger を使う場合、 PositionalArgumentsFormatter を使う代わりに、 logging.Logger.StreamHandler の Formatterとして structlog.stdlib.ProcessorFormatter を使う方法もあります。

このクラスを使うと logging.Logger と structlog をより高度に連携させられるのですが、構成が複雑になるので今回は（多分今後も）詳しくは説明しません。 structlog のドキュメントにある次の章を頑張って読み解いてみてください。

www.structlog.org

また、過去に書いたこの記事では、 "Rendering Using structlog-based Formatters Within logging" が ProcessorFormatter を利用する方法を指していました。

methane.hatenablog.jp