正文

掌握FastAPI高效限制请求：5招轻松防范滥用，保障服务稳定运行

/0 浏览量
0828

在当今的互联网环境中，API服务面临着不断增长的请求量，这既带来了机遇，也带来了挑战。如何有效地限制请求，防止滥用，保障服务的稳定运行，是每个开发者都需要面对的问题。FastAPI作为一款流行的Python Web框架，提供了多种方法来限制请求。以下是五招轻松防范滥用，保障服务稳定运行的方法。

1. 使用依赖注入系统限制请求频率

FastAPI的依赖注入系统允许你在每个请求的生命周期中注入函数或对象。你可以利用这个特性来创建一个简单的频率限制器。

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from functools import lru_cache

app = FastAPI()

# 使用lru_cache缓存结果，限制请求频率
@lru_cache(maxsize=100)
def is_request_allowed(request: Request):
    # 这里可以添加更复杂的逻辑，比如使用Redis等存储来记录请求次数
    return True

@app.middleware("http")
async def limit_requests(request: Request, call_next):
    if not is_request_allowed(request):
        return JSONResponse({"message": "Too many requests"}, status_code=429)
    response = await call_next(request)
    return response

2. 利用HTTP头信息限制请求

你可以通过检查HTTP头信息来限制请求。例如，可以使用X-RateLimit-Limit和X-RateLimit-Remaining来告知客户端请求的频率限制。

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse

app = FastAPI()

# 定义请求频率限制
LIMIT = 100
WINDOW = 60  # 60秒内限制100次请求

@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    # 这里可以添加更复杂的逻辑，比如使用Redis等存储来记录请求次数
    if request.headers.get("X-RateLimit-Remaining") == "0":
        raise HTTPException(status_code=429, detail="Rate limit exceeded")
    response = await call_next(request)
    return response

3. 使用外部服务进行请求限制

对于更复杂的场景，可以使用外部服务如Redis、Memcached等来记录请求次数。以下是一个使用Redis的示例：

from fastapi import FastAPI, Request
import redis

app = FastAPI()
redis_client = redis.StrictRedis(host='localhost', port=6379, db=0)

@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    client_ip = request.client.host
    key = f"rate_limit:{client_ip}"
    try:
        current_count = int(redis_client.get(key) or 0)
    except redis.RedisError:
        raise HTTPException(status_code=500, detail="Internal server error")
    if current_count >= 100:
        raise HTTPException(status_code=429, detail="Rate limit exceeded")
    redis_client.incr(key)
    redis_client.expire(key, 60)  # 设置过期时间为60秒
    response = await call_next(request)
    return response

4. 集成第三方库

市面上有许多第三方库可以帮助你实现请求限制，例如slowapi和flask-limiter。这些库通常提供了更多的功能和灵活性。

from slowapi import Limiter, _request_id
from slowapi.errors import RateLimitExceeded

app = FastAPI()
limiter = Limiter(key_func=lambda request: request.client.host, default_limits=["100 per minute"])

@app.get("/items/")
@limiter.limit("10 per minute")
async def read_items():
    return {"message": "This endpoint is rate limited to 10 requests per minute."}

5. 监控和报警

除了限制请求频率，还需要对API的使用情况进行监控。通过日志记录、性能监控和报警系统，可以及时发现并处理异常情况。

from fastapi import FastAPI, Request
import logging

app = FastAPI()

# 配置日志记录
logging.basicConfig(level=logging.INFO)

@app.middleware("http")
async def log_requests(request: Request, call_next):
    logging.info(f"Request from {request.client.host}")
    response = await call_next(request)
    return response

通过以上五种方法，你可以有效地限制FastAPI服务的请求，防止滥用，保障服务的稳定运行。在实际应用中，可以根据具体需求和场景选择合适的方法。

-- 展开阅读全文 --