在FastAPI中实现基于IP地址的接口限流

要实现基于IP地址的接口限流，可以使用FastAPI提供的装饰器和依赖项。

首先，我们创建一个依赖项，用于获取请求的IP地址：

from fastapi import Request, Depends
from ipaddress import ip_address

def get_client_ip(request: Request) -> str:
    ip = request.client.host
    return str(ip_address(ip))

然后，我们创建一个装饰器，用于限制特定IP地址的请求频率：

from fastapi import HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from functools import wraps
from datetime import datetime, timedelta
from collections import deque

class RateLimit:
    def __init__(self, limit: int, period: int):
        self.limit = limit
        self.period = period
        self.history = deque([], maxlen=limit)
    
    def __call__(self, func):
        @wraps(func)
        async def wrapper(request: Request, bearer: HTTPAuthorizationCredentials = Depends(HTTPBearer())):
            client_ip = get_client_ip(request)
            now = datetime.now().timestamp()
            count = len([t for t in self.history if t > now - self.period])
            if count >= self.limit:
                raise HTTPException(status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail="Rate limit exceeded.")
            self.history.append(now)
            return await func(request, bearer)
        return wrapper

这个装饰器使用一个双端队列来保存最近的请求时间戳，如果客户端IP的请求数超过了限制，则返回429 Too Many Requests错误响应。可以通过传入参数limit和period来设置限制条件。

最后，我们可以将这个装饰器应用到需要进行限流的接口上：

app = FastAPI()

@app.get("/limited")
@RateLimit(limit=10, period=60)
async def limited_endpoint(request: Request, bearer: HTTPAuthorizationCredentials = Depends(HTTPBearer())):
    return {"message": "You are under the rate limit."}

在这个例子中，/limited接口的请求频率限制为每分钟最多10次。如果相同IP地址的客户端在一分钟内超过了这个限制，它们将收到一个429 Too Many Requests错误响应。否则，接口会返回一个成功消息。