rate-limiter/
├── config/
│   └── rules.yaml          # 限流规则配置
├── src/
│   ├── __init__.py
│   ├── limiter.py          # 核心限流器实现
│   ├── middleware.py       # HTTP 中间件集成
│   └── utils.py            # 工具函数
├── tests/
│   ├── test_limiter.py
│   └── test_middleware.py
├── docker-compose.yml       # 本地开发环境
├── requirements.txt
└── README.md

import time
from typing import Optional
import aioredis


class SlidingWindowLimiter:
    """基于 Redis ZSet 的滑动窗口限流器"""

    def __init__(
        self,
        redis_url: str = "redis://localhost:6379/0",
        fallback: str = "allow"  # Redis 不可用时的策略: allow / deny
    ):
        self.redis_url = redis_url
        self.fallback = fallback
        self._redis: Optional[aioredis.Redis] = None

    async def _get_redis(self) -> aioredis.Redis:
        if self._redis is None:
            self._redis = await aioredis.from_url(
                self.redis_url, encoding="utf-8"
            )
        return self._redis

    async def is_allowed(
        self,
        key: str,
        max_requests: int,
        window_seconds: float
    ) -> bool:
        """
        判断请求是否被允许。

        Args:
            key: 限流维度标识（如 user:123 或 ip:10.0.0.1）
            max_requests: 时间窗口内允许的最大请求数
            window_seconds: 滑动窗口大小（秒）

        Returns:
            True 表示允许通过，False 表示被限流。
        """
        try:
            redis = await self._get_redis()
        except Exception:
            return self.fallback == "allow"

        now = time.time()
        window_start = now - window_seconds
        pipe = redis.pipeline()

        # 1. 移除窗口外的旧记录（ZREMRANGEBYSCORE）
        pipe.zremrangebyscore(key, 0, window_start)
        # 2. 统计窗口内已有请求数（ZCARD）
        pipe.zcard(key)
        # 3. 设置过期时间，防止 key 永久占用内存（EXPIRE）
        pipe.expire(key, int(window_seconds) + 1)

        results = await pipe.execute()
        current_count = results[1]

        if current_count >= max_requests:
            return False

        # 4. 记录本次请求（ZADD，score 为时间戳）
        await redis.zadd(key, {f"{now}:{id(self)}": now})
        return True

    async def get_remaining(
        self,
        key: str,
        max_requests: int,
        window_seconds: float
    ) -> int:
        """获取当前剩余可用请求数，用于返回 RateLimit-Remaining 响应头"""
        try:
            redis = await self._get_redis()
        except Exception:
            return max_requests

        now = time.time()
        window_start = now - window_seconds
        count = await redis.zcount(key, window_start, now)
        return max(0, max_requests - count)

from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware
from src.limiter import SlidingWindowLimiter
import yaml


class RateLimitMiddleware(BaseHTTPMiddleware):
    """FastAPI 限流中间件，从 YAML 配置读取规则"""

    def __init__(self, app, config_path: str = "config/rules.yaml"):
        super().__init__(app)
        self.limiter = SlidingWindowLimiter()
        with open(config_path, "r") as f:
            self.rules = yaml.safe_load(f)

    async def dispatch(self, request: Request, call_next):
        # 提取限流 key：优先用用户 ID，其次用 IP 地址
        user_id = request.headers.get("X-User-Id")
        ip = request.client.host
        key = f"user:{user_id}" if user_id else f"ip:{ip}"

        # 匹配限流规则（按接口路径前缀）
        path = request.url.path
        rule = self._match_rule(path)

        if rule:
            allowed = await self.limiter.is_allowed(
                key=key,
                max_requests=rule["max_requests"],
                window_seconds=rule["window_seconds"]
            )
            if not allowed:
                return Response(
                    status_code=429,
                    content='{"error": "请求过于频繁，请稍后重试"}'
                )

            remaining = await self.limiter.get_remaining(
                key=key,
                max_requests=rule["max_requests"],
                window_seconds=rule["window_seconds"]
            )

        response = await call_next(request)

        # 注入标准限流响应头（遵循 RFC 6585）
        if rule:
            response.headers["X-RateLimit-Limit"] = str(rule["max_requests"])
            response.headers["X-RateLimit-Remaining"] = str(remaining)
            response.headers["X-RateLimit-Reset"] = str(
                int(time.time()) + rule["window_seconds"]
            )

        return response

    def _match_rule(self, path: str) -> Optional[dict]:
        """根据请求路径匹配最具体的限流规则"""
        best_match = None
        for rule in self.rules.get("routes", []):
            prefix = rule["path_prefix"]
            if path.startswith(prefix):
                if best_match is None or len(prefix) > len(best_match["path_prefix"]):
                    best_match = rule
        return best_match

# config/rules.yaml
default:
  max_requests: 100
  window_seconds: 60

routes:
  - path_prefix: "/api/v1/search"
    max_requests: 10
    window_seconds: 60
    description: "搜索接口 —— 消耗资源较大，严格限流"

  - path_prefix: "/api/v1/upload"
    max_requests: 5
    window_seconds: 60
    description: "文件上传接口 —— 带宽敏感"

  - path_prefix: "/api/v1/"
    max_requests: 100
    window_seconds: 60
    description: "通用 API 限流"

# 1. 克隆项目
git clone https://github.com/example/rate-limiter.git
cd rate-limiter

# 2. 启动 Redis 容器
docker-compose up -d redis

# 3. 安装 Python 依赖
pip install -r requirements.txt

# 4. 运行测试（确保环境正确）
pytest tests/ -v

# 5. 启动示例服务（带限流中间件）
uvicorn main:app --host 0.0.0.0 --port 8000

# 正常请求（前 10 次应返回 200）
for i in {1..10}; do
  curl -s -o /dev/null -w "%{http_code}\n" http://localhost:8000/api/v1/search

# 第 11 次请求应返回 429
curl -v http://localhost:8000/api/v1/search
# 预期响应头：
# HTTP/1.1 429 Too Many Requests
# X-RateLimit-Limit: 10
# X-RateLimit-Remaining: 0
# X-RateLimit-Reset: 1714234560