ホームページ > バックエンド開発 > Python チュートリアル > 完全な Python ロギング ガイド: ベスト プラクティスと実装

完全な Python ロギング ガイド: ベスト プラクティスと実装

Linda Hamilton
リリース: 2024-12-23 08:19:13
オリジナル
901 人が閲覧しました

Complete Python Logging Guide: Best Practices & Implementation

適切なロギングが重要な理由

技術的な詳細に入る前に、適切なロギングがなぜ重要なのかを理解しましょう。

  • 本番環境で効果的なデバッグが可能
  • アプリケーションの動作に関する洞察を提供します
  • パフォーマンスの監視を容易にします
  • セキュリティインシデントの追跡に役立ちます
  • コンプライアンス要件をサポート
  • メンテナンス効率の向上

Python ロギングのクイック スタート

Python ロギングが初めての方のために、logging.basicConfig を使用した基本的な例を示します。

# Simple python logging example
import logging

# Basic logger in python example
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# Create a logger
logger = logging.getLogger(__name__)

# Logger in python example
logger.info("This is an information message")
logger.warning("This is a warning message")
ログイン後にコピー
ログイン後にコピー
ログイン後にコピー
ログイン後にコピー

この例では、Python のロギング モジュールの基本を示し、アプリケーションで Python ロガー ロギングを使用する方法を示します。

Python のロギング モジュールの入門

基本的なセットアップ

簡単なロギング設定から始めましょう:

import logging

# Basic configuration
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# Your first logger
logger = logging.getLogger(__name__)

# Using the logger
logger.info("Application started")
logger.warning("Watch out!")
logger.error("Something went wrong")
ログイン後にコピー
ログイン後にコピー
ログイン後にコピー
ログイン後にコピー

ログレベルを理解する

Python ロギングには 5 つの標準レベルがあります:

Level Numeric Value When to Use
DEBUG 10 Detailed information for diagnosing problems
INFO 20 General operational events
WARNING 30 Something unexpected happened
ERROR 40 More serious problem
CRITICAL 50 Program may not be able to continue

print() ステートメントを超えて

なぜ print ステートメントではなくログ記録を選択するのですか?

  • より適切に分類するための重大度レベル
  • タイムスタンプ情報
  • ソース情報(ファイル、行番号)
  • 設定可能な出力先
  • 本番環境に対応したフィルタリング
  • スレッドの安全性

ロギング システムの構成

基本的な構成オプション

# Simple python logging example
import logging

# Basic logger in python example
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# Create a logger
logger = logging.getLogger(__name__)

# Logger in python example
logger.info("This is an information message")
logger.warning("This is a warning message")
ログイン後にコピー
ログイン後にコピー
ログイン後にコピー
ログイン後にコピー

高度な構成

より複雑なアプリケーションの場合:

import logging

# Basic configuration
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# Your first logger
logger = logging.getLogger(__name__)

# Using the logger
logger.info("Application started")
logger.warning("Watch out!")
logger.error("Something went wrong")
ログイン後にコピー
ログイン後にコピー
ログイン後にコピー
ログイン後にコピー

高度なログの使用

構造化されたロギング

構造化ログは、ログの分析と監視に不可欠な、一貫性のある機械可読形式を提供します。構造化ログのパターンとベスト プラクティスの包括的な概要については、構造化ログ ガイドを確認してください。 Python で構造化ログを実装しましょう:

logging.basicConfig(
    filename='app.log',
    filemode='w',
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    level=logging.DEBUG,
    datefmt='%Y-%m-%d %H:%M:%S'
)
ログイン後にコピー
ログイン後にコピー
ログイン後にコピー

エラー管理

運用上の問題をデバッグするには、適切なエラー ログが重要です。包括的なアプローチは次のとおりです:

config = {
    'version': 1,
    'formatters': {
        'detailed': {
            'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        }
    },
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
            'level': 'INFO',
            'formatter': 'detailed'
        },
        'file': {
            'class': 'logging.FileHandler',
            'filename': 'app.log',
            'level': 'DEBUG',
            'formatter': 'detailed'
        }
    },
    'loggers': {
        'myapp': {
            'handlers': ['console', 'file'],
            'level': 'DEBUG',
            'propagate': True
        }
    }
}

logging.config.dictConfig(config)
ログイン後にコピー
ログイン後にコピー
ログイン後にコピー

同時ロギング

マルチスレッドアプリケーションにログインする場合は、スレッドの安全性を確保する必要があります。

import json
import logging
from datetime import datetime

class JSONFormatter(logging.Formatter):
    def __init__(self):
        super().__init__()

    def format(self, record):
        # Create base log record
        log_obj = {
            "timestamp": self.formatTime(record, self.datefmt),
            "name": record.name,
            "level": record.levelname,
            "message": record.getMessage(),
            "module": record.module,
            "function": record.funcName,
            "line": record.lineno
        }

        # Add exception info if present
        if record.exc_info:
            log_obj["exception"] = self.formatException(record.exc_info)

        # Add custom fields from extra
        if hasattr(record, "extra_fields"):
            log_obj.update(record.extra_fields)

        return json.dumps(log_obj)

# Usage Example
logger = logging.getLogger(__name__)
handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())
logger.addHandler(handler)

# Log with extra fields
logger.info("User logged in", extra={"extra_fields": {"user_id": "123", "ip": "192.168.1.1"}})
ログイン後にコピー
ログイン後にコピー
ログイン後にコピー

異なる環境でのログイン

さまざまなアプリケーション環境には、特定のログ記録アプローチが必要です。 Web アプリケーション、マイクロサービス、バックグラウンド タスクのいずれを使用している場合でも、各環境には固有のログ要件とベスト プラクティスがあります。さまざまな展開シナリオにわたって効果的なログ記録を実装する方法を見てみましょう。

Web アプリケーションのログ記録

Django ログ構成

包括的な Django ログ設定は次のとおりです:

import traceback
import sys
from contextlib import contextmanager

class ErrorLogger:
    def __init__(self, logger):
        self.logger = logger

    @contextmanager
    def error_context(self, operation_name, **context):
        """Context manager for error logging with additional context"""
        try:
            yield
        except Exception as e:
            # Capture the current stack trace
            exc_type, exc_value, exc_traceback = sys.exc_info()

            # Format error details
            error_details = {
                "operation": operation_name,
                "error_type": exc_type.__name__,
                "error_message": str(exc_value),
                "context": context,
                "stack_trace": traceback.format_exception(exc_type, exc_value, exc_traceback)
            }

            # Log the error with full context
            self.logger.error(
                f"Error in {operation_name}: {str(exc_value)}",
                extra={"error_details": error_details}
            )

            # Re-raise the exception
            raise

# Usage Example
logger = logging.getLogger(__name__)
error_logger = ErrorLogger(logger)

with error_logger.error_context("user_authentication", user_id="123", attempt=2):
    # Your code that might raise an exception
    authenticate_user(user_id)
ログイン後にコピー
ログイン後にコピー
ログイン後にコピー

Flask ログのセットアップ

Flask は、カスタマイズ可能な独自のログ システムを提供します。

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

def setup_thread_safe_logging():
    """Set up thread-safe logging with a queue"""
    # Create the queue
    log_queue = Queue()

    # Create handlers
    console_handler = logging.StreamHandler()
    file_handler = logging.FileHandler('app.log')

    # Create queue handler and listener
    queue_handler = QueueHandler(log_queue)
    listener = QueueListener(
        log_queue,
        console_handler,
        file_handler,
        respect_handler_level=True
    )

    # Configure root logger
    root_logger = logging.getLogger()
    root_logger.addHandler(queue_handler)

    # Start the listener in a separate thread
    listener.start()

    return listener

# Usage
listener = setup_thread_safe_logging()

def worker_function():
    logger = logging.getLogger(__name__)
    logger.info(f"Worker thread {threading.current_thread().name} starting")
    # Do work...
    logger.info(f"Worker thread {threading.current_thread().name} finished")

# Create and start threads
threads = [
    threading.Thread(target=worker_function)
    for _ in range(3)
]
for thread in threads:
    thread.start()
ログイン後にコピー
ログイン後にコピー

FastAPI ロギングの実践

FastAPI は、いくつかのミドルウェア機能拡張により Python のロギングを活用できます。

# settings.py
LOGGING = {
    'version': 1,
    'disable_existing_loggers': False,
    'formatters': {
        'verbose': {
            'format': '{levelname} {asctime} {module} {process:d} {thread:d} {message}',
            'style': '{',
        },
        'simple': {
            'format': '{levelname} {message}',
            'style': '{',
        },
    },
    'filters': {
        'require_debug_true': {
            '()': 'django.utils.log.RequireDebugTrue',
        },
    },
    'handlers': {
        'console': {
            'level': 'INFO',
            'filters': ['require_debug_true'],
            'class': 'logging.StreamHandler',
            'formatter': 'simple'
        },
        'file': {
            'level': 'ERROR',
            'class': 'logging.FileHandler',
            'filename': 'django-errors.log',
            'formatter': 'verbose'
        },
        'mail_admins': {
            'level': 'ERROR',
            'class': 'django.utils.log.AdminEmailHandler',
            'include_html': True,
        }
    },
    'loggers': {
        'django': {
            'handlers': ['console'],
            'propagate': True,
        },
        'django.request': {
            'handlers': ['file', 'mail_admins'],
            'level': 'ERROR',
            'propagate': False,
        },
        'myapp': {
            'handlers': ['console', 'file'],
            'level': 'INFO',
        }
    }
}
ログイン後にコピー
ログイン後にコピー

マイクロサービスのロギング

マイクロサービスの場合、分散トレーシングと相関 ID が不可欠です。

import logging
from logging.handlers import RotatingFileHandler
from flask import Flask, request

app = Flask(__name__)

def setup_logger():
    # Create formatter
    formatter = logging.Formatter(
        '[%(asctime)s] %(levelname)s in %(module)s: %(message)s'
    )

    # File Handler
    file_handler = RotatingFileHandler(
        'flask_app.log',
        maxBytes=10485760,  # 10MB
        backupCount=10
    )
    file_handler.setLevel(logging.INFO)
    file_handler.setFormatter(formatter)

    # Add request context
    class RequestFormatter(logging.Formatter):
        def format(self, record):
            record.url = request.url
            record.remote_addr = request.remote_addr
            return super().format(record)

    # Configure app logger
    app.logger.addHandler(file_handler)
    app.logger.setLevel(logging.INFO)

    return app.logger

# Usage in routes
@app.route('/api/endpoint')
def api_endpoint():
    app.logger.info(f'Request received from {request.remote_addr}')
    # Your code here
    return jsonify({'status': 'success'})
ログイン後にコピー
ログイン後にコピー

バックグラウンドタスクのログ記録

バックグラウンド タスクの場合、適切なログ処理とローテーションを確保する必要があります。

from fastapi import FastAPI, Request
from typing import Callable
import logging
import time

app = FastAPI()

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# Middleware for request logging
@app.middleware("http")
async def log_requests(request: Request, call_next: Callable):
    start_time = time.time()
    response = await call_next(request)
    duration = time.time() - start_time

    log_dict = {
        "url": str(request.url),
        "method": request.method,
        "client_ip": request.client.host,
        "duration": f"{duration:.2f}s",
        "status_code": response.status_code
    }

    logger.info(f"Request processed: {log_dict}")
    return response

# Example endpoint with logging
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    logger.info(f"Retrieving item {item_id}")
    # Your code here
    return {"item_id": item_id}
ログイン後にコピー
ログイン後にコピー

一般的なロギング パターンと解決策

リクエストID追跡

アプリケーション全体にリクエスト追跡を実装する:

import logging
import contextvars
from uuid import uuid4

# Create context variable for trace ID
trace_id_var = contextvars.ContextVar('trace_id', default=None)

class TraceIDFilter(logging.Filter):
    def filter(self, record):
        trace_id = trace_id_var.get()
        record.trace_id = trace_id if trace_id else 'no_trace'
        return True

def setup_microservice_logging(service_name):
    logger = logging.getLogger(service_name)

    # Create formatter with trace ID
    formatter = logging.Formatter(
        '%(asctime)s - %(name)s - [%(trace_id)s] - %(levelname)s - %(message)s'
    )

    # Add handlers with trace ID filter
    handler = logging.StreamHandler()
    handler.setFormatter(formatter)
    handler.addFilter(TraceIDFilter())

    logger.addHandler(handler)
    logger.setLevel(logging.INFO)

    return logger

# Usage in microservice
logger = setup_microservice_logging('order_service')

def process_order(order_data):
    # Generate or get trace ID from request
    trace_id_var.set(str(uuid4()))

    logger.info("Starting order processing", extra={
        'order_id': order_data['id'],
        'customer_id': order_data['customer_id']
    })

    # Process order...

    logger.info("Order processed successfully")
ログイン後にコピー
ログイン後にコピー

ユーザーアクティビティのログ記録

ユーザーのアクションを安全に追跡します:

# Simple python logging example
import logging

# Basic logger in python example
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# Create a logger
logger = logging.getLogger(__name__)

# Logger in python example
logger.info("This is an information message")
logger.warning("This is a warning message")
ログイン後にコピー
ログイン後にコピー
ログイン後にコピー
ログイン後にコピー

トラブルシューティングとデバッグ

ロギングの問題を効果的にトラブルシューティングするには、一般的な問題とその解決策を理解する必要があります。このセクションでは、ロギングの実装時に開発者が直面する最も頻繁な課題について説明し、ロギング構成をデバッグするための実用的な解決策を提供します。

ロギングに関する一般的な問題

欠落しているログエントリ

import logging

# Basic configuration
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# Your first logger
logger = logging.getLogger(__name__)

# Using the logger
logger.info("Application started")
logger.warning("Watch out!")
logger.error("Something went wrong")
ログイン後にコピー
ログイン後にコピー
ログイン後にコピー
ログイン後にコピー

パフォーマンスのボトルネック

logging.basicConfig(
    filename='app.log',
    filemode='w',
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    level=logging.DEBUG,
    datefmt='%Y-%m-%d %H:%M:%S'
)
ログイン後にコピー
ログイン後にコピー
ログイン後にコピー

ロギングの一般的な落とし穴と解決策

構成の問題

config = {
    'version': 1,
    'formatters': {
        'detailed': {
            'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        }
    },
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
            'level': 'INFO',
            'formatter': 'detailed'
        },
        'file': {
            'class': 'logging.FileHandler',
            'filename': 'app.log',
            'level': 'DEBUG',
            'formatter': 'detailed'
        }
    },
    'loggers': {
        'myapp': {
            'handlers': ['console', 'file'],
            'level': 'DEBUG',
            'propagate': True
        }
    }
}

logging.config.dictConfig(config)
ログイン後にコピー
ログイン後にコピー
ログイン後にコピー

メモリとリソースの問題

import json
import logging
from datetime import datetime

class JSONFormatter(logging.Formatter):
    def __init__(self):
        super().__init__()

    def format(self, record):
        # Create base log record
        log_obj = {
            "timestamp": self.formatTime(record, self.datefmt),
            "name": record.name,
            "level": record.levelname,
            "message": record.getMessage(),
            "module": record.module,
            "function": record.funcName,
            "line": record.lineno
        }

        # Add exception info if present
        if record.exc_info:
            log_obj["exception"] = self.formatException(record.exc_info)

        # Add custom fields from extra
        if hasattr(record, "extra_fields"):
            log_obj.update(record.extra_fields)

        return json.dumps(log_obj)

# Usage Example
logger = logging.getLogger(__name__)
handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())
logger.addHandler(handler)

# Log with extra fields
logger.info("User logged in", extra={"extra_fields": {"user_id": "123", "ip": "192.168.1.1"}})
ログイン後にコピー
ログイン後にコピー
ログイン後にコピー

フォーマット文字列とパフォーマンスの問題

import traceback
import sys
from contextlib import contextmanager

class ErrorLogger:
    def __init__(self, logger):
        self.logger = logger

    @contextmanager
    def error_context(self, operation_name, **context):
        """Context manager for error logging with additional context"""
        try:
            yield
        except Exception as e:
            # Capture the current stack trace
            exc_type, exc_value, exc_traceback = sys.exc_info()

            # Format error details
            error_details = {
                "operation": operation_name,
                "error_type": exc_type.__name__,
                "error_message": str(exc_value),
                "context": context,
                "stack_trace": traceback.format_exception(exc_type, exc_value, exc_traceback)
            }

            # Log the error with full context
            self.logger.error(
                f"Error in {operation_name}: {str(exc_value)}",
                extra={"error_details": error_details}
            )

            # Re-raise the exception
            raise

# Usage Example
logger = logging.getLogger(__name__)
error_logger = ErrorLogger(logger)

with error_logger.error_context("user_authentication", user_id="123", attempt=2):
    # Your code that might raise an exception
    authenticate_user(user_id)
ログイン後にコピー
ログイン後にコピー
ログイン後にコピー

ハンドラー構成トラップ

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

def setup_thread_safe_logging():
    """Set up thread-safe logging with a queue"""
    # Create the queue
    log_queue = Queue()

    # Create handlers
    console_handler = logging.StreamHandler()
    file_handler = logging.FileHandler('app.log')

    # Create queue handler and listener
    queue_handler = QueueHandler(log_queue)
    listener = QueueListener(
        log_queue,
        console_handler,
        file_handler,
        respect_handler_level=True
    )

    # Configure root logger
    root_logger = logging.getLogger()
    root_logger.addHandler(queue_handler)

    # Start the listener in a separate thread
    listener.start()

    return listener

# Usage
listener = setup_thread_safe_logging()

def worker_function():
    logger = logging.getLogger(__name__)
    logger.info(f"Worker thread {threading.current_thread().name} starting")
    # Do work...
    logger.info(f"Worker thread {threading.current_thread().name} finished")

# Create and start threads
threads = [
    threading.Thread(target=worker_function)
    for _ in range(3)
]
for thread in threads:
    thread.start()
ログイン後にコピー
ログイン後にコピー

スレッドの安全性に関する考慮事項

# settings.py
LOGGING = {
    'version': 1,
    'disable_existing_loggers': False,
    'formatters': {
        'verbose': {
            'format': '{levelname} {asctime} {module} {process:d} {thread:d} {message}',
            'style': '{',
        },
        'simple': {
            'format': '{levelname} {message}',
            'style': '{',
        },
    },
    'filters': {
        'require_debug_true': {
            '()': 'django.utils.log.RequireDebugTrue',
        },
    },
    'handlers': {
        'console': {
            'level': 'INFO',
            'filters': ['require_debug_true'],
            'class': 'logging.StreamHandler',
            'formatter': 'simple'
        },
        'file': {
            'level': 'ERROR',
            'class': 'logging.FileHandler',
            'filename': 'django-errors.log',
            'formatter': 'verbose'
        },
        'mail_admins': {
            'level': 'ERROR',
            'class': 'django.utils.log.AdminEmailHandler',
            'include_html': True,
        }
    },
    'loggers': {
        'django': {
            'handlers': ['console'],
            'propagate': True,
        },
        'django.request': {
            'handlers': ['file', 'mail_admins'],
            'level': 'ERROR',
            'propagate': False,
        },
        'myapp': {
            'handlers': ['console', 'file'],
            'level': 'INFO',
        }
    }
}
ログイン後にコピー
ログイン後にコピー

構成ファイルの問題

import logging
from logging.handlers import RotatingFileHandler
from flask import Flask, request

app = Flask(__name__)

def setup_logger():
    # Create formatter
    formatter = logging.Formatter(
        '[%(asctime)s] %(levelname)s in %(module)s: %(message)s'
    )

    # File Handler
    file_handler = RotatingFileHandler(
        'flask_app.log',
        maxBytes=10485760,  # 10MB
        backupCount=10
    )
    file_handler.setLevel(logging.INFO)
    file_handler.setFormatter(formatter)

    # Add request context
    class RequestFormatter(logging.Formatter):
        def format(self, record):
            record.url = request.url
            record.remote_addr = request.remote_addr
            return super().format(record)

    # Configure app logger
    app.logger.addHandler(file_handler)
    app.logger.setLevel(logging.INFO)

    return app.logger

# Usage in routes
@app.route('/api/endpoint')
def api_endpoint():
    app.logger.info(f'Request received from {request.remote_addr}')
    # Your code here
    return jsonify({'status': 'success'})
ログイン後にコピー
ログイン後にコピー

ロギングのテスト

ログを使用した単体テスト

from fastapi import FastAPI, Request
from typing import Callable
import logging
import time

app = FastAPI()

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# Middleware for request logging
@app.middleware("http")
async def log_requests(request: Request, call_next: Callable):
    start_time = time.time()
    response = await call_next(request)
    duration = time.time() - start_time

    log_dict = {
        "url": str(request.url),
        "method": request.method,
        "client_ip": request.client.host,
        "duration": f"{duration:.2f}s",
        "status_code": response.status_code
    }

    logger.info(f"Request processed: {log_dict}")
    return response

# Example endpoint with logging
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    logger.info(f"Retrieving item {item_id}")
    # Your code here
    return {"item_id": item_id}
ログイン後にコピー
ログイン後にコピー

モックロガーを使用したテスト

import logging
import contextvars
from uuid import uuid4

# Create context variable for trace ID
trace_id_var = contextvars.ContextVar('trace_id', default=None)

class TraceIDFilter(logging.Filter):
    def filter(self, record):
        trace_id = trace_id_var.get()
        record.trace_id = trace_id if trace_id else 'no_trace'
        return True

def setup_microservice_logging(service_name):
    logger = logging.getLogger(service_name)

    # Create formatter with trace ID
    formatter = logging.Formatter(
        '%(asctime)s - %(name)s - [%(trace_id)s] - %(levelname)s - %(message)s'
    )

    # Add handlers with trace ID filter
    handler = logging.StreamHandler()
    handler.setFormatter(formatter)
    handler.addFilter(TraceIDFilter())

    logger.addHandler(handler)
    logger.setLevel(logging.INFO)

    return logger

# Usage in microservice
logger = setup_microservice_logging('order_service')

def process_order(order_data):
    # Generate or get trace ID from request
    trace_id_var.set(str(uuid4()))

    logger.info("Starting order processing", extra={
        'order_id': order_data['id'],
        'customer_id': order_data['customer_id']
    })

    # Process order...

    logger.info("Order processed successfully")
ログイン後にコピー
ログイン後にコピー

代替ロギング ソリューション

ろぐる

Loguru は、すぐに使用できる強力な機能を備えた、よりシンプルなログ インターフェイスを提供します。

from logging.handlers import RotatingFileHandler
import logging
import threading
from datetime import datetime

class BackgroundTaskLogger:
    def __init__(self, task_name):
        self.logger = logging.getLogger(f'background_task.{task_name}')
        self.setup_logging()

    def setup_logging(self):
        # Create logs directory if it doesn't exist
        import os
        os.makedirs('logs', exist_ok=True)

        # Setup rotating file handler
        handler = RotatingFileHandler(
            filename=f'logs/task_{datetime.now():%Y%m%d}.log',
            maxBytes=5*1024*1024,  # 5MB
            backupCount=5
        )

        # Create formatter
        formatter = logging.Formatter(
            '%(asctime)s - [%(threadName)s] - %(levelname)s - %(message)s'
        )
        handler.setFormatter(formatter)

        self.logger.addHandler(handler)
        self.logger.setLevel(logging.INFO)

    def log_task_status(self, status, **kwargs):
        """Log task status with additional context"""
        extra = {
            'thread_id': threading.get_ident(),
            'timestamp': datetime.now().isoformat(),
            **kwargs
        }
        self.logger.info(f"Task status: {status}", extra=extra)

# Usage example
def background_job():
    logger = BackgroundTaskLogger('data_processing')
    try:
        logger.log_task_status('started', job_id=123)
        # Do some work...
        logger.log_task_status('completed', records_processed=1000)
    except Exception as e:
        logger.logger.error(f"Task failed: {str(e)}", exc_info=True)
ログイン後にコピー

構造体ログ

Structlog は、コンテキストを含む構造化ログに優れています:

import logging
from contextlib import contextmanager
import threading
import uuid

# Store request ID in thread-local storage
_request_id = threading.local()

class RequestIDFilter(logging.Filter):
    def filter(self, record):
        record.request_id = getattr(_request_id, 'id', 'no_request_id')
        return True

@contextmanager
def request_context(request_id=None):
    """Context manager for request tracking"""
    if request_id is None:
        request_id = str(uuid.uuid4())

    old_id = getattr(_request_id, 'id', None)
    _request_id.id = request_id
    try:
        yield request_id
    finally:
        if old_id is None:
            del _request_id.id
        else:
            _request_id.id = old_id

# Setup logging with request ID
def setup_request_logging():
    logger = logging.getLogger()
    formatter = logging.Formatter(
        '%(asctime)s - [%(request_id)s] - %(levelname)s - %(message)s'
    )

    handler = logging.StreamHandler()
    handler.setFormatter(formatter)
    handler.addFilter(RequestIDFilter())

    logger.addHandler(handler)
    return logger

# Usage example
logger = setup_request_logging()

def process_request(data):
    with request_context() as request_id:
        logger.info("Processing request", extra={
            'data': data,
            'operation': 'process_request'
        })
        # Process the request...
        logger.info("Request processed successfully")
ログイン後にコピー

Python-JSON-ロガー

JSON 形式のログの場合:

# Simple python logging example
import logging

# Basic logger in python example
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# Create a logger
logger = logging.getLogger(__name__)

# Logger in python example
logger.info("This is an information message")
logger.warning("This is a warning message")
ログイン後にコピー
ログイン後にコピー
ログイン後にコピー
ログイン後にコピー

ベストプラクティスとガイドライン

ロギング標準

import logging

# Basic configuration
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# Your first logger
logger = logging.getLogger(__name__)

# Using the logger
logger.info("Application started")
logger.warning("Watch out!")
logger.error("Something went wrong")
ログイン後にコピー
ログイン後にコピー
ログイン後にコピー
ログイン後にコピー

パフォーマンスの最適化

logging.basicConfig(
    filename='app.log',
    filemode='w',
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    level=logging.DEBUG,
    datefmt='%Y-%m-%d %H:%M:%S'
)
ログイン後にコピー
ログイン後にコピー
ログイン後にコピー

ケーススタディ

現実世界の実装: E コマース プラットフォーム

config = {
    'version': 1,
    'formatters': {
        'detailed': {
            'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        }
    },
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
            'level': 'INFO',
            'formatter': 'detailed'
        },
        'file': {
            'class': 'logging.FileHandler',
            'filename': 'app.log',
            'level': 'DEBUG',
            'formatter': 'detailed'
        }
    },
    'loggers': {
        'myapp': {
            'handlers': ['console', 'file'],
            'level': 'DEBUG',
            'propagate': True
        }
    }
}

logging.config.dictConfig(config)
ログイン後にコピー
ログイン後にコピー
ログイン後にコピー

マイクロサービスアーキテクチャの例

import json
import logging
from datetime import datetime

class JSONFormatter(logging.Formatter):
    def __init__(self):
        super().__init__()

    def format(self, record):
        # Create base log record
        log_obj = {
            "timestamp": self.formatTime(record, self.datefmt),
            "name": record.name,
            "level": record.levelname,
            "message": record.getMessage(),
            "module": record.module,
            "function": record.funcName,
            "line": record.lineno
        }

        # Add exception info if present
        if record.exc_info:
            log_obj["exception"] = self.formatException(record.exc_info)

        # Add custom fields from extra
        if hasattr(record, "extra_fields"):
            log_obj.update(record.extra_fields)

        return json.dumps(log_obj)

# Usage Example
logger = logging.getLogger(__name__)
handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())
logger.addHandler(handler)

# Log with extra fields
logger.info("User logged in", extra={"extra_fields": {"user_id": "123", "ip": "192.168.1.1"}})
ログイン後にコピー
ログイン後にコピー
ログイン後にコピー

結論

重要なポイント

  1. まずは基礎: 適切な基本構成から始めます
  • 適切なログレベルを設定します
  • 意味のある形式を構成する
  • 適切なハンドラーを選択してください
  1. 構造化アプローチ: 分析を改善するために構造化ログを使用します
  • 一貫したログ形式
  • コンテキスト情報
  • 機械解析可能な出力
  1. パフォーマンスの問題: 本番環境向けにログを最適化する
  • ログローテーションを実装する
  • 必要に応じて非同期ログを使用します
  • サンプリング戦略を検討する
  1. セキュリティ意識: 機密情報を保護する
    • 機密データをフィルタリングする
    • 適切なアクセス制御を実装する
    • コンプライアンス要件に従います

実装チェックリスト

import traceback
import sys
from contextlib import contextmanager

class ErrorLogger:
    def __init__(self, logger):
        self.logger = logger

    @contextmanager
    def error_context(self, operation_name, **context):
        """Context manager for error logging with additional context"""
        try:
            yield
        except Exception as e:
            # Capture the current stack trace
            exc_type, exc_value, exc_traceback = sys.exc_info()

            # Format error details
            error_details = {
                "operation": operation_name,
                "error_type": exc_type.__name__,
                "error_message": str(exc_value),
                "context": context,
                "stack_trace": traceback.format_exception(exc_type, exc_value, exc_traceback)
            }

            # Log the error with full context
            self.logger.error(
                f"Error in {operation_name}: {str(exc_value)}",
                extra={"error_details": error_details}
            )

            # Re-raise the exception
            raise

# Usage Example
logger = logging.getLogger(__name__)
error_logger = ErrorLogger(logger)

with error_logger.error_context("user_authentication", user_id="123", attempt=2):
    # Your code that might raise an exception
    authenticate_user(user_id)
ログイン後にコピー
ログイン後にコピー
ログイン後にコピー

追加のリソース

  1. 公式ドキュメント:
  • Python のロギング方法
  • ロギングクックブック
  1. ツールとライブラリ:
  • Loguru ドキュメント
  • Structlog ドキュメント
  • Python-JSON-ロガー

このガイドでは、基本的なセットアップから高度な実装まで、Python ロギングの重要な側面を説明します。ロギングはアプリケーションの可観測性とメンテナンスに不可欠な部分であることに注意してください。最良の結果を得るために、慎重に実装し、定期的にメンテナンスしてください。

アプリケーションが進化し、新しい要件が出現するにつれて、ロギング実装を定期的に確認して更新することを忘れないでください。

以上が完全な Python ロギング ガイド: ベスト プラクティスと実装の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。

ソース:dev.to
このウェブサイトの声明
この記事の内容はネチズンが自主的に寄稿したものであり、著作権は原著者に帰属します。このサイトは、それに相当する法的責任を負いません。盗作または侵害の疑いのあるコンテンツを見つけた場合は、admin@php.cn までご連絡ください。
著者別の最新記事
人気のチュートリアル
詳細>
最新のダウンロード
詳細>
ウェブエフェクト
公式サイト
サイト素材
フロントエンドテンプレート