---
url: /fastapi_best_architecture_docs/backend/reference/limit.md
---
在现代 Web 开发中，API 限流（Rate
Limiting）是保护后端服务、防止资源滥用、保证服务稳定性的重要机制，我们有一个关于路由器的历史讨论，如果你感兴趣，可以查看：[#70](https://github.com/fastapi-practices/fastapi-best-architecture/discussions/70)

## 处理流程

以下是 RateLimiter 处理一次请求的完整流程：

```mermaid
graph TD
    A[请求进入路由依赖] --> B[初始化 Bucket 和 Limiter]
    B --> C[获取 Identifier]
    C --> D[异步尝试获取]
    D -->|获取成功| E[放行请求，继续业务处理]
    D -->|获取失败| F[计算 Retry-After]
    F --> G[执行 Callback<br>（默认抛出 429 异常）]
```

## 使用方法

RateLimiter 设计为 FastAPI 的依赖项，直接在路由中使用 `Depends` 注入

### 单规则限流

```python
# 每分钟最多 60 次
from fastapi import Depends
from pyrate_limiter import Duration, Rate

from backend.utils.limiter import RateLimiter


@app.get(
    "/api/example", 
    dependencies=[Depends(RateLimiter(Rate(60, Duration.MINUTE)))]
)
async def example():
    return {"message": "success"}
```

### 多规则复合限流

```python
# 每秒 10 次 + 每分钟 100 次
from fastapi import Depends
from pyrate_limiter import Duration, Rate

from backend.utils.limiter import RateLimiter


@app.post(
    "/api/heavy", 
    dependencies=[
        Depends(
            RateLimiter(
                Rate(10, Duration.SECOND),
                Rate(100, Duration.MINUTE),
            )
        )
    ]
)
async def heavy_endpoint():
    return {"status": "ok"}
```

### 自定义 Identifier

```python
from fastapi import Depends, Request
from pyrate_limiter import Duration, Rate

from backend.utils.limiter import RateLimiter


async def user_identifier(request: Request) -> str:
    return f'user:{request.user.id}'


@app.get(
    "/api/user-data",
    dependencies=[
        Depends(
            RateLimiter(
                Rate(50, Duration.MINUTE),
                identifier=user_identifier,
            )
        )
    ]    
)
async def user_data():
    return {"data": "protected"}
```