我感觉像个小丑，拼接5个库只为构建一个弹性API客户端

Source: Dev.to

于是我写了一个统一所有功能的客户端。

我只想要一个简单的 API 客户端。

import httpx

async def fetch_user(user_id: str):
    async with httpx.AsyncClient() as client:
        r = await client.get(f"https://api.example.com/users/{user_id}")
        return r.json()

这段代码大约只能坚持 5 分钟。

因为真实的 API：

• 会对你进行速率限制
• 会超时
• 会返回 503 错误
• 有时会彻底宕机
• 重试甚至可能对你自己的服务造成 DDoS

于是我做了每个 Python 开发者都会做的事：开始堆叠各种库。

装饰器灾难塔

首先：速率限制。
然后：重试。
再然后：熔断器。

@breaker
@retry(...)
@sleep_and_retry
@limits(...)
async def fetch_user(...):
    ...

我讨厌它。

不是因为它不起作用，而是因为它无法扩展。

问题

• 超过 3 个库
• 脆弱的装饰器顺序
• 冲突的抽象
• 异步怪癖
• 痛苦的测试
• 分散的可观测性
• 依赖蔓延

而这只是针对 一个函数。想象一下 10 个 API、每用户限制、后台任务、Webhook。你不再编写业务逻辑，而是在照看弹性粘合代码。

思路：弹性如管道

如果弹性不是装饰器的堆砌呢？如果每一次调用都通过一个 单一编排器 流动？

from limitpal import (
    AsyncResilientExecutor,
    AsyncTokenBucket,
    RetryPolicy,
    CircuitBreaker,
)

executor = AsyncResilientExecutor(
    limiter=AsyncTokenBucket(capacity=10, refill_rate=100 / 60),
    retry_policy=RetryPolicy(max_attempts=3),
    circuit_breaker=CircuitBreaker(failure_threshold=5),
)

result = await executor.run("user:123", api_call)

没有装饰器。没有堆叠的库。没有脆弱的胶水。只有一个执行管道。这就是 LimitPal 的理念。

LimitPal 实际提供的功能

LimitPal 是一个用于构建弹性 Python 客户端和服务的工具包。它结合了：

• 速率限制（令牌桶 / 漏桶）
• 带指数退避 + 抖动的重试
• 熔断器
• 可组合的限制器
• 一个协调一切的弹性执行器

并且还提供：

• 完全的异步 + 同步支持
• 零依赖
• 默认线程安全
• 用于测试的确定性时间控制
• 基于键的隔离（每用户 / 每 IP / 每租户）

目标不是增加更多功能，而是 减少活动部件。

弹性管道（核心理念）

每次调用都会经过：

Circuit breaker check
→ Rate limiter
→ Execute + retry loop
→ Record result

顺序很重要。你并不是仅仅“添加重试”；而是在设计系统的故障行为：

• 断路器阻止级联故障
• 限流器保护基础设施
• 重试处理临时问题
• 执行器保持一致性

一个思维模型取代五个。

没有人谈论的测试问题

基于时间的逻辑非常难以测试。

传统做法：

time.sleep(1)

慢。易碎。非确定性。

LimitPal 使用可插拔时钟，使测试变为：

clock.advance(1.0)

瞬间。确定性。你可以在毫秒内模拟几分钟的重试。对于关注可靠性的团队来说，这是一场游戏规则的改变。

实际示例

一个约 10 行的弹性 HTTP 客户端：

executor = AsyncResilientExecutor(
    limiter=AsyncTokenBucket(capacity=10, refill_rate=100 / 60),
    retry_policy=RetryPolicy(max_attempts=3, base_delay=0.5),
    circuit_breaker=CircuitBreaker(failure_threshold=5),
)

async def fetch():
    return await httpx.get("https://api.example.com")

result = await executor.run("api", fetch)

您将自动获得：

• 突发控制
• 指数重试
• 级联故障保护
• 简洁的异步语义

没有装饰器层叠。

何时使用它？

如果你：

• 构建 API 客户端
• 调用不稳定的第三方服务
• 运行后台任务
• 需要每用户的限制
• 在意确定性的测试
• 想要干净的异步支持

如果你只需要重试，较小的库就足够。如果你需要 组合，这就是它的定位。

第2部分：内部实现

这篇文章讨论的是概念。在第2部分，我将深入探讨：

• 执行器流水线的工作方式
• 断路器状态机
• 时钟抽象设计
• 复合限制器架构
• 故障建模

因为弹性并非魔法，而是架构。

试一试

pip install limitpal

文档：
仓库：