From 36f7cde2cb69292b488ced9e072c6986a1f0fa8c Mon Sep 17 00:00:00 2001 From: wehub-skill-sync Date: Mon, 13 Jul 2026 21:36:56 +0800 Subject: [PATCH] chore: import zh skill python-resilience --- README.wehub.md | 9 ++ SKILL.md | 195 ++++++++++++++++++++++++++++++++++++++++++ references/details.md | 186 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 390 insertions(+) create mode 100644 README.wehub.md create mode 100644 SKILL.md create mode 100644 references/details.md diff --git a/README.wehub.md b/README.wehub.md new file mode 100644 index 0000000..acab495 --- /dev/null +++ b/README.wehub.md @@ -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 清单整理 +- 原作者、版权和许可证信息以上游仓库为准 diff --git a/SKILL.md b/SKILL.md new file mode 100644 index 0000000..43f6955 --- /dev/null +++ b/SKILL.md @@ -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. **监控重试率**——高重试率表明存在根本性问题 diff --git a/references/details.md b/references/details.md new file mode 100644 index 0000000..a1d2139 --- /dev/null +++ b/references/details.md @@ -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]: + """获取推荐，失败时返回空列表。""" + ... +```