快速开始

本指南帮助你在 5 分钟内为 FastAPI 应用添加限流能力。

前置要求

• Python >= 3.11
• FastAPI 项目（或新建一个）

安装

pip install olc[fastapi]

这会同时安装 OLC 核心库和 FastAPI 适配器。

最小示例

1. 创建配置文件

创建配置目录，例如 config/，在其中创建两个文件：

config/overload-config.properties：

olc.sdk.domain=my-service
olc.sdk.switch=on
olc.sdk.config.stub=jsonfile
olc.sdk.policy.update.period=6

config/olc.json：

{
  "domain": "my-service",
  "rules": [
    {
      "group": {
        "priority": 4,
        "enabled": true,
        "name": "apiGroup",
        "tags": [
          {
            "tag": "URL",
            "match": "equal",
            "values": ["/api/chat"]
          }
        ]
      },
      "flow": {
        "name": "apiFlow",
        "enabled": true,
        "flowControlMode": "qps",
        "timeUnit": "second",
        "timeInterval": 1,
        "rateLimit": 10,
        "burstLimit": 10
      }
    }
  ]
}

这个配置表示：对 /api/chat 接口限制每秒最多 10 个请求。

2. 创建 FastAPI 应用

main.py：

import os

from fastapi import FastAPI
from fastapi.responses import JSONResponse

from olc.adapters.fastapi import OlcFastAPIAdapter

app = FastAPI()

# 设置配置文件路径
config_dir = os.path.join(os.path.dirname(__file__), "config")
os.environ["OLC_CONFIG_PATH"] = config_dir

# 添加 OLC 中间件
app.add_middleware(OlcFastAPIAdapter)


@app.post("/api/chat")
async def chat():
    return JSONResponse(content={"message": "Hello!"})


@app.get("/health")
async def health():
    return {"status": "healthy"}


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

3. 运行并验证

python main.py

正常请求：

curl -X POST http://localhost:8000/api/chat
# {"message":"Hello!"}

超过限流阈值后：

# 快速发送超过 10 次请求后
curl -X POST http://localhost:8000/api/chat
# {"error":"Too Many Requests","message":"Request blocked by rate limiter","path":"/api/chat"}

不限流的接口不受影响：

curl http://localhost:8000/health
# {"status":"healthy"}

自定义标签提取

默认情况下，OLC 只提取 URL 和 Method 两个维度。你可以自定义标签提取函数来添加更多维度：

from olc.adapters.fastapi import OlcFastAPIAdapter, OlcAdapterConfig
from olc.bean.dimension import URL, Method, Model

def extract_tags(request):
    return {
        f"{URL}": request.url.path,
        f"{Method}": request.method,
        f"{Model}": request.headers.get("X-Model", "default"),
    }

config = OlcAdapterConfig(tag_extractor=extract_tags)
app.add_middleware(OlcFastAPIAdapter, config=config)

然后在 olc.json 中就可以按 Model 维度限流：

{
  "group": {
    "name": "modelGroup",
    "tags": [
      {"tag": "URL", "match": "equal", "values": ["/api/chat"]},
      {"tag": "Model", "match": "equal", "values": ["gpt-4"]}
    ]
  },
  "flow": {
    "flowControlMode": "qps",
    "timeUnit": "second",
    "timeInterval": 1,
    "rateLimit": 5,
    "burstLimit": 5
  }
}

自定义拦截响应

默认返回 429 状态码和 JSON 错误信息。你可以自定义拦截响应：

from fastapi import Request
from fastapi.responses import JSONResponse
from olc.adapters.fastapi import OlcFastAPIAdapter, OlcAdapterConfig

def custom_block_response(request: Request, block_msg: str):
    return JSONResponse(
        status_code=429,
        content={
            "error": "Rate limit exceeded",
            "message": block_msg,
            "retry_after": 1,
        },
        headers={"Retry-After": "1"},
    )

config = OlcAdapterConfig(
    block_msg="请求过于频繁，请稍后重试",
    block_response_builder=custom_block_response,
)
app.add_middleware(OlcFastAPIAdapter, config=config)

配置文件说明

overload-config.properties

配置项	说明	示例
olc.sdk.domain	域名，需与 olc.json 中的 domain 一致	my-service
olc.sdk.switch	全局开关，on 启用，off 关闭	on
olc.sdk.config.stub	配置源类型，默认 jsonfile	jsonfile
olc.sdk.policy.update.period	规则刷新间隔（秒）	6

olc.json

顶层结构：

{
  "domain": "my-service",
  "rules": [
    {
      "group": { ... },      // 匹配规则
      "flow": { ... },       // 流控策略（可选）
      "admission": { ... }   // 准入策略（可选）
    }
  ]
}

详细的限流配置请参阅 限流策略使用指南，准入控制请参阅 准入控制指南。