# python-resilience — 详细实操示例

## 进阶模式

### 模式 5：重试尝试日志记录

跟踪重试行为，用于调试和告警。

```python
from tenacity import retry, stop_after_attempt, wait_exponential
import structlog

logger = structlog.get_logger()

def log_retry_attempt(retry_state):
    """记录详细的重试信息。"""
    exception = retry_state.outcome.exception()
    logger.warning(
        "正在重试操作",
        attempt=retry_state.attempt_number,
        exception_type=type(exception).__name__,
        exception_message=str(exception),
        next_wait_seconds=retry_state.next_action.sleep if retry_state.next_action else None,
    )

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, max=10),
    before_sleep=log_retry_attempt,
)
def call_with_logging(request: dict) -> dict:
    """带有重试日志的外部调用。"""
    ...
```

### 模式 6：超时装饰器

创建可复用的超时装饰器，以实现一致的超时处理。

```python
import asyncio
from functools import wraps
from typing import TypeVar, Callable

T = TypeVar("T")

def with_timeout(seconds: float):
    """为异步函数添加超时功能的装饰器。"""
    def decorator(func: Callable[..., T]) -> Callable[..., T]:
        @wraps(func)
        async def wrapper(*args, **kwargs) -> T:
            return await asyncio.wait_for(
                func(*args, **kwargs),
                timeout=seconds,
            )
        return wrapper
    return decorator

@with_timeout(30)
async def fetch_with_timeout(url: str) -> dict:
    """以 30 秒超时获取 URL。"""
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.json()
```

### 模式 7：通过装饰器实现横切关注点

堆叠装饰器，将基础设施与业务逻辑分离。

```python
from functools import wraps
from typing import TypeVar, Callable
import structlog

logger = structlog.get_logger()
T = TypeVar("T")

def traced(name: str | None = None):
    """为函数调用添加追踪。"""
    def decorator(func: Callable[..., T]) -> Callable[..., T]:
        span_name = name or func.__name__

        @wraps(func)
        async def wrapper(*args, **kwargs) -> T:
            logger.info("操作已开始", operation=span_name)
            try:
                result = await func(*args, **kwargs)
                logger.info("操作已完成", operation=span_name)
                return result
            except Exception as e:
                logger.error("操作失败", operation=span_name, error=str(e))
                raise
        return wrapper
    return decorator

# 堆叠多个关注点
@traced("fetch_user_data")
@with_timeout(30)
@retry(stop=stop_after_attempt(3), wait=wait_exponential_jitter())
async def fetch_user_data(user_id: str) -> dict:
    """获取用户信息，包含追踪、超时和重试。"""
    ...
```

### 模式 8：为可测试性而做的依赖注入

通过构造函数传入基础设施组件，便于测试。

```python
from dataclasses import dataclass
from typing import Protocol

class Logger(Protocol):
    def info(self, msg: str, **kwargs) -> None: ...
    def error(self, msg: str, **kwargs) -> None: ...

class MetricsClient(Protocol):
    def increment(self, metric: str, tags: dict | None = None) -> None: ...
    def timing(self, metric: str, value: float) -> None: ...

@dataclass
class UserService:
    """注入了基础设施的服务。"""

    repository: UserRepository
    logger: Logger
    metrics: MetricsClient

    async def get_user(self, user_id: str) -> User:
        self.logger.info("正在获取用户", user_id=user_id)
        start = time.perf_counter()

        try:
            user = await self.repository.get(user_id)
            self.metrics.increment("user.fetch.success")
            return user
        except Exception as e:
            self.metrics.increment("user.fetch.error")
            self.logger.error("获取用户失败", user_id=user_id, error=str(e))
            raise
        finally:
            elapsed = time.perf_counter() - start
            self.metrics.timing("user.fetch.duration", elapsed)

# 使用假对象轻松测试
service = UserService(
    repository=FakeRepository(),
    logger=FakeLogger(),
    metrics=FakeMetrics(),
)
```

### 模式 9：安全降级默认值

当非关键操作失败时优雅降级。

```python
from typing import TypeVar
from collections.abc import Callable

T = TypeVar("T")

def fail_safe(default: T, log_failure: bool = True):
    """失败时返回默认值，而非抛出异常。"""
    def decorator(func: Callable[..., T]) -> Callable[..., T]:
        @wraps(func)
        async def wrapper(*args, **kwargs) -> T:
            try:
                return await func(*args, **kwargs)
            except Exception as e:
                if log_failure:
                    logger.warning(
                        "操作失败，正在使用默认值",
                        function=func.__name__,
                        error=str(e),
                    )
                return default
        return wrapper
    return decorator

@fail_safe(default=[])
async def get_recommendations(user_id: str) -> list[str]:
    """获取推荐，失败时返回空列表。"""
    ...
```