构建一个 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}")

1. 有关在 Python 中处理异常的内容受到 Python 官方文档的启发。欲了解更多信息，请访问 Python 异常文档 。
2. 有关异步 HTTP 客户端的详细信息基于 httpx库官方文档 ，这解释了其非阻塞 HTTP 请求的功能。
3. 结构化日志记录的原则以以下见解为指导： Azure 监视器 ，分布式系统中集中日志记录的工具。
4. 有关包装 Python 函数的装饰器的指南由以下教程提供： 真正的Python 。
5. 了解瞬态错误和重试机制基于以下文章 AWS 架构博客 ，讨论分布式环境中的错误恢复能力。