structlog.get_logger() と configure()

structlogシリーズ、今回（たぶん最終回）は get_logger() について。ソースコードを解説していく前に、まず重要な挙動から。

>>> import structlog
>>> from structlog import PrintLogger
>>> def logger_factory():
...     print("creating logger")
...     return PrintLogger()
...

>>> structlog.configure(logger_factory=logger_factory)
>>> logger = structlog.get_logger()
>>> logger
<BoundLoggerLazyProxy(logger=None, wrapper_class=None, processors=None, context_class=None, initial_values={}, logger_factory_args=())>
>>> logger.info("hello")
creating logger
2024-06-24 15:31:41 [info     ] hello
>>> logger.info("hello")
creating logger
2024-06-24 15:31:43 [info     ] hello

• logger_factory は Bound Logger に渡す wrapped logger を生成する関数
• get_logger() は Bound Logger ではなく、BoundLoggerLazyProxy というクラスのインスタンスを返す
• BoundLoggerLazyProxyのメソッドを呼び出すたびに Bound Logger が作られ、その度に logger_factory も実行される

structlog は単なるアプリのロギングだけでなく標準ライブラリの logging の代わりに使われることも考えて作られているので、構成設定前に get_logger が呼ばれることも考えてデフォルトでは毎回設定からBoundLoggerを生成するようになっています。

しかし流石にこれは効率や性能面で良くないので、 cache_logger_on_first_use というオプションを使うことをお勧めします。このオプションを有効にすると logger (BoundLoggerLazyProxy) は一度作ったBoundLoggerを再利用するようになります。

>>> structlog.configure(cache_logger_on_first_use=True)
>>> logger.info("hello")
creating logger
2024-06-24 15:56:03 [info     ] hello
>>> logger.info("hello")
2024-06-24 15:56:06 [info     ] hello

• BoundLoggerLazyProxy はデフォルトでグローバルの cache_logger_on_first_use 設定を参照する
  • get_logger() で logger が作られた後に cache_logger_on_first_use=True を設定しても良いようになっている
• ただし、異なる BoundLoggerLazyProxy 間ではこのキャッシュは共有されていない
  • 同一の logger インスタンスを複数の logger から使いたい場合は、 logger factory が何度呼ばれても同一の logger インスタンスを返すようにする

ソースコード解説

get_logger() ã‚‚ BoundLoggerLazyProxy も、 structlog._config というモジュールで実装されています。

https://github.com/hynek/structlog/blob/527f492fcdef65619ae1de9d4f9ec692ca64a81e/src/structlog/_config.py

def get_logger(*args: Any, **initial_values: Any) -> Any:
    return wrap_logger(None, logger_factory_args=args, **initial_values)

def wrap_logger(
    logger: WrappedLogger | None,
    processors: Iterable[Processor] | None = None,
    wrapper_class: type[BindableLogger] | None = None,
    context_class: type[Context] | None = None,
    cache_logger_on_first_use: bool | None = None,
    logger_factory_args: Iterable[Any] | None = None,
    **initial_values: Any,
) -> Any:

    return BoundLoggerLazyProxy(
        logger,
        wrapper_class=wrapper_class,
        processors=processors,
        context_class=context_class,
        cache_logger_on_first_use=cache_logger_on_first_use,
        initial_values=initial_values,
        logger_factory_args=logger_factory_args,
    )

• get_logger() は logger_factory_args という引数（後述）を渡しやすくなっている以外は、 wrap_logger() をデフォルト引数のまま呼んでいるだけ
• wrap_logger() は BoundLoggerLazyProxy() を生成する。グローバルの設定を引数で上書きできるようになっている。

class BoundLoggerLazyProxy:

    def __getattr__(self, name: str) -> Any:
        bl = self.bind()
        return getattr(bl, name)

    def bind(self, **new_values: Any) -> BindableLogger:
        ...
        cls = self._wrapper_class or _CONFIG.default_wrapper_class
        logger = cls(_logger, processors=procs, context=ctx)

        def finalized_bind(**new_values: Any) -> BindableLogger:
            """
            Use cached assembled logger to bind potentially new values.
            """
            if new_values:
                return logger.bind(**new_values)
            return logger

        if self._cache_logger_on_first_use is True or (
            self._cache_logger_on_first_use is None
            and _CONFIG.cache_logger_on_first_use is True
        ):
            self.bind = finalized_bind  # type: ignore[method-assign]

        return finalized_bind(**new_values)

• BoundLoggerLazyProxy はログメソッドを呼ばれるたびに bind() を実行している
• 先に明示的に log = logger.bind() しておけば、 BoundLoggerLazyProxy を回避して直接 Bound Logger のメソッドを呼び出せる

structlog.stdlibについて

structlogのデフォルトの設定はPrintLoggerを使うようになっているのですが、 structlog.stdlib モジュールの recreate_defaults() を使うとほぼ同じ出力を、 logging 経由で出力できるようになっています。

https://github.com/hynek/structlog/blob/527f492fcdef65619ae1de9d4f9ec692ca64a81e/src/structlog/stdlib.py#L57

def recreate_defaults(*, log_level: int | None = logging.NOTSET) -> None:

    if log_level is not None:
        kw = {"force": True}

        logging.basicConfig(
            format="%(message)s",
            stream=sys.stdout,
            level=log_level,
            **kw,  # type: ignore[arg-type]
        )

    _config.reset_defaults()
    _config.configure(
        processors=[
           ...  # デフォルトと似たテキストにフォーマットする設定
        ],
        wrapper_class=BoundLogger,
        logger_factory=LoggerFactory(),
    )

def get_logger(*args: Any, **initial_values: Any) -> BoundLogger:
    return _config.get_logger(*args, **initial_values)

class LoggerFactory:

    def __call__(self, *args: Any) -> logging.Logger:

        if args:
            return logging.getLogger(args[0])

        _, name = _find_first_app_frame_and_name(self._ignore)
        return logging.getLogger(name)

• stdlib.recreates_defaults() を使うと、 stdlib.BoundLogger と stdlib.LoggerFactory を使うようになる
• BoundLoggerは以前に説明した通り、 logging.Logger とシグネチャを似せて static type hint をつけた上で、ログメッセージの位置引数に対応したもの
• ただしprocessorsには PositionalArgumentsFormatter が入っていないので、位置引数はフォーマットされず、 event_dict["positional_args"] に入ったままフォーマットされる
• log_level を渡すと logging.basicConfig() もしてくれる。ログフォーマットは "%(message)s" で、すでに structlog がフォーマットしたログをそのまま表示するだけ（タイムスタンプ等はstructlogがつける）
• stdlib.get_logger() は戻り値の型ヒントが stdlib.BoundLogger になるだけで、あとは structlog.get_logger() のまま。
  • recreates_defaults() を使ってなくても型ヒントのためだけに使うことはできるが、その場合は型ヒントでは位置引数を取ることになるので注意が必要。
• stdlib.LoggerFactory は logging.getLogger() を呼ぶ。
  • structlog.get_logger() の引数が LoggerFactory に渡される仕組みがここで生きてくる。
  • structlog.get_logger("foo.bar") は logging.getLogger("foo.bar") を呼ぶ。
  • 引数を渡さない場合は呼び出し元のモジュール名を利用する。

アプリケーションでstructlogを使う場合

stdlib.LoggerFactoryやstdlib.BoundLoggerを使うかどうかはどちらでも良い。loggingとstructlogを組み合わせる別の方法は過去に書いた記事を参考に。

structlogとloggingの併用方法について - methaneのブログ

アプリケーション内でログを書く場合、ライブラリが書く場合と違ってアプリケーションの起動直後にまずロギングの設定を始められる。そうすると、 structlog.get_logger() を使う理由はあまりなくなる。 myapp.log モジュールで logger = structlog.BoundLogger(...) のように直接 BoundLogger インスタンスを作って、それ以外のモジュールは from myapp.log import logger すれば良い。

loggingを使っていた時に慣習として logging.getLogger(__name__) を使っていた場合は、structlogに移行する時にその慣習を残すかどうか考えなすと良いだろう。

• stdlib.LoggerFactory とモジュールごとの logger = structlog.get_logger(__name__) を使う
• logger = myapp.log.logger.bind(module=__name__) のようにして、構造化ログの中身に module を入れる
• 完全にこの慣習を止め、代わりに呼び出し元の場所を CallsiteParameterAdder で行数まで残す。

また、複数種類のログを残したい場合も、 structlog.get_logger() を使う代わりに直接 BoundLogger インスタンスを作ると良いだろう。そうすればログの種類ごとに出力先やフォーマットなどの設定を変えられる。例えばデバッグ用のテキストログを標準出力に出して、別にリクエストごとに構造化イベントをJSONフォーマットでfluentbit経由でBigQueryに送るなどの組み合わせが可能だ。