ニーズはあるか？pythonのloggingモジュールを使いやすくクラス化してみた。

Pythonのloggingモジュール使って、ログファイルを出力するクラスを作ってみました。
特に同様の機能を検索したことがないので、もし同様の機能を実現している人がいた場合は教えていただけると参考にいたします。
コピー、改変等自由、できれば「こうしたよ」とかあればラッキーです。

作成の背景

• Raspberry Pi制御系開発をしていて、稼働状況把握やデバッグのためにログファイルを書いている
• GPSデータを受信しながら制御するなどの場合、GPS生データだけを別ファイルにしたかった
• エラーログだけ別ファイルにしたかった
• ターミナルでテスト実行中に、ログと同じ内容をターミナルに出したいがprint文は書きたくない

などの必要性があり作成した。

テスト環境

環境１

PC:Raspberry Pi 400
OS:bullseye
version:Python 3.9.2

環境２

PC:Raspberry Pi 500
OS:trixie
version:Python 3.13.5

エディタ

Visual Studio Code 1.105.1

myAppLogClass.py

以下が作成したclassのコードです。

myAppLogClass.py

# ------------------------------------------------------------------------
# Python logging module use simple
# https://www.simpleware.jp
#   Ver.1.0.1
# 2025/09/12    固有のロガー名を引数に追加。１PG内での複数ロガー実現のため。
#               ログフォルダ作成失敗時、とりあえずraiseする。
# ------------------------------------------------------------------------

import logging
import logging.handlers
from pathlib import Path
import sys # 標準出力へのログのためにsysモジュールをインポート

class AppLogger:
    def __init__(self,
        log_directory_name="logs",
        info_log_name="all_logs.log",
        error_log_name="error_logs.log",
        max_log_file_size_mb=2,
        backup_count_info=200,
        backup_count_error=200,
        console_log_level=logging.INFO,
        file_log_level=logging.DEBUG,
        logger_name="application_logger"):
        """
        アプリケーションのロガーを初期化します。

        Args:
            log_directory_name (str): ログを保管するフォルダ名。
            info_log_name (str): 通常ログ (INFO, DEBUGなど) のファイル名。
            error_log_name (str): エラーログ (ERROR, CRITICAL) のファイル名。
            max_log_file_size_mb (int): ログファイルの最大サイズ (MB)。
            backup_count_info (int): 通常ログのバックアップファイル数。
            backup_count_error (int): エラーログのバックアップファイル数。
            console_log_level (int): コンソールに出力するログの最低レベル。
            file_log_level (int): ファイルに出力するログの最低レベル。
            logger_name (str): ロガーインスタンス名。
        """
        self.log_dir = Path(log_directory_name)
        self.info_log_file = self.log_dir / info_log_name
        self.error_log_file = self.log_dir / error_log_name
        self.max_bytes = max_log_file_size_mb * 1024 * 1024
        self.backup_count_info = backup_count_info
        self.backup_count_error = backup_count_error
        self.console_log_level = console_log_level
        self.file_log_level = file_log_level
        self.logger_name = logger_name

        self._setup_log_directory()
        self.logger = self._setup_logger()

    def _setup_log_directory(self):
        """ログ保管用フォルダを作成します。"""
        try:
            self.log_dir.mkdir(parents=True, exist_ok=True)
            print(f"ログフォルダ '{self.log_dir.resolve()}' を準備しました。")
        except OSError as e:
            print(f"エラー: ログフォルダ '{self.log_dir.resolve()}' の作成に失敗しました: {e}", file=sys.stderr)
            # ログフォルダ作成失敗は致命的なので、ここで例外を発生させるか、アプリケーションを終了する選択肢もあります
            raise RuntimeError(f"エラー: ログフォルダ '{self.log_dir.resolve()}' の作成に失敗しました")

    def _setup_logger(self):
        """
        ロガーインスタンスを設定し、ハンドラを追加します。
        """
        logger = logging.getLogger(self.logger_name) # アプリケーション固有のロガー名を設定
        logger.setLevel(logging.DEBUG) # 全てのハンドラに渡す最低レベル

        # 既存のハンドラをクリア (クラスのインスタンスを複数作成する場合などに重複を防ぐため)
        if logger.handlers:
            for handler in list(logger.handlers):
                logger.removeHandler(handler)

        # --- 通常ログ用のローテーションハンドラ ---
        info_handler = logging.handlers.RotatingFileHandler(
            self.info_log_file,
            maxBytes=self.max_bytes,
            backupCount=self.backup_count_info,
            encoding='utf-8'
        )
        info_handler.setLevel(self.file_log_level)
        info_formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
        info_handler.setFormatter(info_formatter)
        logger.addHandler(info_handler)

        # --- エラーログ用のローテーションハンドラ ---
        error_handler = logging.handlers.RotatingFileHandler(
            self.error_log_file,
            maxBytes=self.max_bytes,
            backupCount=self.backup_count_error,
            encoding='utf-8'
        )
        error_handler.setLevel(logging.ERROR) # エラーログはERRORレベル以上のみ
        error_formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s')
        error_handler.setFormatter(error_formatter)
        logger.addHandler(error_handler)

        # --- コンソール出力用のハンドラ ---
        console_handler = logging.StreamHandler(sys.stdout)
        console_handler.setLevel(self.console_log_level)
        console_formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
        console_handler.setFormatter(console_formatter)
        logger.addHandler(console_handler)

        return logger

    def get_logger(self):
        """
        設定済みのロガーインスタンスを返します。
        """
        return self.logger

クラスの説明

コメントを読んでもらえばいいのですが、簡単に説明記載します。
loggingの基本については、ここでは省きますので、別途調べるなどで確認ください。

引数

• log_directory_name (str): ログを保管するフォルダ名
• info_log_name (str): 通常ログ (INFO, DEBUGなど) のファイル名
• error_log_name (str): エラーログ (ERROR, CRITICAL) のファイル名
• max_log_file_size_mb (int): ログファイルの最大サイズ (MB)
• backup_count_info (int): 通常ログのバックアップファイル数
• backup_count_error (int): エラーログのバックアップファイル数
• console_log_level (int): コンソールに出力するログの最低レベル
• file_log_level (int): ファイルに出力するログの最低レベル
• logger_name (str): ロガーインスタンス名

指定されたフォルダに、指定されたファイル名で通常ログとエラーログを書き出します。
　エラーログはレベル「error」以上を書き出します。
指定されたファイルサイズを上限に、指定されたバックアップファイル数まで保管されます。
　ファイル数上限を超えるものは破棄です。
コンソール出力、ファイル出力とも、出力するエラーレベルを指定できます。

logger_nameを変更してインスタンス生成すれば、複数のロガーを定義可能です。
　１プログラムで複数ログファイルを作りたいときはこの名前を別にしてインスタンス生成してください。

使い方

loggingとこのクラスをimportしてインスタンス生成します。
指定されない引数は規定値が使用されます。

import logging
import myAppLogClass as myAppLog        #logging ラッパークラス

# log setting
# ロガーを初期化
my_app_logger_instance = myAppLog.AppLogger(
    logger_name = "my_logger",
    log_directory_name = "my_app_logs",
    info_log_name="my_logs.log",
    error_log_name="my_error_logs.log",
    console_log_level = logging.DEBUG,
    file_log_level = logging.DEBUG
)
my_logger = my_app_logger_instance.get_logger()

以下のように使いえばOKです。
ファイル出力とコンソール出力が同時に行えます。

my_logger.info(f"{__file__} start!")
my_logger.debug(f"テスト　ログです")

サンプルプログラム

使い方の実例として上記のクラスを使ったサンプルプログラム以下に記載します。

multiprocess_logging_sample.py

# ------------------------------------------------------------------------
# myAppLogClass.py used sample
# https://www.simpleware.jp
#   Ver.0.0.0
# 2025/09/12    
# multiprocessingでのmyAppLogClass.pyの使用サンプル
#   main,funa,funbそれぞれ別のログファイルを書くサンプルプログラム。
#       main処理：関数A、関数Bを起動し、30秒待って終了する。
#       funca関数：ランダム文字列生成、書き込みOKとなったら共有メモリへ書き込む。
#       funcb関数：共有メモリが読み込み可能となったら、読み込む。
# ------------------------------------------------------------------------

import multiprocessing
import random
import string
import time
import logging

import myAppLogClass as myAppLog        #logging ラッパークラス

# log setting
# ロガーを初期化
# 異なる設定で複数のロガーインスタンスを作成することも可能
my_app_logger_instance1 = myAppLog.AppLogger(
    logger_name = "main_logger",
    log_directory_name = "my_app_logs",
    info_log_name="main_logs.log",
    error_log_name="main_error_logs.log",
    console_log_level = logging.DEBUG, # 既定値はコンソールにはINFO以上を表示
    file_log_level = logging.DEBUG  # 既定値はファイルにはDEBUG以上を記録
)
my_logger_main = my_app_logger_instance1.get_logger()
my_logger_main.info(f"{__file__} start!")
my_logger_main.info(f"ログは '{my_app_logger_instance1.info_log_file.resolve()}' と '{my_app_logger_instance1.error_log_file.resolve()}' に出力されます。")

my_app_logger_instance2 = myAppLog.AppLogger(
    logger_name = "funca_logger",
    log_directory_name = "my_app_logs",
    info_log_name="funca_logs.log",
    error_log_name="funca_error_logs.log",
    console_log_level = logging.DEBUG, # 既定値はコンソールにはINFO以上を表示
    file_log_level = logging.DEBUG  # 既定値はファイルにはDEBUG以上を記録
)
my_logger_funca = my_app_logger_instance2.get_logger()
my_logger_funca.info(f"{__file__} start!")
my_logger_funca.info(f"ログは '{my_app_logger_instance2.info_log_file.resolve()}' と '{my_app_logger_instance2.error_log_file.resolve()}' に出力されます。")

my_app_logger_instance3 = myAppLog.AppLogger(
    logger_name = "funcb_logger",
    log_directory_name = "my_app_logs",
    info_log_name="funcb_logs.log",
    error_log_name="funcb_error_logs.log",
    console_log_level = logging.DEBUG, # 既定値はコンソールにはINFO以上を表示
    file_log_level = logging.DEBUG  # 既定値はファイルにはDEBUG以上を記録
)
my_logger_funcb = my_app_logger_instance3.get_logger()
my_logger_funcb.info(f"{__file__} start!")
my_logger_funcb.info(f"ログは '{my_app_logger_instance3.info_log_file.resolve()}' と '{my_app_logger_instance3.error_log_file.resolve()}' に出力されます。")

# ランダムな文字列を生成するヘルパー関数
def _generate_random_string(length=10):
    """指定された長さのランダムな英数字文字列を生成します。"""
    characters = string.ascii_letters + string.digits
    return ''.join(random.choice(characters) for i in range(length))

# 共有メモリフラグ　セット関数
def _set_shared_flag_value(flag, value=1):
    flag.value = value
    
## 関数A: データ生成・送信側
def function_a(data, flag, end):
    my_logger_funca.info("関数Aを開始します。")
    prc_count = 0
    while True:
        # フラグが0になるのを待つ（関数Bが読み取りを終えるのを待つ）
        while flag.value == 1:
            if end.value == 1:
                my_logger_funca.info("関数Aがデータ読み取り待ち中に終了指示あり。")
                break                      
            time.sleep(0.01)
        # MAINから終了通知を受け取ったらループを抜ける
        if end.value == 1:
            my_logger_funca.info("関数Aが終了します。")
            break

        # ランダムな文字列を生成
        random_string = _generate_random_string(15)  # 例として15文字の文字列を生成
        
        # 共有メモリに値を書き込む
        # `data.value`にはバイト文字列を代入する必要がある
        prc_count += 1
        st_conma = ","
        data.value = random_string.encode('utf-8') + st_conma.encode('utf-8') + str(prc_count).encode('utf-8')
        
        my_logger_funca.debug(f"関数A: 共有メモリに書き込みました -> {random_string}")

        # 書き込みが完了したことを示すためにフラグを1に設定
        _set_shared_flag_value(flag, 1)
        
        # 0.5秒間隔を空ける
        time.sleep(0.5)

## 関数B: データ受信・表示側
def function_b(data, flag, end):
    my_logger_funcb.info("関数Bを開始します。")
    start_time = time.time()
    
    while True:
        # フラグが1になるのを待つ（関数Aが書き込みを終えるのを待つ）
        while flag.value == 0:
            if end.value == 1:
                my_logger_funcb.info("関数Bがデータ書き込み待ち中に終了指示あり。")
                break            
            time.sleep(0.01)
        # mainから終了通知を受け取ったらループを抜ける
        if end.value == 1:
            
            my_logger_funcb.info("関数Bが終了します。")
            break
                
        # 共有メモリからデータを読み取る
        # `data.value`はバイト文字列なので、デコードして文字列に戻す
        received_data = data.value.decode('utf-8')
        my_logger_funcb.debug(f"関数B: 共有メモリから読み取りました -> {received_data}")
        
        # 読み取りが完了したことを示すためにフラグを0に設定
        _set_shared_flag_value(flag, 0)
    

# プロセスの起動と共有オブジェクトの管理
if __name__ == '__main__':
    my_logger_main.info(f"MAIN started!!")
    # 共有メモリと共有変数を定義
    # `multiprocessing.Array`で共有メモリを確保 (c_char_pはバイト文字列)
    # 文字列の最大長を考慮し、サイズを256バイトに変更
    shared_data = multiprocessing.Array('c', 256)
    
    # `multiprocessing.Value`で共有変数を確保 (iは整数型)
    shared_flag = multiprocessing.Value('i', 0)
    shared_end = multiprocessing.Value('i', 0)

    # 各関数を独立したプロセスとして作成
    process_a = multiprocessing.Process(target=function_a, args=(shared_data, shared_flag, shared_end))
    process_b = multiprocessing.Process(target=function_b, args=(shared_data, shared_flag, shared_end))
    
    # プロセスを開始
    process_a.start()
    process_b.start()
    
    # 30秒間待って、終了する
    time.sleep(30)
    _set_shared_flag_value(shared_end, 1)  #終了する
    # 両方のプロセスが終了するのを待つ
    process_a.join()
    process_b.join()
    
    my_logger_main.info("すべての関数が正常に終了しました。")

サンプルプログラムの説明

処理概要

メイン処理からランダムな文字列を生成する関数Aと、その文字列を共有メモリ経由で読み込む関数Bを起動し、一定時間経過後、メイン処理が停止命令（共有メモリ上のフラグ変数に値をセットする）を出して２関数を停止するプログラムです。
マルチプロセスで関数Aと関数Bを実行します。
メイン処理、関数A、関数Bそれぞれ別にログ出力を行います。

結び

タイトルの通り、ニーズがあるのかはわかりませんが、ログ出力を行いたいプログラマーの方々の参考になればと思います。
もっと簡便なコーディングや、フェイルセーフな作り方あれば教えてください。