コンテキストを保持しながら例外を記録する Python デコレータの構築

コンテキストを使用して例外をキャプチャしてログに記録するための Python デコレータの設計

このソリューションでは、バックエンド スクリプトに Python を使用し、元のコンテキストを保持しながら例外を処理するためのモジュール式で再利用可能な設計原則に重点を置いています。

import functools
import logging
# Define a custom decorator for error handling
def error_handler_decorator(func):
    @functools.wraps(func)
    async def wrapper(*args, kwargs):
        original_message = kwargs.get("eventHubMessage", "Unknown message")
        try:
            return await func(*args, kwargs)
        except Exception as e:
            logging.error(f"Error: {e}. Original message: {original_message}")
            # Re-raise with combined context
            raise Exception(f"{e} | Original message: {original_message}")
    return wrapper
# Example usage
@error_handler_decorator
async def main(eventHubMessage):
    data = json.loads(eventHubMessage)
    logging.info(f"Processing data: {data}")
    # Simulate potential error
    if not data.get("RequestID"):
        raise ValueError("Missing RequestID")
    # Simulate successful processing
    return "Processed successfully"
# Test
try:
    import asyncio
    asyncio.run(main(eventHubMessage='{"ProductType": "Test"}'))
except Exception as e:
    print(f"Caught exception: {e}")

クラスを使用した構造化エラー処理アプローチの作成

このソリューションでは、Python クラスベースのデコレータを使用してモジュール性と再利用性を向上させ、より構造化された方法で例外を管理します。

import logging
# Define a class-based decorator
class ErrorHandler:
    def __init__(self, func):
        self.func = func
    async def __call__(self, *args, kwargs):
        original_message = kwargs.get("eventHubMessage", "Unknown message")
        try:
            return await self.func(*args, kwargs)
        except Exception as e:
            logging.error(f"Error: {e}. Original message: {original_message}")
            raise Exception(f"{e} | Original message: {original_message}")
# Example usage
@ErrorHandler
async def process_event(eventHubMessage):
    data = json.loads(eventHubMessage)
    logging.info(f"Data: {data}")
    if "RequestType" not in data:
        raise KeyError("Missing RequestType")
    return "Event processed!"
# Test
try:
    import asyncio
    asyncio.run(process_event(eventHubMessage='{"RequestID": "123"}'))
except Exception as e:
    print(f"Caught exception: {e}")

ミドルウェアを活用したグローバル例外処理

このソリューションは、Python でミドルウェアのような構造を実装し、複数の関数呼び出しにわたる例外の集中処理を可能にします。

import logging
async def middleware(handler, message):
    try:
        return await handler(message)
    except Exception as e:
        logging.error(f"Middleware caught error: {e} | Message: {message}")
        raise
# Handlers
async def handler_one(message):
    if not message.get("ProductType"):
        raise ValueError("Missing ProductType")
    return "Handler one processed."
# Test middleware
message = {"RequestID": "123"}
try:
    import asyncio
    asyncio.run(middleware(handler_one, message))
except Exception as e:
    print(f"Middleware exception: {e}")

Python での例外処理に関するよくある質問

1. 例外処理にデコレータを使用する目的は何ですか?
2. デコレータなど @error_handler_decorator、複数の機能にわたるエラーのログ記録と処理を一元化します。これにより、例外の一貫した処理が保証され、元のメッセージのような重要なコンテキストが保持されます。
3. どのようにして httpx.AsyncClient API インタラクションを改善するには?
4. これにより、非同期 HTTP 要求が有効になり、プログラムが複数の API 呼び出しを同時に処理できるようになります。これは、Azure Functions のような高スループット システムにとって重要です。
5. 構造化ログの利点は何ですか?
6. JSON ログのような構造化されたログ形式により、Azure Monitor や Splunk などのツールを使用してリアルタイムでエラーを分析および監視することが容易になります。
7. 一時的なエラーを効果的に管理するにはどうすればよいでしょうか?
8. 指数バックオフを使用した再試行ロジックと、失敗をキャプチャするデコレーターを実装することで、一時的な問題が永続的なエラーにつながることがなくなります。
9. 例外処理において元のコンテキストを維持することが重要なのはなぜですか?
10. 処理中のペイロードなど、元のメッセージを保存すると、特に分散システムにおいて問題のデバッグや追跡に貴重な情報が得られます。

Python イベント処理のエラー耐性をマスターする

Azure Functions などの分散システムでの例外処理は、中断のない操作を確保するために重要です。エラーをデコレーターでラップし、元のコンテキストを保持することにより、開発者はデバッグを簡素化し、システムの透過性を合理化します。このアプローチは、問題が避けられない動的な現実の環境で特に役立ちます。

非同期プログラミングや構造化ログなどの高度な技術を組み合わせることで、Python は回復力のあるシステムを作成するための強力なツールになります。これらのソリューションは、一時的なエラーに効果的に対処することで、トラブルシューティングの時間を節約し、パフォーマンスを向上させます。これらのプラクティスを採用すると、開発者は堅牢でスケーラブルなアプリケーションを構築できるようになり、日常的な課題を管理しやすくなります。 🛠️