structlogを採用した場合、アプリ側がログを書くために使うインスタンスはBoundLoggerかそれをラップしたクラスのインスタンスになります。このクラスは重要なのでstructlogのドキュメントにもこのクラスのためのページがあります。
structlogを本格的にカスタマイズして使っていく場合は、まずこのクラスを理解する必要があります。このクラスの動作と実装について解説していきます。
BoundLogger の役割
BoundLoggerは、コンテキスト変数を管理し、 .debug() などのログメソッドが呼ばれたらそれを処理して "wrapped logger" と呼ばれるクラスのメソッドを呼び出すのが役割です。まずはその役割を示す例を見てみましょう。
>>> import structlog >>> logger = structlog.get_logger() # (1) >>> log = logger.bind(a=1) # (2) >>> structlog.get_context(logger) {} >>> structlog.get_context(log) {'a': 1} >>> log.debug("hello", target="world") # (3) 2024-06-13 18:42:23 [debug ] hello a=1 target=world
(1) で structlog のデフォルト構成で BoundLogger (実際にはそれをラップしているクラス) のインスタンスを作っています。
(2) で a=1 というコンテキストをbindして、新たなBoundLoggerを作っています。これで logger と log は、同一のプロセッサーや wrapped logger を共有していますが、異なるコンテキストを持つBoundLoggerインスタンスになります。 structlog.get_context() にBoundLoggerを渡せばロガーが持っているコンテキストが異なることを確認できます。
(3) で .debug() というログメソッドを呼び出しました。このメソッドはキーワード引数とコンテキスト変数、そしてログメッセージからイベントを作り、それを processor で処理してから wrapped logger に渡します。デフォルトの wrapped logger は単に print() を実行するだけの PrintLogger です。デフォルトのprocessorが、ターミナル出力向けに色をつけたりカラムに揃えたりするようになっているので、このように綺麗な出力が得られます。
コンテキスト管理
ここからstructlogのソースコードを読んでいきましょう。BoundLoggerの主な実装はBoundLoggerBaseにあります。
WrappedLogger = Any # typing.py で定義されている class BoundLoggerBase: _logger: WrappedLogger def __init__( self, logger: WrappedLogger, processors: Iterable[Processor], context: Context, ): self._logger = logger self._processors = processors self._context = context
ここでは引数で与えられた logger (wrapped logger) と processors, context を単純に持つだけのクラスを作っています。
def bind(self, **new_values: Any) -> BoundLoggerBase: """ Return a new logger with *new_values* added to the existing ones. """ return self.__class__( self._logger, self._processors, self._context.__class__(self._context, **new_values), )
コンテキスト変数を追加した新たな BoundLogger を作って返します。 logger や processor は同一のものが利用されるので、BoundLoggerのインスタンスが分かれていてもログのフォーマットや出力は同一のものになります。
ログメソッドの提供
.debug() や .info() のようなログメソッドはBoundLoggerBaseではなく、それを継承した具象クラスで実装されます。
具象クラスには大きく二つあります。1つ目は標準ライブラリの logging.Logger と互換性の高い structlog.stdlib.BoundLogger です。次のように普通にログメソッドを定義し、 self._proxy_to_logger() に転送しています。
class BoundLogger(BoundLoggerBase): ... def debug(self, event: str | None = None, *args: Any, **kw: Any) -> Any: return self._proxy_to_logger("debug", event, *args, **kw) def info(self, event: str | None = None, *args: Any, **kw: Any) -> Any: return self._proxy_to_logger("info", event, *args, **kw) ...
もう一つの具象クラスは FilteringBoundLogger です。このクラスは実際には具象クラスではなく structlog.typing で定義されているだけの静的型で、実際にはフィルタリングするログレベルごとにクラスが生成されています。
def _make_filtering_bound_logger(min_level: int) -> type[FilteringBoundLogger]: def make_method( level: int, ) -> tuple[Callable[..., Any], Callable[..., Any]]: if level < min_level: return _nop, _anop # !!! min_level 未満のログレベルに対応するログメソッドは return None するだけのメソッドを使う ... meth.__name__ = name ameth.__name__ = f"a{name}" return meth, ameth meths: dict[str, Callable[..., Any]] = {"log": log, "alog": alog} for lvl, name in LEVEL_TO_NAME.items(): meths[name], meths[f"a{name}"] = make_method(lvl) ... return type( f"BoundLoggerFilteringAt{LEVEL_TO_NAME.get(min_level, 'Notset').capitalize()}", (BoundLoggerBase,), meths, ) # Pre-create all possible filters to make them pickleable. BoundLoggerFilteringAtNotset = _make_filtering_bound_logger(NOTSET) BoundLoggerFilteringAtDebug = _make_filtering_bound_logger(DEBUG) BoundLoggerFilteringAtInfo = _make_filtering_bound_logger(INFO) BoundLoggerFilteringAtWarning = _make_filtering_bound_logger(WARNING) BoundLoggerFilteringAtError = _make_filtering_bound_logger(ERROR) BoundLoggerFilteringAtCritical = _make_filtering_bound_logger(CRITICAL)
ちょっとトリッキーですが、現在のログレベルをチェックする if 文などがなく、指定したログレベル未満のログメソッドの実装が単に return None するだけの実装になっているので高速です。デフォルトで structlog.get_logger() した時に得られるBoundLoggerは最低ログレベルを指定しないですべてのログを出力する BoundLoggerFilteringAtNotset になります。
動的なログレベルによるフィルタリングの実装がありませんが、 processorsでフィルタリングするか、あるいは wrapped logger 自体が logging.Logger のようなレベル付きloggerになっている場合はそちらにフィルタリングを任せることになります。
