wehub-skill-sync
187 lines
5.2 KiB
Markdown
187 lines
5.2 KiB
Markdown
# 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]:
|
|
"""获取推荐，失败时返回空列表。"""
|
|
...
|
|
```
|