Pythonのロギングライブラリ「Loguru」を試す

GitHubレポジトリ
https://github.com/Delgan/loguru

 Loguru: Pythonのログ記録を（驚くほど）簡単にLoguruは、Pythonでのログ記録を楽しめるようにすることを目的としたライブラリです。
ログの設定が面倒で、ついprint()を使ってしまったことはありませんか？… 私もそうです。しかし、ログ記録はアプリケーションにとって基本的な機能であり、デバッグプロセスを大いに助けてくれます。Loguruを使用すれば、ログ記録を最初から導入しない理由はありません。from loguru import loggerと書くだけで始められます。
また、Loguruは、標準のロガーの問題点を解決する便利な機能を追加することで、Pythonのログ記録をより簡単にすることを目指しています。アプリケーションでログを活用するのは自動的に行うべきことであり、Loguruはそれを「快適」かつ「強力」にしようとしています。

 特徴すぐに使える: 面倒な設定なしでそのまま使える
シンプルな設計: ハンドラー、フォーマッター、フィルターが不要。1つの関数で全てを処理
便利なファイルログ機能: 回転（rotation）/ 保持（retention）/ 圧縮（compression）が簡単
最新の文字列フォーマット: 中括弧スタイルによるモダンな書き方
例外のキャッチ: スレッドやメインプロセス内の例外も捕捉可能
カラフルで見やすいログ出力: 見た目が美しいログ表示
安全性: 非同期、スレッドセーフ、マルチプロセスセーフ
詳細な例外情報: 説明が豊富な例外情報の提供
構造化されたログ記録: 必要に応じたログ構造の提供
遅延評価: 高コストな関数の評価を遅延させて効率化
カスタマイズ可能なレベル: ログレベルの自由なカスタマイズ
優れた日時管理: 日時の取り扱いが簡単
スクリプトやライブラリに最適: 大規模アプリケーションでも小規模スクリプトでも利用可能
標準のloggingと互換性あり: 既存のコードとの統合もスムーズ
環境変数を通じたデフォルト設定のパーソナライズ: 柔軟なカスタマイズが可能
便利なパーサー: ログ解析が簡単
豊富な通知機能: あらゆる通知ニーズに対応

kun432

インストール

ドキュメントはこちら

ロギングなのでなるべく標準的な環境で試したい。ローカルのMacでやる。

作業ディレクトリ作成

mkdir loguru-test && cd loguru-test

仮想環境作成。自分はmiseを使う。

mise use python@3.12

cat << 'EOS' >> .mise.toml

[env]
_.python.venv = { path = ".venv", create = true }
EOS

mise trust

loguruをインストール

pip install loguru

pip freeze | grep -i loguru

出力

loguru==0.7.3

Take the tourに従って進める。

kun432

基本

Loguruのメインコンセプトは、ロガーは1つだけということ。

sample.py

from loguru import logger


def main() -> None:
  logger.debug("これだけ！美しくシンプルなロギング！")


if __name__ == '__main__':
  main()

python sample.py

デフォルトでは標準エラー出力に出力される

出力

2024-12-18 22:38:27.608 | DEBUG    | __main__:main:5 - これだけ！美しくシンプルなロギング！

kun432

ハンドラ・フィルタ・フォーマッタ

標準のloggingモジュールにあるようなハンドラ、フィルタ、フォーマッタはすべてadd()で定義する。

まずハンドラとログレベル

sample.py

import sys
from loguru import logger

# デフォルトではstderrハンドラに出力されるので、これを削除してゼロから設定する
logger.remove()

# DEBUGレベル以上をstderrハンドラに
logger.add(sys.stderr, level="DEBUG")

# INFOレベル以上をstdoutハンドラに
logger.add(sys.stdout, level="INFO")

# WARNINGレベル以上をファイルに
logger.add("WARNING.log", level="WARNING")


def main() -> None:
  logger.debug("これはDEBUGです")
  logger.info("これはINFOです")
  logger.warning("これはWARNです")


if __name__ == '__main__':
  main()

標準出力と標準エラー出力を分けてファイルに書き出して確認

python sample.py > stdout.log 2> stderr.log

ファイルをそれぞれ見てみる。

stdout.log

2024-12-18 22:55:06.217 | INFO     | __main__:main:19 - これはINFOです
2024-12-18 22:55:06.217 | WARNING  | __main__:main:20 - これはWARNです

stderr.log

2024-12-18 22:55:06.217 | DEBUG    | __main__:main:18 - これはDEBUGです
2024-12-18 22:55:06.217 | INFO     | __main__:main:19 - これはINFOです
2024-12-18 22:55:06.217 | WARNING  | __main__:main:20 - これはWARNです

WARNING.log

2024-12-18 22:55:06.217 | WARNING  | __main__:main:20 - これはWARNです

次にフォーマッタ。

sample.py

import sys
from loguru import logger

logger.remove()
logger.add(sys.stdout, format="time={time},level={level},message={message}")


def main() -> None:
  logger.info("これはINFOです")


if __name__ == '__main__':
  main()

python sample.py

出力

time=2024-12-18T23:01:41.805654+0900,level=INFO,message=これはINFOです

フォーマッタで使えるキーは以下にある

最後にフィルタ

sample.py

from loguru import logger
from func import func

logger.remove()
logger.add("func.log", filter="func", level="INFO")
logger.add("main.log", level="INFO")


def main() -> None:
  logger.info("main()が呼ばれました")
  logger.info("func()を呼びます")
  func()
  logger.info("main()が終了しました")


if __name__ == '__main__':
  main()

func.py

from loguru import logger

def func() -> None:
  logger.info("func()が呼ばれました")
  logger.info("func()が終了しました")

python sample.py

main.log

2024-12-18 23:22:24.924 | INFO     | __main__:main:10 - main()が呼ばれました
2024-12-18 23:22:24.924 | INFO     | __main__:main:11 - func()を呼びます
2024-12-18 23:22:24.924 | INFO     | func:func:4 - func()が呼ばれました
2024-12-18 23:22:24.924 | INFO     | func:func:5 - func()が終了しました
2024-12-18 23:22:24.924 | INFO     | __main__:main:13 - main()が終了しました

func.log

2024-12-18 23:22:24.924 | INFO     | func:func:4 - func()が呼ばれました
2024-12-18 23:22:24.924 | INFO     | func:func:5 - func()が終了しました

kun432

ログファイル

ファイルの場合はファイル名に日時をつけることができる。

実際にこんな書き方するのかな？とりあえずサンプルということで。

sample.py

from loguru import logger
import datetime
import time

def main() -> None:
  while True:
    logger.remove()
    logger.add("log_{time}.log", level="INFO")
    logger.add("log_{time:YYYYMMDD_HHmm}.log", level="INFO")
    now = datetime.datetime.now()
    logger.info(f"ログ出力しました {now}")
    time.sleep(10)

if __name__ == '__main__':
  main()

python sample.py

しばらく回していると以下のようなファイルが出力される。

ls *.log

出力

log_2024-12-19_00-08-34_877330.log	log_2024-12-19_00-09-24_967736.log
log_2024-12-19_00-08-44_888062.log	log_2024-12-19_00-09-34_990885.log
log_2024-12-19_00-08-54_905035.log	log_2024-12-19_00-09-45_020018.log
log_2024-12-19_00-09-04_916360.log	log_20241219_0008.log
log_2024-12-19_00-09-14_941098.log	log_20241219_0009.log

for i in $(ls *.log); do echo "=== ${i} ==="; cat ${i}; done

出力

=== log_2024-12-19_00-08-34_877330.log ===
2024-12-19 00:08:34.882 | INFO     | __main__:main:11 - ログ出力しました 2024-12-19 00:08:34.882147
=== log_2024-12-19_00-08-44_888062.log ===
2024-12-19 00:08:44.899 | INFO     | __main__:main:11 - ログ出力しました 2024-12-19 00:08:44.899681
=== log_2024-12-19_00-08-54_905035.log ===
2024-12-19 00:08:54.910 | INFO     | __main__:main:11 - ログ出力しました 2024-12-19 00:08:54.910239
=== log_2024-12-19_00-09-04_916360.log ===
2024-12-19 00:09:04.938 | INFO     | __main__:main:11 - ログ出力しました 2024-12-19 00:09:04.938965
=== log_2024-12-19_00-09-14_941098.log ===
2024-12-19 00:09:14.962 | INFO     | __main__:main:11 - ログ出力しました 2024-12-19 00:09:14.962304
=== log_2024-12-19_00-09-24_967736.log ===
2024-12-19 00:09:24.986 | INFO     | __main__:main:11 - ログ出力しました 2024-12-19 00:09:24.986628
=== log_2024-12-19_00-09-34_990885.log ===
2024-12-19 00:09:35.012 | INFO     | __main__:main:11 - ログ出力しました 2024-12-19 00:09:35.012859
=== log_2024-12-19_00-09-45_020018.log ===
2024-12-19 00:09:45.025 | INFO     | __main__:main:11 - ログ出力しました 2024-12-19 00:09:45.025537
=== log_20241219_0008.log ===
2024-12-19 00:08:34.882 | INFO     | __main__:main:11 - ログ出力しました 2024-12-19 00:08:34.882147
2024-12-19 00:08:44.899 | INFO     | __main__:main:11 - ログ出力しました 2024-12-19 00:08:44.899681
2024-12-19 00:08:54.910 | INFO     | __main__:main:11 - ログ出力しました 2024-12-19 00:08:54.910239
=== log_20241219_0009.log ===
2024-12-19 00:09:04.938 | INFO     | __main__:main:11 - ログ出力しました 2024-12-19 00:09:04.938965
2024-12-19 00:09:14.962 | INFO     | __main__:main:11 - ログ出力しました 2024-12-19 00:09:14.962304
2024-12-19 00:09:24.986 | INFO     | __main__:main:11 - ログ出力しました 2024-12-19 00:09:24.986628
2024-12-19 00:09:35.012 | INFO     | __main__:main:11 - ログ出力しました 2024-12-19 00:09:35.012859
2024-12-19 00:09:45.025 | INFO     | __main__:main:11 - ログ出力しました 2024-12-19 00:09:45.025537

log_{time:YYYYMMDD_HHmm}.logのほうだけど、こちらはログファイルがすでに存在している場合は追記されているのがわかる。

時間のフォーマットはこちら。もちろんフォーマッタでも使える。

ファイルのローテーションも設定できる。

sample.py

from loguru import logger
import datetime
import time

logger.remove()
logger.add("file.log", rotation="100 B", level="INFO")

def main() -> None:
  while True:
    now = datetime.datetime.now()
    logger.info(f"ログ出力しました {now}")
    time.sleep(10)

if __name__ == '__main__':
  main()

python sample.py

しばらく回すとこんな感じでログが生成される。

ls -lt *.log

出力

-rw-r--r--@ 1 kun432  staff  108 12 19 00:16 file.log
-rw-r--r--@ 1 kun432  staff  108 12 19 00:16 file.2024-12-19_00-16-14_596354.log
-rw-r--r--@ 1 kun432  staff  108 12 19 00:16 file.2024-12-19_00-16-04_591944.log
-rw-r--r--@ 1 kun432  staff  108 12 19 00:15 file.2024-12-19_00-15-54_579975.log

ローテーションされているのがわかる。

ローテーションはファイルサイズ・時刻・期間等で指定ができる。また、ローテーション済みファイルを圧縮するcompressionや保持期間を指定するretentionなども指定できる。

kun432

文字列フォーマットを使ったログ出力

format string的な書き方もできる。

sample.py

import sys
from loguru import logger
import datetime

logger.remove()
logger.add(sys.stdout, level="INFO")

def main() -> None:
  dt_today = datetime.datetime.now().date()
  dt_time = datetime.datetime.now().time()
  logger.info("{} - {} : {text}", dt_today, dt_time, text="ログ出力しました")

if __name__ == '__main__':
  main()

python sample.py

出力


```text:出力
2024-12-19 00:45:47.279 | INFO     | __main__:main:11 - 2024-12-19 - 00:45:47.279184 : ログ出力しました

kun432

例外をキャッチしてログ出力

スレッドやメインプロセスで発生した例外をログに記録することができる。

catch()デコレータの例

sample.py

from loguru import logger
import threading


# メインプロセス内で例外をキャッチ
@logger.catch
def divide_numbers(a, b):
    return a / b


# スレッドで例外をキャッチ
@logger.catch
def thread_function():
    logger.info("スレッドが開始されました")
    divide_numbers(10, 0)  # この例外がキャッチされる
    logger.info("スレッドが終了しました")


if __name__ == "__main__":
    # メインプロセスのログを設定
    logger.remove()
    logger.add("file.log", format="{time} {level} {message}", level="DEBUG")

    # スレッドを起動
    thread = threading.Thread(target=thread_function)
    thread.start()

    # メインプロセス内で例外を発生させる
    divide_numbers(5, 0)

    # スレッドの終了を待機
    thread.join()

    logger.info("プログラムが終了しました")

python sample.py

file.log

2024-12-19T00:55:22.409347+0900 INFO スレッドが開始されました
2024-12-19T00:55:22.409426+0900 ERROR An error has been caught in function '<module>', process 'MainProcess' (63145), thread 'MainThread' (8628752192):
Traceback (most recent call last):

> File "/Users/kun432/work/loguru-test/sample.py", line 29, in <module>
    divide_numbers(5, 0)
    └ <function divide_numbers at 0x100338f40>

  File "/Users/kun432/work/loguru-test/sample.py", line 8, in divide_numbers
    return a / b
           │   └ 0
           └ 5

ZeroDivisionError: division by zero
2024-12-19T00:55:22.409470+0900 ERROR An error has been caught in function 'thread_function', process 'MainProcess' (63145), thread 'Thread-1 (thread_function)' (6186610688):
Traceback (most recent call last):

  File "/Users/kun432/.local/share/mise/installs/python/3.12.7/lib/python3.12/threading.py", line 1032, in _bootstrap
    self._bootstrap_inner()
    │    └ <function Thread._bootstrap_inner at 0x1004867a0>
    └ <Thread(Thread-1 (thread_function), started 6186610688)>
  File "/Users/kun432/.local/share/mise/installs/python/3.12.7/lib/python3.12/threading.py", line 1075, in _bootstrap_inner
    self.run()
    │    └ <function Thread.run at 0x100486480>
    └ <Thread(Thread-1 (thread_function), started 6186610688)>
  File "/Users/kun432/.local/share/mise/installs/python/3.12.7/lib/python3.12/threading.py", line 1012, in run
    self._target(*self._args, **self._kwargs)
    │    │        │    │        │    └ {}
    │    │        │    │        └ <Thread(Thread-1 (thread_function), started 6186610688)>
    │    │        │    └ ()
    │    │        └ <Thread(Thread-1 (thread_function), started 6186610688)>
    │    └ <function thread_function at 0x1009a4180>
    └ <Thread(Thread-1 (thread_function), started 6186610688)>

> File "/Users/kun432/work/loguru-test/sample.py", line 15, in thread_function
    divide_numbers(10, 0)  # この例外がキャッチされる
    └ <function divide_numbers at 0x100338f40>

  File "/Users/kun432/work/loguru-test/sample.py", line 8, in divide_numbers
    return a / b
           │   └ 0
           └ 10

ZeroDivisionError: division by zero
2024-12-19T00:55:22.410559+0900 INFO スレッドが終了しました
2024-12-19T00:55:22.410605+0900 INFO プログラムが終了しました

context managerの例

from loguru import logger
import threading

# スレッドで例外をキャッチ
def thread_function():
    logger.info("スレッドが開始されました")
    # コンテキストマネージャで例外をキャッチ
    with logger.catch():
        result = 10 / 0
    logger.info("スレッドが終了しました")

if __name__ == "__main__":
    logger.remove()
    logger.add("file.log", format="{time} {level} {message}", level="DEBUG")

    # メインプロセス内で例外をキャッチ
    logger.info("メインプロセスが開始されました")
    with logger.catch():
        result = 5 / 0

    # スレッドを起動
    thread = threading.Thread(target=thread_function)
    thread.start()
    thread.join()

    logger.info("プログラムが終了しました")

python sample.py

file.log

2024-12-19T01:02:01.145102+0900 INFO メインプロセスが開始されました
2024-12-19T01:02:01.145216+0900 ERROR An error has been caught in function '<module>', process 'MainProcess' (63294), thread 'MainThread' (8628752192):
Traceback (most recent call last):

> File "/Users/kun432/work/loguru-test/sample.py", line 19, in <module>
    result = 5 / 0

ZeroDivisionError: division by zero
2024-12-19T01:02:01.145687+0900 INFO スレッドが開始されました
2024-12-19T01:02:01.145739+0900 ERROR An error has been caught in function 'thread_function', process 'MainProcess' (63294), thread 'Thread-1 (thread_function)' (6144487424):
Traceback (most recent call last):

  File "/Users/kun432/.local/share/mise/installs/python/3.12.7/lib/python3.12/threading.py", line 1032, in _bootstrap
    self._bootstrap_inner()
    │    └ <function Thread._bootstrap_inner at 0x102cb67a0>
    └ <Thread(Thread-1 (thread_function), started 6144487424)>
  File "/Users/kun432/.local/share/mise/installs/python/3.12.7/lib/python3.12/threading.py", line 1075, in _bootstrap_inner
    self.run()
    │    └ <function Thread.run at 0x102cb6480>
    └ <Thread(Thread-1 (thread_function), started 6144487424)>
  File "/Users/kun432/.local/share/mise/installs/python/3.12.7/lib/python3.12/threading.py", line 1012, in run
    self._target(*self._args, **self._kwargs)
    │    │        │    │        │    └ {}
    │    │        │    │        └ <Thread(Thread-1 (thread_function), started 6144487424)>
    │    │        │    └ ()
    │    │        └ <Thread(Thread-1 (thread_function), started 6144487424)>
    │    └ <function thread_function at 0x102b64e00>
    └ <Thread(Thread-1 (thread_function), started 6144487424)>

> File "/Users/kun432/work/loguru-test/sample.py", line 9, in thread_function
    result = 10 / 0

ZeroDivisionError: division by zero
2024-12-19T01:02:01.146559+0900 INFO スレッドが終了しました
2024-12-19T01:02:01.146600+0900 INFO プログラムが終了しました

kun432

カラーリング

ターミナル上でカラーリングを設定できる。

自分の環境だと、ほぼデフォルトでもこうなってる。

sample.py

import sys
from loguru import logger

logger.remove()
logger.add(sys.stdout)


def main() -> None:
  logger.debug("これだけ！美しくシンプルなロギング！")


if __name__ == '__main__':
  main()

python sample.py

マークアップ用のタグを使って、カスタイマイズしてみる。

sample.py

import sys
from loguru import logger


logger.remove()
logger.add(
  sys.stdout,
  colorize=True,
  format="<green>{time}</green> <level>{message}</level>"
)

def main() -> None:
  logger.debug("これだけ！美しくシンプルなロギング！")


if __name__ == '__main__':
  main()

python sample.py

タグについては以下。

kun432

非同期・スレッド・マルチプロセスでのログ出力

Loguruはデフォルトでスレッドセーフだが、マルチプロセスや非同期の場合はenqueue=Trueを指定することで、安全にログ出力できる。

マルチスレッドの場合は何も気にしなくてよい。

sample.py

from loguru import logger
import threading

logger.remove()
logger.add("thread.log")

def thread_worker(task_id):
    logger.info(f"スレッド {task_id} を開始")
    for i in range(1,6):
        logger.debug(f"スレッド{task_id}: ステップ{i}を実行中")
    logger.info(f"スレッド{task_id}を終了")

if __name__ == "__main__":
    threads = [threading.Thread(target=thread_worker, args=(i,)) for i in range(1,4)]

    # スレッドを開始
    for t in threads:
        t.start()

    # スレッドの終了を待機
    for t in threads:
        t.join()

    logger.info("すべてのスレッドが終了")

python sample.py

thread.log

2024-12-19 15:36:57.639 | INFO     | __main__:thread_worker:8 - スレッド 1 を開始
2024-12-19 15:36:57.639 | INFO     | __main__:thread_worker:8 - スレッド 2 を開始
2024-12-19 15:36:57.639 | DEBUG    | __main__:thread_worker:10 - スレッド1: ステップ1を実行中
2024-12-19 15:36:57.639 | INFO     | __main__:thread_worker:8 - スレッド 3 を開始
2024-12-19 15:36:57.639 | DEBUG    | __main__:thread_worker:10 - スレッド2: ステップ1を実行中
2024-12-19 15:36:57.639 | DEBUG    | __main__:thread_worker:10 - スレッド1: ステップ2を実行中
2024-12-19 15:36:57.639 | DEBUG    | __main__:thread_worker:10 - スレッド3: ステップ1を実行中
2024-12-19 15:36:57.639 | DEBUG    | __main__:thread_worker:10 - スレッド2: ステップ2を実行中
2024-12-19 15:36:57.639 | DEBUG    | __main__:thread_worker:10 - スレッド1: ステップ3を実行中
2024-12-19 15:36:57.639 | DEBUG    | __main__:thread_worker:10 - スレッド3: ステップ2を実行中
2024-12-19 15:36:57.639 | DEBUG    | __main__:thread_worker:10 - スレッド2: ステップ3を実行中
2024-12-19 15:36:57.639 | DEBUG    | __main__:thread_worker:10 - スレッド1: ステップ4を実行中
2024-12-19 15:36:57.639 | DEBUG    | __main__:thread_worker:10 - スレッド3: ステップ3を実行中
2024-12-19 15:36:57.639 | DEBUG    | __main__:thread_worker:10 - スレッド2: ステップ4を実行中
2024-12-19 15:36:57.639 | DEBUG    | __main__:thread_worker:10 - スレッド1: ステップ5を実行中
2024-12-19 15:36:57.639 | DEBUG    | __main__:thread_worker:10 - スレッド3: ステップ4を実行中
2024-12-19 15:36:57.639 | DEBUG    | __main__:thread_worker:10 - スレッド2: ステップ5を実行中
2024-12-19 15:36:57.640 | INFO     | __main__:thread_worker:11 - スレッド1を終了
2024-12-19 15:36:57.640 | DEBUG    | __main__:thread_worker:10 - スレッド3: ステップ5を実行中
2024-12-19 15:36:57.640 | INFO     | __main__:thread_worker:11 - スレッド2を終了
2024-12-19 15:36:57.640 | INFO     | __main__:thread_worker:11 - スレッド3を終了
2024-12-19 15:36:57.640 | INFO     | __main__:<module>:24 - すべてのスレッドが終了

マルチプロセスの場合はenqueue=Trueを付与する。

sample.py

from loguru import logger
from multiprocessing import Process

logger.remove()
logger.add("process.log", enqueue=True)  # enqueueを有効

def worker(task_id):
    logger.info(f"プロセス{task_id}を開始")
    for i in range(1,6):
        logger.debug(f"プロセス{task_id}: ステップ{i}を実行中")
    logger.info(f"プロセス{task_id}を終了")

if __name__ == "__main__":
    processes = [Process(target=worker, args=(i,)) for i in range(1,4)]

    # プロセスを開始
    for p in processes:
        p.start()

    # プロセスの終了を待機
    for p in processes:
        p.join()

    logger.info("すべてのプロセスが終了")

python sample.py

process.log

2024-12-19 15:48:54.331 | INFO     | __mp_main__:worker:9 - プロセス1を開始
2024-12-19 15:48:54.332 | DEBUG    | __mp_main__:worker:12 - プロセス1: ステップ1を実行中
2024-12-19 15:48:54.331 | INFO     | __mp_main__:worker:9 - プロセス2を開始
2024-12-19 15:48:54.332 | DEBUG    | __mp_main__:worker:12 - プロセス2: ステップ1を実行中
2024-12-19 15:48:54.332 | DEBUG    | __mp_main__:worker:12 - プロセス2: ステップ2を実行中
2024-12-19 15:48:54.332 | DEBUG    | __mp_main__:worker:12 - プロセス2: ステップ3を実行中
2024-12-19 15:48:54.332 | DEBUG    | __mp_main__:worker:12 - プロセス1: ステップ2を実行中
2024-12-19 15:48:54.332 | DEBUG    | __mp_main__:worker:12 - プロセス2: ステップ4を実行中
2024-12-19 15:48:54.332 | DEBUG    | __mp_main__:worker:12 - プロセス2: ステップ5を実行中
2024-12-19 15:48:54.332 | INFO     | __mp_main__:worker:13 - プロセス2を終了
2024-12-19 15:48:54.332 | INFO     | __mp_main__:worker:9 - プロセス3を開始
2024-12-19 15:48:54.332 | DEBUG    | __mp_main__:worker:12 - プロセス1: ステップ3を実行中
2024-12-19 15:48:54.332 | DEBUG    | __mp_main__:worker:12 - プロセス3: ステップ1を実行中
2024-12-19 15:48:54.332 | DEBUG    | __mp_main__:worker:12 - プロセス1: ステップ4を実行中
2024-12-19 15:48:54.332 | DEBUG    | __mp_main__:worker:12 - プロセス1: ステップ5を実行中
2024-12-19 15:48:54.332 | INFO     | __mp_main__:worker:13 - プロセス1を終了
2024-12-19 15:48:54.332 | DEBUG    | __mp_main__:worker:12 - プロセス3: ステップ2を実行中
2024-12-19 15:48:54.332 | DEBUG    | __mp_main__:worker:12 - プロセス3: ステップ3を実行中
2024-12-19 15:48:54.332 | DEBUG    | __mp_main__:worker:12 - プロセス3: ステップ4を実行中
2024-12-19 15:48:54.332 | DEBUG    | __mp_main__:worker:12 - プロセス3: ステップ5を実行中
2024-12-19 15:48:54.332 | INFO     | __mp_main__:worker:13 - プロセス3を終了
2024-12-19 15:48:54.341 | INFO     | __main__:<module>:26 - すべてのプロセスが終了

非同期の場合。

from loguru import logger
import asyncio

logger.remove()
logger.add("async.log", enqueue=True)

async def async_worker(task_id):
    logger.info(f"タスク{task_id}を開始")
    for i in range(1,6):
        logger.debug(f"タスク{task_id}: ステップ{i}を実行中")
        await asyncio.sleep(0.1)
    logger.info(f"タスク{task_id}を終了")

async def main():
    tasks = [async_worker(i) for i in range(1,4)]
    await asyncio.gather(*tasks)

if __name__ == "__main__":
    asyncio.run(main())
    logger.info("すべてのタスクが終了")

python sample.py

async.log

2024-12-19 16:01:12.851 | INFO     | __main__:async_worker:8 - タスク1を開始
2024-12-19 16:01:12.852 | DEBUG    | __main__:async_worker:10 - タスク1: ステップ1を実行中
2024-12-19 16:01:12.852 | INFO     | __main__:async_worker:8 - タスク2を開始
2024-12-19 16:01:12.852 | DEBUG    | __main__:async_worker:10 - タスク2: ステップ1を実行中
2024-12-19 16:01:12.852 | INFO     | __main__:async_worker:8 - タスク3を開始
2024-12-19 16:01:12.852 | DEBUG    | __main__:async_worker:10 - タスク3: ステップ1を実行中
2024-12-19 16:01:12.954 | DEBUG    | __main__:async_worker:10 - タスク1: ステップ2を実行中
2024-12-19 16:01:12.954 | DEBUG    | __main__:async_worker:10 - タスク2: ステップ2を実行中
2024-12-19 16:01:12.954 | DEBUG    | __main__:async_worker:10 - タスク3: ステップ2を実行中
2024-12-19 16:01:13.055 | DEBUG    | __main__:async_worker:10 - タスク1: ステップ3を実行中
2024-12-19 16:01:13.056 | DEBUG    | __main__:async_worker:10 - タスク2: ステップ3を実行中
2024-12-19 16:01:13.056 | DEBUG    | __main__:async_worker:10 - タスク3: ステップ3を実行中
2024-12-19 16:01:13.157 | DEBUG    | __main__:async_worker:10 - タスク1: ステップ4を実行中
2024-12-19 16:01:13.158 | DEBUG    | __main__:async_worker:10 - タスク2: ステップ4を実行中
2024-12-19 16:01:13.158 | DEBUG    | __main__:async_worker:10 - タスク3: ステップ4を実行中
2024-12-19 16:01:13.259 | DEBUG    | __main__:async_worker:10 - タスク1: ステップ5を実行中
2024-12-19 16:01:13.260 | DEBUG    | __main__:async_worker:10 - タスク2: ステップ5を実行中
2024-12-19 16:01:13.260 | DEBUG    | __main__:async_worker:10 - タスク3: ステップ5を実行中
2024-12-19 16:01:13.361 | INFO     | __main__:async_worker:12 - タスク1を終了
2024-12-19 16:01:13.361 | INFO     | __main__:async_worker:12 - タスク2を終了
2024-12-19 16:01:13.362 | INFO     | __main__:async_worker:12 - タスク3を終了
2024-12-19 16:01:13.364 | INFO     | __main__:<module>:20 - すべてのタスクが終了

kun432

ここまでで全然触れてこなかったけど、add()で登録する出力先を"sink"というらしい。で、このsinkには関数を指定することもできる。

sample.py

from loguru import logger


def sample_sink(message):
    print(f"[SINK] {message}", end="")


logger.remove()
logger.add(sample_sink)


def main():
    logger.info("カスタムsink関数でログを出力")


if __name__ == "__main__":
    main()

python sample.py

出力

[SINK] 2024-12-19 16:20:22.707 | INFO     | __main__:main:14 - sink関数でログを出力

sinkに非同期な関数を使用する場合、complete()で非同期処理の完了を待つ必要がある。

sample.py

from loguru import logger
import asyncio

# 非同期シンク関数
async def async_sink(message):
    await asyncio.sleep(0.1)
    print(f"[ASYNC SINK] {message}", end="")

logger.remove()
logger.add(async_sink)

async def main():
    logger.info("非同期シンクでログを出力")
    await logger.complete()  # 非同期シンクの完了

if __name__ == "__main__":
    asyncio.run(main())

python sample.py

出力

[ASYNC SINK] 2024-12-19 16:24:10.371 | INFO     | __main__:main:13 - 非同期シンクでログを出力

kun432

詳細なスタックトレース

Loguruは内部でbetter-exceptionsを使用しており、例外をキャッチした際に変数の値などを含む詳細なスタックトレースを取得することができる。

sample.py

from loguru import logger

logger.remove()
logger.add("trace.log", backtrace=True, diagnose=True)


def func(a, b):
    return a / b


def nested(c):
    try:
        func(5, c)
    except ZeroDivisionError:
        logger.exception("What?!")


def main():
    nested(0)


if __name__ == "__main__":
    main()

python sample.py

trace.log

2024-12-19 16:52:58.717 | ERROR    | __main__:nested:15 - What?!
Traceback (most recent call last):

  File "/Users/kun432/work/loguru-test/sample.py", line 23, in <module>
    main()
    └ <function main at 0x10556be20>

  File "/Users/kun432/work/loguru-test/sample.py", line 19, in main
    nested(0)
    └ <function nested at 0x10556bd80>

> File "/Users/kun432/work/loguru-test/sample.py", line 13, in nested
    func(5, c)
    │       └ 0
    └ <function func at 0x104d3e3e0>

  File "/Users/kun432/work/loguru-test/sample.py", line 8, in func
    return a / b
           │   └ 0
           └ 5

ZeroDivisionError: division by zero

ただし、diagnose=Trueを有効にすると不必要に詳細な情報がログに出力されてしまう可能性もある。そのあたりは以下を参照。

kun432

構造化ログ

serialize=Trueを付与すると、JSONで出力される。

sample.py

import sys
from loguru import logger

logger.remove()
logger.add(sys.stdout, serialize=True)


def main() -> None:
  logger.debug("これだけ！美しくシンプルなロギング！")


if __name__ == '__main__':
  main()

python sample.py | jq -r .

出力

{
  "text": "2024-12-19 17:03:44.956 | DEBUG    | __main__:main:9 - これだけ！美しくシンプルなロギング！\n",
  "record": {
    "elapsed": {
      "repr": "0:00:00.009138",
      "seconds": 0.009138
    },
    "exception": null,
    "extra": {},
    "file": {
      "name": "sample.py",
      "path": "/Users/kun432/work/loguru-test/sample.py"
    },
    "function": "main",
    "level": {
      "icon": "🐞",
      "name": "DEBUG",
      "no": 10
    },
    "line": 9,
    "message": "これだけ！美しくシンプルなロギング！",
    "module": "sample",
    "name": "__main__",
    "process": {
      "id": 75939,
      "name": "MainProcess"
    },
    "thread": {
      "id": 8628752192,
      "name": "MainThread"
    },
    "time": {
      "repr": "2024-12-19 17:03:44.956883+09:00",
      "timestamp": 1734595424.956883
    }
  }
}

メッセージがJSONになるのはsinkに送られる前になる。

sample.py

import sys
from loguru import logger


def custom_sink(message):
    print(f"[SINK] {message}", end="")


logger.remove()
logger.add(custom_sink, serialize=True)


def main() -> None:
  logger.debug("これだけ！美しくシンプルなロギング！")


if __name__ == '__main__':
  main()

python sample.py

出力

[SINK] {"text": "2024-12-19 17:16:10.051 | DEBUG    | __main__:main:12 - これだけ！美しくシンプルなロギング！\n", (snip)

ログ出力時にキーワード引数を追加するとそれもJSONのキーとして含まれる。

sample.py

import sys
from loguru import logger


logger.remove()
logger.add("json.log", serialize=True)


def main() -> None:
  logger.info("ログ出力します。", user_id=123, ip_addr="192.168.1.1")
  logger.error("エラーログ出力します。", error_code=400, error_message="not found")


if __name__ == '__main__':
  main()

python sample.py

cat json.log | jq -r .

extrasというキーの下に追加されている。あー、これは便利。

出力

{
  "text": "2024-12-19 17:21:25.812 | INFO     | __main__:main:10 - ログ出力します。\n",
  "record": {
    "elapsed": {
      "repr": "0:00:00.007376",
      "seconds": 0.007376
    },
    "exception": null,
    "extra": {
      "user_id": 123,
      "ip_addr": "192.168.1.1"
    },
    "file": {
      "name": "sample.py",
      "path": "/Users/kun432/work/loguru-test/sample.py"
    },
    "function": "main",
    "level": {
      "icon": "ℹ️",
      "name": "INFO",
      "no": 20
    },
    "line": 10,
    "message": "ログ出力します。",
    "module": "sample",
    "name": "__main__",
    "process": {
      "id": 79011,
      "name": "MainProcess"
    },
    "thread": {
      "id": 8628752192,
      "name": "MainThread"
    },
    "time": {
      "repr": "2024-12-19 17:21:25.812759+09:00",
      "timestamp": 1734596485.812759
    }
  }
}
{
  "text": "2024-12-19 17:21:25.812 | ERROR    | __main__:main:11 - エラーログ出力します。\n",
  "record": {
    "elapsed": {
      "repr": "0:00:00.007504",
      "seconds": 0.007504
    },
    "exception": null,
    "extra": {
      "error_code": 400,
      "error_message": "not found"
    },
    "file": {
      "name": "sample.py",
      "path": "/Users/kun432/work/loguru-test/sample.py"
    },
    "function": "main",
    "level": {
      "icon": "❌",
      "name": "ERROR",
      "no": 40
    },
    "line": 11,
    "message": "エラーログ出力します。",
    "module": "sample",
    "name": "__main__",
    "process": {
      "id": 79011,
      "name": "MainProcess"
    },
    "thread": {
      "id": 8628752192,
      "name": "MainThread"
    },
    "time": {
      "repr": "2024-12-19 17:21:25.812887+09:00",
      "timestamp": 1734596485.812887
    }
  }
}

bind()を使うと、コンテキスト化できる。

sample.py

import sys
from loguru import logger

logger.remove()
logger.add("json.log", serialize=True)


def main() -> None:
  context_logger = logger.bind(user_id=123, ip_addr="192.168.1.1")
  context_logger.info("コンテキスト化されたログ")
  context_logger.bind(user_name="太郎").info("一時的にコンテキストを追加することもできる")
  context_logger.info("文字列フォーマットできる ユーザ名: {user_name}", user_name="花子")


if __name__ == '__main__':
  main()

python sample.py

cat json.log | jq -r .

出力

{
  "text": "2024-12-19 17:53:38.008 | INFO     | __main__:main:12 - コンテキスト化されたログ\n",
  "record": {
    "elapsed": {
      "repr": "0:00:00.007188",
      "seconds": 0.007188
    },
    "exception": null,
    "extra": {
      "user_id": 123,
      "ip_addr": "192.168.1.1"
    },
    "file": {
      "name": "sample.py",
      "path": "/Users/kun432/work/loguru-test/sample.py"
    },
    "function": "main",
    "level": {
      "icon": "ℹ️",
      "name": "INFO",
      "no": 20
    },
    "line": 12,
    "message": "コンテキスト化されたログ",
    "module": "sample",
    "name": "__main__",
    "process": {
      "id": 88080,
      "name": "MainProcess"
    },
    "thread": {
      "id": 8628752192,
      "name": "MainThread"
    },
    "time": {
      "repr": "2024-12-19 17:53:38.008422+09:00",
      "timestamp": 1734598418.008422
    }
  }
}
{
  "text": "2024-12-19 17:53:38.008 | INFO     | __main__:main:13 - 一時的にコンテキストを追加することもできる\n",
  "record": {
    "elapsed": {
      "repr": "0:00:00.007315",
      "seconds": 0.007315
    },
    "exception": null,
    "extra": {
      "user_id": 123,
      "ip_addr": "192.168.1.1",
      "user_name": "太郎"
    },
    "file": {
      "name": "sample.py",
      "path": "/Users/kun432/work/loguru-test/sample.py"
    },
    "function": "main",
    "level": {
      "icon": "ℹ️",
      "name": "INFO",
      "no": 20
    },
    "line": 13,
    "message": "一時的にコンテキストを追加することもできる",
    "module": "sample",
    "name": "__main__",
    "process": {
      "id": 88080,
      "name": "MainProcess"
    },
    "thread": {
      "id": 8628752192,
      "name": "MainThread"
    },
    "time": {
      "repr": "2024-12-19 17:53:38.008549+09:00",
      "timestamp": 1734598418.008549
    }
  }
}
{
  "text": "2024-12-19 17:53:38.008 | INFO     | __main__:main:14 - 文字列フォーマットできる ユーザ名: 花子\n",
  "record": {
    "elapsed": {
      "repr": "0:00:00.007360",
      "seconds": 0.00736
    },
    "exception": null,
    "extra": {
      "user_id": 123,
      "ip_addr": "192.168.1.1",
      "user_name": "花子"
    },
    "file": {
      "name": "sample.py",
      "path": "/Users/kun432/work/loguru-test/sample.py"
    },
    "function": "main",
    "level": {
      "icon": "ℹ️",
      "name": "INFO",
      "no": 20
    },
    "line": 14,
    "message": "文字列フォーマットできる ユーザ名: 花子",
    "module": "sample",
    "name": "__main__",
    "process": {
      "id": 88080,
      "name": "MainProcess"
    },
    "thread": {
      "id": 8628752192,
      "name": "MainThread"
    },
    "time": {
      "repr": "2024-12-19 17:53:38.008594+09:00",
      "timestamp": 1734598418.008594
    }
  }
}

addでフォーマッティングする場合はextraでキーを指定する。

sample.py

import sys
from loguru import logger


logger.remove()
logger.add("json.log", format="{extra[ip_addr]} {extra[user_id]} {message}", serialize=True)


def main() -> None:
  context_logger = logger.bind(user_id=123, ip_addr="192.168.1.1")
  context_logger.info("コンテキスト化されたログ")

if __name__ == '__main__':
  main()

python sample.py

cat json.log | jq -r .

出力

{
  "text": "192.168.1.1 123 コンテキスト化されたログ\n",
  "record": {
    "elapsed": {
      "repr": "0:00:00.007324",
      "seconds": 0.007324
    },
    (snip)
    "extra": {
      "user_id": 123,
      "ip_addr": "192.168.1.1"
    },
    (snip)

contextualize()をwithで指定するとそのブロックだけ有効なログのコンテキスト化ができる。

sample.py

import sys
from loguru import logger

logger.remove()
logger.add(sys.stdout, serialize=True)


def main():
    logger.info("タスク開始")
    with logger.contextualize(task_id=1):
        logger.info("タスク実行")
    with logger.contextualize(task_id=2):
        logger.info("タスク実行")
    logger.info("タスク終了")


if __name__ == '__main__':
  main()

python sample.py  | jq -r .

出力

{
  "text": "2024-12-19 18:08:19.292 | INFO     | __main__:main:9 - タスク開始\n",
  "record": {
    "elapsed": {
      "repr": "0:00:00.007095",
      "seconds": 0.007095
    },
    "exception": null,
    "extra": {},
    "file": {
      "name": "sample.py",
      "path": "/Users/kun432/work/loguru-test/sample.py"
    },
    "function": "main",
    "level": {
      "icon": "ℹ️",
      "name": "INFO",
      "no": 20
    },
    "line": 9,
    "message": "タスク開始",
    "module": "sample",
    "name": "__main__",
    "process": {
      "id": 92376,
      "name": "MainProcess"
    },
    "thread": {
      "id": 8628752192,
      "name": "MainThread"
    },
    "time": {
      "repr": "2024-12-19 18:08:19.292867+09:00",
      "timestamp": 1734599299.292867
    }
  }
}
{
  "text": "2024-12-19 18:08:19.292 | INFO     | __main__:main:11 - タスク実行\n",
  "record": {
    "elapsed": {
      "repr": "0:00:00.007193",
      "seconds": 0.007193
    },
    "exception": null,
    "extra": {
      "task_id": 1
    },
    "file": {
      "name": "sample.py",
      "path": "/Users/kun432/work/loguru-test/sample.py"
    },
    "function": "main",
    "level": {
      "icon": "ℹ️",
      "name": "INFO",
      "no": 20
    },
    "line": 11,
    "message": "タスク実行",
    "module": "sample",
    "name": "__main__",
    "process": {
      "id": 92376,
      "name": "MainProcess"
    },
    "thread": {
      "id": 8628752192,
      "name": "MainThread"
    },
    "time": {
      "repr": "2024-12-19 18:08:19.292965+09:00",
      "timestamp": 1734599299.292965
    }
  }
}
{
  "text": "2024-12-19 18:08:19.293 | INFO     | __main__:main:13 - タスク実行\n",
  "record": {
    "elapsed": {
      "repr": "0:00:00.007243",
      "seconds": 0.007243
    },
    "exception": null,
    "extra": {
      "task_id": 2
    },
    "file": {
      "name": "sample.py",
      "path": "/Users/kun432/work/loguru-test/sample.py"
    },
    "function": "main",
    "level": {
      "icon": "ℹ️",
      "name": "INFO",
      "no": 20
    },
    "line": 13,
    "message": "タスク実行",
    "module": "sample",
    "name": "__main__",
    "process": {
      "id": 92376,
      "name": "MainProcess"
    },
    "thread": {
      "id": 8628752192,
      "name": "MainThread"
    },
    "time": {
      "repr": "2024-12-19 18:08:19.293015+09:00",
      "timestamp": 1734599299.293015
    }
  }
}
{
  "text": "2024-12-19 18:08:19.293 | INFO     | __main__:main:14 - タスク終了\n",
  "record": {
    "elapsed": {
      "repr": "0:00:00.007276",
      "seconds": 0.007276
    },
    "exception": null,
    "extra": {},
    "file": {
      "name": "sample.py",
      "path": "/Users/kun432/work/loguru-test/sample.py"
    },
    "function": "main",
    "level": {
      "icon": "ℹ️",
      "name": "INFO",
      "no": 20
    },
    "line": 14,
    "message": "タスク終了",
    "module": "sample",
    "name": "__main__",
    "process": {
      "id": 92376,
      "name": "MainProcess"
    },
    "thread": {
      "id": 8628752192,
      "name": "MainThread"
    },
    "time": {
      "repr": "2024-12-19 18:08:19.293048+09:00",
      "timestamp": 1734599299.293048
    }
  }
}

構造化とは異なるが、bind()をフィルタと組み合わせて出力を分岐したりもできる。

sample.py

from loguru import logger

logger.remove()

# 標準的なログ出力の場合
logger.add("normal.log")

# 特定のフィルタ条件を満たすログを別ファイルに出力
logger.add(
    "special.log",
    filter=lambda record: "special" in record["extra"]
)

def main():
    logger.info("通常ログ1")
    logger.bind(special=True).info("フィルタ条件を満たすログ")
    logger.info("通常ログ2")

if __name__ == "__main__":
    main()

python sample.py

normal.log

2024-12-19 18:15:07.120 | INFO     | __main__:main:15 - 通常ログ1
2024-12-19 18:15:07.120 | INFO     | __main__:main:16 - フィルタ条件を満たすログ
2024-12-19 18:15:07.120 | INFO     | __main__:main:17 - 通常ログ2

special.log

2024-12-19 18:15:07.120 | INFO     | __main__:main:16 - フィルタ条件を満たすログ

patch()を使うと、新しいログメッセージに動的に値を追加できる。

sample.py

import sys
from loguru import logger
from datetime import datetime
import time

logger.remove()
logger.add(sys.stdout, format="{time} {message} {extra}")
logger = logger.patch(lambda record: record["extra"].update(utc=datetime.utcnow()))


def main():
    for i in range(1,4):
        logger.info("ログ出力")
        time.sleep(1)


if __name__ == "__main__":
    main()

python sample.py

出力

2024-12-19T18:29:34.976534+0900 ログ出力 {'utc': datetime.datetime(2024, 12, 19, 9, 29, 34, 976645)}
2024-12-19T18:29:35.981906+0900 ログ出力 {'utc': datetime.datetime(2024, 12, 19, 9, 29, 35, 982205)}
2024-12-19T18:29:36.987755+0900 ログ出力 {'utc': datetime.datetime(2024, 12, 19, 9, 29, 36, 988223)}

kun432

コストの高い関数の遅延評価

opt()を使うと、遅延評価ができる。コストの高い関数などを実行する際に、ログレベルなどの条件に応じて必要な場合にのみ関数を実行させる、といったことができる。

まず遅延評価しない場合。

sample.py

import sys
from loguru import logger

logger.remove()
logger.add(sys.stdout, level="INFO")


def expensive_function():
    print("高コスト関数が実行されました")
    return "高コストな結果"


def main():
    logger.debug("結果: {}", expensive_function())
    logger.info("通常のログ出力")


if __name__ == "__main__":
    main()

python sample.py

高コスト関数が実行されました
2024-12-20 01:19:13.811 | INFO     | __main__:main:15 - 通常のログ出力

ログ出力はINFOレベルに設定しているので、該当の関数部分のログは出力されていないが、関数自体は実行されているのがわかる。

次に遅延評価を行う場合の例。opt(lazy=True)を付与してlambdaでラップする。

sample.py

import sys
from loguru import logger

logger.remove()
logger.add(sys.stdout, level="INFO")


def expensive_function():
    print("高コスト関数が実行されました")
    return "高コストな結果"


def main():
    logger.opt(lazy=True).debug("DEBUGの場合の結果: {}", lambda: expensive_function())
    logger.info("通常のログ出力")

if __name__ == "__main__":
    main()

python sample.py

出力

2024-12-20 01:27:27.926 | INFO     | __main__:main:15 - 通常のログ出力

INFOレベルではないので関数が実行されていないことがわかる。

当然DEBUGレベルに変えれば実行される。

sample.py

import sys
from loguru import logger

logger.remove()
logger.add(sys.stdout, level="DEBUG")  # DEBUGレベルに変更


def expensive_function():
    print("高コスト関数が実行されました")
    return "高コストな結果"


def main():
    logger.opt(lazy=True).debug("DEBUG時の結果: {}", lambda: expensive_function())
    logger.info("通常のログ出力")


if __name__ == "__main__":
    main()

python sample.py

出力

2024-12-20 01:30:43.024 | DEBUG    | __main__:main:14 - DEBUG時の結果: 高コストな結果
2024-12-20 01:30:43.024 | INFO     | __main__:main:15 - 通常のログ出力

なお、opt()には他にも色々な使い方ができる。

`exception`

exceptionを使うと、例外情報をログに追加できる。

sample.py

import sys
from loguru import logger

logger.remove()
logger.add(sys.stdout, level="INFO")


def main():
    try:
        1 / 0
    except ZeroDivisionError:
        logger.opt(exception=True).error("例外が発生しました")


if __name__ == "__main__":
    main()

python sample.py

出力

2024-12-20 01:37:50.462 | ERROR    | __main__:main:12 - 例外が発生しました
Traceback (most recent call last):

  File "/Users/kun432/work/loguru-test/sample.py", line 16, in <module>
    main()
    └ <function main at 0x10102a3e0>

> File "/Users/kun432/work/loguru-test/sample.py", line 10, in main
    1 / 0

ZeroDivisionError: division by zero

`depth`

depthを使うと、呼び出し元の情報を変更できる。

sample.py

import sys
from loguru import logger

logger.remove()
logger.add(sys.stdout, level="INFO")

def func():
    logger.info("呼び出し元変更なし")
    logger.opt(depth=1).info("呼び出し元を変更")

def main():
    func()

if __name__ == "__main__":
    main()

python sample.py

出力

2024-12-20 01:43:10.630 | INFO     | __main__:func:8 - 呼び出し元変更なし
2024-12-20 01:43:10.630 | INFO     | __main__:main:12 - 呼び出し元を変更

共通のロギングヘルパー関数何かを用意する場合に便利

sample.py

import sys
from loguru import logger

logger.remove()
logger.add(sys.stdout, level="INFO")

# ログを記録するヘルパー関数
def log_request(level, endpoint, status_code):
    logger.opt(depth=1).log(level, f"エンドポイント: {endpoint}, ステータスコード: {status_code}")

# エンドポイント処理関数
def get_user():
    log_request("INFO", "/users", 200)

def create_user():
    log_request("ERROR", "/users/create", 500)

def main():
    get_user()
    create_user()

if __name__ == "__main__":
    main()

python sample.py

出力

2024-12-20 01:45:19.182 | INFO     | __main__:get_user:13 - エンドポイント: /users, ステータスコード: 200
2024-12-20 01:45:19.182 | ERROR    | __main__:create_user:16 - エンドポイント: /users/create, ステータスコード: 500

`colors`

colorsを使うと、カラーリングの有効無効を指定できる。

sample.py

import sys
from loguru import logger

logger.remove()
logger.add(sys.stdout, level="INFO")


def main():
    logger.info("<red>エラー発生</red> <green>成功</green>")
    logger.opt(colors=True).info("<red>エラー発生</red> <green>成功</green>")


if __name__ == "__main__":
    main()

python sample.py

`raw`

rawを使うとフォーマッターを無効化できる。

sample.py

import sys
from loguru import logger

logger.remove()
logger.add(sys.stdout, format="{time} {level} {message}", level="INFO")

def main():
    logger.opt(raw=True).info("フォーマット無効化されたログ。\nそのまま出力される。")
    logger.opt(raw=True).info("改行などもそのまま保持される。\n")
    logger.info("通常のフォーマットされたログ")

if __name__ == "__main__":
    main()

python sample.py

出力

フォーマット無効化されたログ。
そのまま出力される。改行などもそのまま保持される。
2024-12-20T02:01:36.768924+0900 INFO 通常のフォーマットされたログ

改行が保持される、というか行単位のログなら文末の改行は必須になりそう。

`record`

recordを使うと、ログのレコード情報を参照することができる。

sample.py

import sys
from loguru import logger

logger.remove()
logger.add(sys.stdout, level="INFO")


def main():
    logger.opt(record=True).info("現在のスレッド名: {record[thread].name}")
    logger.opt(record=True).info("現在の関数: {record[function]}")
    logger.opt(record=True).info("ログレベル: {record[level].name}")
    logger.opt(record=True).info("現在のモジュール: {record[module]}")

if __name__ == "__main__":
    main()

python sample.py

出力

2024-12-20 02:04:16.237 | INFO     | __main__:main:9 - 現在のスレッド名: MainThread
2024-12-20 02:04:16.237 | INFO     | __main__:main:10 - 現在の関数: main
2024-12-20 02:04:16.237 | INFO     | __main__:main:11 - ログレベル: INFO
2024-12-20 02:04:16.237 | INFO     | __main__:main:12 - 現在のモジュール: sample

`capture`

captureはキーワード引数を追加した場合にそれをextraフィールドに追加するかどうかを制御できる。

sample.py

import sys
from loguru import logger

logger.remove()
logger.add(sys.stdout, level="INFO", serialize=True)


def main():
    logger.opt(capture=True).info("extraにキーワード引数が含まれる: {data}", data="追加情報")
    logger.opt(capture=False).info("extraにキーワード引数が含まれない: {data}", data="追加情報")

if __name__ == "__main__":
    main()

python sample.py | jq -r .

出力

{
  "text": "2024-12-20 02:16:12.181 | INFO     | __main__:main:9 - extraにキーワード引数が含まれる: 追加情報\n",
  "record": {
    "elapsed": {
      "repr": "0:00:00.010138",
      "seconds": 0.010138
    },
    "exception": null,
    "extra": {
      "data": "追加情報"
    },
(snip)
}
{
  "text": "2024-12-20 02:16:12.181 | INFO     | __main__:main:10 - extraにキーワード引数が含まれない: 追加情報\n",
  "record": {
    "elapsed": {
      "repr": "0:00:00.010243",
      "seconds": 0.010243
    },
    "exception": null,
    "extra": {},
(snip)
}

kun432

ログレベルのカスタマイズ

デフォルトのログレベル以外にカスタムなログレベルを追加することができる。

sample.py

import sys
from loguru import logger

logger.remove()
logger.add(sys.stdout, format="{time} | {level: <10}{level.icon} | {message}", level="INFO")
logger.level(
    "CUSTOM",       # ログレベルの名前
    no=25,          # ログレベルの優先度（低いほど優先度が高い）
    color="<cyan>", # ターミナル出力時の色
    icon="🤖"       # ログ出力時に表示されるアイコン
)


def main():
    logger.info("通常のINFOレベルのログ")
    logger.log("CUSTOM", "カスタムなログ")


if __name__ == "__main__":
    main()

python sample.py

出力

2024-12-20T02:52:42.871489+0900 | INFO      ℹ️ | 通常のINFOレベルのログ
2024-12-20T02:52:42.871572+0900 | CUSTOM    🤖 | カスタムなログ

ログレベルの優先度はこちら

kun432

タイムスタンプのカスタマイズ

上の方でも少し触れたけど、デフォルトのログフォーマットのタイムスタンプをカスタマイズできる。

自分の環境でのデフォルトは以下のような感じ。

2024-12-20T02:52:42.871489+0900

十分であるが、カスタマイズしてみる。

sample.py

import sys
from loguru import logger

logger.remove()
logger.add(sys.stdout, format="{time:YYYY年MM月DD時_HH時mm分ss秒} | {level} | {message}")

def main():
    logger.info("カスタムな日付フォーマットで出力")

if __name__ == "__main__":
    main()

python sample.py

出力

2024年12月20時_03時02分28秒 | INFO | カスタムな日付フォーマットで出力

日付／時刻のフォーマットは以下

内部でPendulumを使用しているらしい。

kun432

スクリプトやライブラリでの使用

スクリプトやライブラリで使用しやすいような機能が提供されている。

スクリプトではconfigure()を使って、ログ出力の設定を行うことができる。

sample.py

import sys
from loguru import logger

# ログ設定
config = {
    "handlers": [
        {"sink": sys.stdout, "format": "{time} - {level} - {message}", "level": "INFO"},
        {"sink": "debug.log", "format": "{time} - {level} - {message} - {extra[app]}", "level": "DEBUG"},
    ],
    "extra": {"app": "MyScript"},
}

logger.configure(**config)

def main():
    logger.info("スクリプトが開始されました")
    logger.debug("デバッグ情報を記録します")
    logger.error("エラーが発生しました")

if __name__ == "__main__":
    main()

標準出力

2024-12-20T03:13:29.954217+0900 - INFO - スクリプトが開始されました
2024-12-20T03:13:29.954409+0900 - ERROR - エラーが発生しました

debug.log

2024-12-20T03:13:29.954217+0900 - INFO - スクリプトが開始されました - MyScript
2024-12-20T03:13:29.954375+0900 - DEBUG - デバッグ情報を記録します - MyScript
2024-12-20T03:13:29.954409+0900 - ERROR - エラーが発生しました - MyScript

なお、loguru-configパッケージを使用すると、設定をファイルから読み込むこともできる。

ライブラリでは設定を最小化して、ライブラリを使用するアプリケーション側で制御できるようにする。

mylibrary.py

from loguru import logger

# ライブラリ側では`add()`を使わない
# `disable()`でライブラリ側でのロギングを無効化しておく
logger.disable("mylibrary")     # ライブラリの`__name__`を指定

def library_function():
    logger.info("ライブラリ関数が呼び出されました")
    logger.debug("ライブラリ関数のデバッグメッセージ")

アプリケーション側でライブラリのロギングを有効にしない場合

sample.py

import sys
from loguru import logger
import mylibrary

logger.remove()
logger.add(sys.stdout, level="DEBUG")

# `enable()`でライブラリのロギングを有効化
#logger.enable("mylibrary")

def main():
    logger.info("アプリケーションが開始されました")
    mylibrary.library_function()

if __name__ == "__main__":
    main()

python sample.py

出力

2024-12-20 03:24:01.809 | INFO     | __main__:main:9 - アプリケーションが開始されました

アプリケーション側でライブラリのロギングを有効にした場合

sample.py

(snip)
logger.enable("mylibrary")
(snip)

python sample.py

出力

2024-12-20 03:25:11.435 | INFO     | __main__:main:11 - アプリケーションが開始されました
2024-12-20 03:25:11.435 | INFO     | mylibrary:library_function:7 - ライブラリ関数が呼び出されました
2024-12-20 03:25:11.435 | DEBUG    | mylibrary:library_function:8 - ライブラリ関数のデバッグメッセージ

kun432

標準のloggingモジュールとの完全互換性

Loguruは、Pythonの標準loggingモジュールと完全に互換性がある。既存でloggingを使用していても、Loguruを統合することができる。

標準loggingのハンドラをLoguruのsinkとして使う

import logging
from loguru import logger

# loggingのハンドラの設定
stdout_handler = logging.StreamHandler()
formatter = logging.Formatter("%(asctime)s | %(levelname)s | %(message)s")
stdout_handler.setFormatter(formatter)

# Loguruにloggingのハンドラを追加
logger.remove()
logger.add(stdout_handler, format="{time} | {level} | {message}")

def main():
    logger.info("INFOログです")
    logger.warning("WARNINGログです")
    logger.debug("DEBUGログです")

if __name__ == "__main__":
    main()

python sample.py

出力

2024-12-20 03:45:01,444 | INFO | 2024-12-20T03:45:01.444520+0900 | INFO | INFOログです
2024-12-20 03:45:01,444 | WARNING | 2024-12-20T03:45:01.444600+0900 | WARNING | WARNINGログです
2024-12-20 03:45:01,444 | DEBUG | 2024-12-20T03:45:01.444643+0900 | DEBUG | DEBUGログです

Loguruから標準loggingにプロパゲートする

sample.py

import sys
import logging
from loguru import logger

# 標準のlogging設定
logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)s,%(levelname)s,%(message)s",
    stream=sys.stdout
)

# Loguruのログを標準loggingに転送するHandler
class PropagateHandler(logging.Handler):
    def emit(self, record):
        logging.getLogger(record.name).handle(record)

# LoguruにHandlerを追加
logger.remove()
logger.add(PropagateHandler(), format="{time} | {level} | {message}")


def main():
    logger.info("LoguruからINFOログを送信")
    logger.debug("LoguruからDEBUGログを送信")


if __name__ == "__main__":
    main()

python sample.py

出力

2024-12-20 03:36:36,946,INFO,2024-12-20T03:36:36.946356+0900 | INFO | LoguruからINFOログを送信
2024-12-20 03:36:36,946,DEBUG,2024-12-20T03:36:36.946460+0900 | DEBUG | LoguruからDEBUGログを送信

逆に標準loggingからLoguruにプロパゲート

sample.py

import sys
import logging
from loguru import logger
import inspect

# 標準のloggingメッセージをLoguruに転送するHandler
class InterceptHandler(logging.Handler):
    def emit(self, record):
        # 標準`logging`のレベルをLoguruに変換
        level = logger.level(record.levelname).name if record.levelname in logger._core.levels else record.levelno

        # スタックフレームを調整
        frame, depth = inspect.currentframe(), 2
        while frame and frame.f_code.co_filename == logging.__file__:
            frame = frame.f_back
            depth += 1

        # Loguruにログを送信
        logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())

# 標準`logging`の設定
logging.basicConfig(handlers=[InterceptHandler()], level=logging.DEBUG)


# 標準`logging`を使用した例
def log_with_standard_logging():
    logging.getLogger().info("標準loggingからのINFOメッセージ")
    logging.getLogger().debug("標準loggingからのDEBUGメッセージ")


# Loguruを使用した例
def log_with_loguru():
    logger.info("LoguruからのINFOメッセージ")
    logger.debug("LoguruからのDEBUGメッセージ")


if __name__ == "__main__":
    log_with_standard_logging()
    log_with_loguru()

python sample.py

出力

2024-12-20 03:40:07.494 | INFO     | logging:callHandlers:1762 - 標準loggingからのINFOメッセージ
2024-12-20 03:40:07.494 | DEBUG    | logging:callHandlers:1762 - 標準loggingからのDEBUGメッセージ
2024-12-20 03:40:07.494 | INFO     | __main__:log_with_loguru:31 - LoguruからのINFOメッセージ
2024-12-20 03:40:07.494 | DEBUG    | __main__:log_with_loguru:32 - LoguruからのDEBUGメッセージ

kun432

環境変数によるデフォルト設定のカスタマイズ

環境変数でデフォルト設定をカスタマイズすることができる。例えば以下のような内容のスクリプト。

import sys
from loguru import logger

def main():
    logger.debug("DEBUGログ")
    logger.info("INFOログ")

if __name__ == "__main__":
    main()

デフォルトだとDEBUGになるので、普通に実行すると両方とも表示される。

python sample.py

出力

2024-12-20 03:55:20.124 | DEBUG    | __main__:main:5 - DEBUGログ
2024-12-20 03:55:20.124 | INFO     | __main__:main:6 - INFOログ

INFOレベルに設定、あとフォーマットも変えてみる。

export LOGURU_LEVEL=INFO
export LOGURU_FORMAT="{time},{level},{message}"
python sample.py

出力

2024-12-20T03:57:15.707539+0900,INFO,INFOログ

基本的に、add()で使用できるパラメータがLOGURU_*として環境変数で定義できると考えれば良いみたい。

環境変数の設定については以下

kun432

ログパーサー

parse()を使うと、生成されたログファイルを簡単に解析することができる。

まずログを出力する。

sample.py

from loguru import logger

logger.remove()
logger.add("sample.log", format="{time} | {level} | {message}")

def main():
    logger.info("ユーザー123がログインしました")
    logger.warning("CPU使用率が高いです")
    logger.error("アプリケーションでエラーが発生しました")

if __name__ == "__main__":
    main()

python sample.py

以下のようにログファイルが出力される

sample.log

2024-12-20T04:09:12.922447+0900 | INFO | ユーザー123がログインしました
2024-12-20T04:09:12.922535+0900 | WARNING | CPU使用率が高いです
2024-12-20T04:09:12.922562+0900 | ERROR | アプリケーションでエラーが発生しました

これをパースする

parse.py

from loguru import logger

pattern = r"(?P<time>.*?) \| (?P<level>[A-Z]+) \| (?P<message>.*)"  # 正規表現パターン
parsed_logs = logger.parse("sample.log", pattern)

for log in parsed_logs:
    print(log)

python parse.py

出力

{'time': '2024-12-20T04:09:12.922447+0900', 'level': 'INFO', 'message': 'ユーザー123がログインしました'}
{'time': '2024-12-20T04:09:12.922535+0900', 'level': 'WARNING', 'message': 'CPU使用率が高いです'}
{'time': '2024-12-20T04:09:12.922562+0900', 'level': 'ERROR', 'message': 'アプリケーションでエラーが発生しました'}

castパラメータを使って型変換を行うこともできる。dateutilを使って日付の変換を行う。

pip install python-dateutil

parse.py

from loguru import logger
import dateutil.parser

pattern = r"(?P<time>.*?) \| (?P<level>[A-Z]+) \| (?P<message>.*)"
cast_dict = {
    "time": dateutil.parser.parse,  # 日時をdatetimeオブジェクトに変換
    "level": str,
    "message": str,
}
parsed_logs = logger.parse("sample.log", pattern, cast=cast_dict)

for log in parsed_logs:
    print(log)

python parse.py

出力

{'time': datetime.datetime(2024, 12, 20, 4, 9, 12, 922447, tzinfo=tzoffset(None, 32400)), 'level': 'INFO', 'message': 'ユーザー123がログインしました'}
{'time': datetime.datetime(2024, 12, 20, 4, 9, 12, 922535, tzinfo=tzoffset(None, 32400)), 'level': 'WARNING', 'message': 'CPU使用率が高いです'}
{'time': datetime.datetime(2024, 12, 20, 4, 9, 12, 922562, tzinfo=tzoffset(None, 32400)), 'level': 'ERROR', 'message': 'アプリケーションでエラーが発生しました'}

kun432

通知

Loguruはappriseを使用した通知に対応している。これにより、アプリケーションで発生したエラーや特定のログレベルのメッセージを、slackやメールなどで通知することができる。

appriseをインストール

pip install apprise

今回はSlackに通知してみる。slackのIncoming WebhookのURLが以下の場合。

https://hooks.slack.com/services/AAAAAAAA/BBBBBBBB/CCCCCCCC

import sys
from loguru import logger
import apprise

# Appriseで通知先を設定。slackの例。
notifier = apprise.Apprise()
notifier.add("slack://AAAAAAAA/BBBBBBBB/CCCCCCCC")

# LoguruにApprise通知を組み込む
logger.remove()
logger.add(sys.stdout, level="INFO")
logger.add(
    notifier.notify,  # Appriseの通知メソッド
    level="ERROR",    # ERRORレベル以上のログを通知
    format="{time} | {level} | {message}",
)

def main():
    logger.info("これは通常のINFOログです")
    logger.error("これは通知されるERRORログです")
    try:
        1 / 0
    except ZeroDivisionError:
        logger.exception("例外が発生しました")

if __name__ == "__main__":
    main()

python sample.py

標準出力

2024-12-20T04:36:54.274050+0900,INFO,これは通常のINFOログです
2024-12-20T04:36:54.274125+0900,ERROR,これは通知されるERRORログです
2024-12-20T04:36:54.861337+0900,ERROR,例外が発生しました
Traceback (most recent call last):

  File "/Users/kun432/work/loguru-test/sample.py", line 27, in <module>
    main()
    └ <function main at 0x104127ce0>

> File "/Users/kun432/work/loguru-test/sample.py", line 22, in main
    1 / 0

ZeroDivisionError: division by zero

こんな感じで通知される。

appriseで対応している通知手段は以下を参照

kun432

 まとめLoguru自体は別に新しいライブラリというわけではなくて、GitHubのReleasesを見ていると2019年頃からあったみたい。全然知らなかった。
標準loggingについては以前に一通りやってみたのだけど、なんというかとてもわかりにくい印象があって、未だに雰囲気で使っている感が自分にはある。
https://zenn.dev/kun432/scraps/39389b320ae38c
それに比べると、Loguruはとてもわかりやすいし、便利な機能も多い。もう今後はこちらを使っていこうと思った。

このスクラップは4日前にクローズされました