chore: import zh skill python-resilience

2026-07-13 21:36:56 +08:00
commit 36f7cde2cb
3 changed files with 390 additions and 0 deletions
@@ -0,0 +1,9 @@
+# WeHub 来源说明
+
+- Skill 名称：`python-resilience`
+- 中文类目：Python 运行时加固与错误处理
+- 上游仓库：`wshobson__agents`
+- 上游路径：`plugins/python-development/skills/python-resilience/SKILL.md`
+- 上游链接：https://github.com/wshobson/agents/blob/HEAD/plugins/python-development/skills/python-resilience/SKILL.md
+- 本仓库为 WeHub 中文 Skill 汉化包，基于 skill 市场筛选 Top200 清单整理
+- 原作者、版权和许可证信息以上游仓库为准
@@ -0,0 +1,195 @@
+---
+name: python-resilience
+description: Python 弹性模式，包括自动重试、指数退避、超时和容错装饰器。在需要添加重试逻辑、实现超时、构建容错服务或处理瞬时故障时使用。
+---
+
+# Python 弹性模式
+
+构建容错的 Python 应用程序，优雅地处理瞬时故障、网络问题和服务中断。当依赖项不可靠时，弹性模式能让系统保持运行。
+
+## 何时使用本技能
+
+- 为外部服务调用添加重试逻辑
+- 为网络操作实现超时
+- 构建容错的微服务
+- 处理速率限制和背压
+- 创建基础设施装饰器
+- 设计断路器
+
+## 核心概念
+
+### 1. 瞬时故障与永久故障
+
+重试瞬时错误（网络超时、临时服务问题）。不要重试永久错误（无效凭据、错误请求）。
+
+### 2. 指数退避
+
+增加重试之间的等待时间，避免让正在恢复的服务不堪重负。
+
+### 3. 抖动
+
+为退避算法添加随机性，防止多个客户端同时重试时产生惊群效应。
+
+### 4. 有界重试
+
+同时限制重试次数和总持续时间，防止无限重试循环。
+
+## 快速开始
+
+```python
+from tenacity import retry, stop_after_attempt, wait_exponential_jitter
+
+@retry(
+    stop=stop_after_attempt(3),
+    wait=wait_exponential_jitter(initial=1, max=10),
+)
+def call_external_service(request: dict) -> dict:
+    return httpx.post("https://api.example.com", json=request).json()
+```
+
+## 基础模式
+
+### 模式 1：使用 Tenacity 的基础重试
+
+使用 `tenacity` 库实现生产级重试逻辑。对于更简单的场景，可考虑内置重试功能或轻量级自定义实现。
+
+```python
+from tenacity import (
+    retry,
+    stop_after_attempt,
+    stop_after_delay,
+    wait_exponential_jitter,
+    retry_if_exception_type,
+)
+
+TRANSIENT_ERRORS = (ConnectionError, TimeoutError, OSError)
+
+@retry(
+    retry=retry_if_exception_type(TRANSIENT_ERRORS),
+    stop=stop_after_attempt(5) | stop_after_delay(60),
+    wait=wait_exponential_jitter(initial=1, max=30),
+)
+def fetch_data(url: str) -> dict:
+    """在瞬时故障时自动重试获取数据。"""
+    response = httpx.get(url, timeout=30)
+    response.raise_for_status()
+    return response.json()
+```
+
+### 模式 2：仅重试适当的错误
+
+白名单特定的瞬时异常。绝不重试：
+
+- `ValueError`、`TypeError`——这些是代码缺陷，不是瞬时问题
+- `AuthenticationError`——无效凭据不会自动变有效
+- HTTP 4xx 错误（429 除外）——客户端错误是永久性的
+
+```python
+from tenacity import retry, retry_if_exception_type
+import httpx
+
+# 定义哪些是可重试的
+RETRYABLE_EXCEPTIONS = (
+    ConnectionError,
+    TimeoutError,
+    httpx.ConnectTimeout,
+    httpx.ReadTimeout,
+)
+
+@retry(
+    retry=retry_if_exception_type(RETRYABLE_EXCEPTIONS),
+    stop=stop_after_attempt(3),
+    wait=wait_exponential_jitter(initial=1, max=10),
+)
+def resilient_api_call(endpoint: str) -> dict:
+    """在出现网络问题时带重试的 API 调用。"""
+    return httpx.get(endpoint, timeout=10).json()
+```
+
+### 模式 3：HTTP 状态码重试
+
+重试表示瞬时问题的特定 HTTP 状态码。
+
+```python
+from tenacity import retry, retry_if_result, stop_after_attempt
+import httpx
+
+RETRY_STATUS_CODES = {429, 502, 503, 504}
+
+def should_retry_response(response: httpx.Response) -> bool:
+    """检查响应是否表示可重试的错误。"""
+    return response.status_code in RETRY_STATUS_CODES
+
+@retry(
+    retry=retry_if_result(should_retry_response),
+    stop=stop_after_attempt(3),
+    wait=wait_exponential_jitter(initial=1, max=10),
+)
+def http_request(method: str, url: str, **kwargs) -> httpx.Response:
+    """在瞬时状态码上带重试的 HTTP 请求。"""
+    return httpx.request(method, url, timeout=30, **kwargs)
+```
+
+### 模式 4：异常与状态码组合重试
+
+同时处理网络异常和 HTTP 状态码。
+
+```python
+from tenacity import (
+    retry,
+    retry_if_exception_type,
+    retry_if_result,
+    stop_after_attempt,
+    wait_exponential_jitter,
+    before_sleep_log,
+)
+import logging
+import httpx
+
+logger = logging.getLogger(__name__)
+
+TRANSIENT_EXCEPTIONS = (
+    ConnectionError,
+    TimeoutError,
+    httpx.ConnectError,
+    httpx.ReadTimeout,
+)
+RETRY_STATUS_CODES = {429, 500, 502, 503, 504}
+
+def is_retryable_response(response: httpx.Response) -> bool:
+    return response.status_code in RETRY_STATUS_CODES
+
+@retry(
+    retry=(
+        retry_if_exception_type(TRANSIENT_EXCEPTIONS) |
+        retry_if_result(is_retryable_response)
+    ),
+    stop=stop_after_attempt(5),
+    wait=wait_exponential_jitter(initial=1, max=30),
+    before_sleep=before_sleep_log(logger, logging.WARNING),
+)
+def robust_http_call(
+    method: str,
+    url: str,
+    **kwargs,
+) -> httpx.Response:
+    """带全面重试处理的 HTTP 调用。"""
+    return httpx.request(method, url, timeout=30, **kwargs)
+```
+
+## 详细的完整示例与模式
+
+详细章节（以 `## 高级模式` 开头）位于 `references/details.md` 中。当以上导航摘要不足以满足需求时，请阅读该文件。
+
+## 最佳实践总结
+
+1. **仅重试瞬时错误**——不要重试代码缺陷或认证失败
+2. **使用指数退避**——给服务恢复的时间
+3. **添加抖动**——防止同步重试导致惊群效应
+4. **限制总时长**——`stop_after_attempt(5) | stop_after_delay(60)`
+5. **记录每次重试**——静默重试会隐藏系统性问题
+6. **使用装饰器**——将重试逻辑与业务逻辑分离
+7. **注入依赖**——使基础设施可测试
+8. **处处设置超时**——每个网络调用都需要超时
+9. **优雅降级**——为非关键路径返回缓存/默认值
+10. **监控重试率**——高重试率表明存在根本性问题
@@ -0,0 +1,186 @@
+# 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]:
+    """获取推荐，失败时返回空列表。"""
+    ...
+```