【Streamlit】高度なロギング入門：基本からCI環境まで

import streamlit as st
import logging

# loggerの作成とレベル設定
logger = logging.getLogger('streamlit_app')
logger.setLevel(logging.DEBUG)

# handlerの設定（コンソール出力）
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.DEBUG)

# フォーマッターの設定
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
console_handler.setFormatter(formatter)

# loggerにhandlerを追加
if not logger.hasHandlers():
    logger.addHandler(console_handler)

def main():
    logger.info('アプリが起動しました')
    st.title("Streamlit Logging Example")

    user_input = st.text_input("文字を入力してください")
    if user_input:
        logger.debug(f'ユーザー入力: {user_input}')
        st.write(f'あなたの入力: {user_input}')

if __name__ == "__main__":
    main()

import os

def setup_custom_logger(name, log_file='app.log', level=logging.DEBUG):
    logger = logging.getLogger(name)
    logger.setLevel(level)

    # 既にhandlerがある場合は重複防止
    if logger.hasHandlers():
        logger.handlers.clear()

    # コンソール用ハンドラー
    ch = logging.StreamHandler()
    ch.setLevel(level)
    ch_formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
    ch.setFormatter(ch_formatter)

    # ファイル用ハンドラー
    fh = logging.FileHandler(log_file, encoding='utf-8')
    fh.setLevel(level)
    fh_formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    fh.setFormatter(fh_formatter)

    logger.addHandler(ch)
    logger.addHandler(fh)

    return logger

logger = setup_custom_logger('streamlit_custom_logger', 'streamlit_app.log')

def main():
    logger.info('アプリ開始')
    st.title("カスタムロガーの例")

    val = st.number_input("数値を入力")
    if val:
        logger.debug(f'入力された数値: {val}')
    st.write(f"あなたの入力値は {val} です。")

if __name__ == "__main__":
    main()

import queue
from logging.handlers import QueueHandler, QueueListener
import threading

def setup_async_logger(name, log_file='async_app.log', level=logging.DEBUG):
    log_queue = queue.Queue(-1)  # 無制限サイズのキュー
    logger = logging.getLogger(name)
    logger.setLevel(level)

    # 基本のファイルハンドラー（Listenerが使用）
    file_handler = logging.FileHandler(log_file, encoding='utf-8')
    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    file_handler.setFormatter(formatter)

    # QueueHandlerをloggerにセット（非同期でキューに積むのみ）
    queue_handler = QueueHandler(log_queue)
    logger.handlers.clear()
    logger.addHandler(queue_handler)

    # QueueListenerを作成し専用スレッドでハンドリング
    listener = QueueListener(log_queue, file_handler)
    listener.start()

    return logger, listener

logger, listener = setup_async_logger('async_streamlit')

def main():
    logger.info('非同期ロギング開始')
    st.title("非同期ロギングのサンプル")

    text = st.text_input("入力してください")
    if text:
        logger.debug(f'入力内容: {text}')
        st.write(f'はじめまして、{text}さん。')

if __name__ == "__main__":
    try:
        main()
    finally:
        listener.stop()  # アプリ終了時にListener停止

pip install python-json-logger

from pythonjsonlogger import jsonlogger

def setup_json_logger(name, log_file='json_app.log', level=logging.DEBUG):
    logger = logging.getLogger(name)
    logger.setLevel(level)

    handler = logging.FileHandler(log_file, encoding='utf-8')
    formatter = jsonlogger.JsonFormatter('%(asctime)s %(name)s %(levelname)s %(message)s %(filename)s %(lineno)d')
    handler.setFormatter(formatter)

    logger.handlers.clear()
    logger.addHandler(handler)
    return logger

json_logger = setup_json_logger('json_logger')

json_logger.info('ユーザーログ', extra={'user_id': 1234, 'action': 'login'})

import time

start = time.perf_counter()

for i in range(1000):
    logger.debug(f'ログメッセージ{i}')

end = time.perf_counter()
st.write(f'同期ロギングにかかった時間: {end - start:.4f}秒')

# ルートロガー設定（main.pyなど）
root_logger = setup_custom_logger('my_project')

# 各モジュール
import logging

logger = logging.getLogger('my_project.moduleA')

def process():
    logger.info("モジュールAの処理が開始されました")

import uuid

def log_with_session_id(logger, message, session_id=None, level=logging.INFO):
    if not session_id:
        session_id = str(uuid.uuid4())
    logger.log(level, f"[session_id={session_id}] {message}")
    return session_id

session_id = log_with_session_id(logger, "処理開始")
# 処理中...
log_with_session_id(logger, "処理中のイベント発生", session_id, logging.DEBUG)

import logging

def setup_test_logger():
    logger = logging.getLogger('streamlit_test')
    logger.setLevel(logging.WARNING)
    if not logger.hasHandlers():
        handler = logging.StreamHandler()
        formatter = logging.Formatter('%(levelname)s - %(message)s')
        handler.setFormatter(formatter)
        logger.addHandler(handler)
    return logger

def test_sample():
    logger = setup_test_logger()
    logger.info("このログは表示されません")
    logger.error("テスト失敗時のエラーログ")
    assert 1 == 1