接口限流

youlai-django 提供基于 IP 的接口限流机制，防止恶意请求和资源滥用。

限流方式

方式	说明	适用场景
IP 限流	基于 IP 地址的访问频率限制	短信验证码、邮箱验证码等敏感接口

使用方式

装饰器方式

使用 @ip_rate_limit 装饰器对视图方法进行限流：

from system.utils.rate_limit import ip_rate_limit

class UserViewSet(BaseModelViewSet):
    
    @ip_rate_limit(60)  # 60秒内同一IP只能访问一次
    @action(methods=["post"], detail=False, url_path=r"mobile/code")
    def mobile_code(self, request, *args, **kwargs):
        """发送手机验证码"""
        # 业务逻辑
        pass

参数说明：

参数	类型	默认值	说明
seconds	int	60	限制时间（秒），在此时间内同一 IP 只能访问一次

应用场景

短信验证码接口

@ip_rate_limit(60)  # 60秒限制
@extend_schema(summary="发送手机验证码")
@action(methods=["post"], detail=False, url_path=r"mobile/code")
def mobile_code(self, request, *args, **kwargs):
    mobile = request.query_params.get('mobile')
    # 发送验证码逻辑
    pass

限流效果：

• 同一 IP 在 60 秒内只能请求一次
• 超出限制返回：操作频繁，请{剩余秒数}秒后再试

邮箱验证码接口（未限流对比）

@extend_schema(summary="发送邮箱验证码")
@action(methods=["post"], detail=False, url_path=r"email/code")
def email_code(self, request, *args, **kwargs):
    email = request.query_params.get('email')
    # 发送验证码逻辑
    pass

说明：邮箱验证码接口未添加限流，建议补充以防止滥用。

实现原理

Redis Key 设计

Key: rate_limit:{view_func_name}:{ip}
Value: 1
TTL: {seconds}

示例：

rate_limit:mobile_code:192.168.1.100

限流流程

核心代码

def ip_rate_limit(seconds: int = 60):
    """IP访问频率限制装饰器"""
    
    def decorator(view_func):
        @wraps(view_func)
        def wrapper(self, request: HttpRequest, *args, **kwargs):
            # 获取客户端 IP
            x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
            if x_forwarded_for:
                ip = x_forwarded_for.split(',')[0]
            else:
                ip = request.META.get('REMOTE_ADDR')
            
            # Redis Key
            redis_conn = get_redis_connection("default")
            redis_key = f"rate_limit:{view_func.__name__}:{ip}"
            
            # 检查是否在限制时间内
            if redis_conn.exists(redis_key):
                ttl = redis_conn.ttl(redis_key)
                return error(f"操作频繁，请{ttl}秒后再试", code="A0502", status=400)
            
            # 设置限制
            redis_conn.setex(redis_key, seconds, 1)
            
            # 执行视图函数
            return view_func(self, request, *args, **kwargs)
        
        return wrapper
    
    return decorator

错误响应

当触发限流时，返回以下响应：

{
  "code": "A0502",
  "msg": "操作频繁，请45秒后再试",
  "data": null
}

错误码说明：

错误码	说明
A0502	用户请求频率过高

最佳实践

1. 合理设置限流时间

接口类型	建议限流时间	说明
短信验证码	60 秒	防止短信轰炸
邮箱验证码	60 秒	防止邮件滥用
敏感操作	30-120 秒	根据业务场景调整

2. 多维度限流

当前仅支持 IP 限流，建议扩展：

用户级限流：

# 基于 user_id 的限流
redis_key = f"rate_limit:{view_func.__name__}:user:{user_id}"

全局限流：

# 基于接口的全局限流
redis_key = f"rate_limit:{view_func.__name__}:global"

3. 配置化限流参数

建议将限流时间配置化：

# settings.py
RATE_LIMIT_CONFIG = {
    'sms_code': 60,
    'email_code': 60,
    'sensitive_action': 30,
}

# 使用
@ip_rate_limit(settings.RATE_LIMIT_CONFIG['sms_code'])

相关文件

文件	说明
system/utils/rate_limit.py	IP 限流装饰器实现
system/users/views.py	限流使用示例（短信验证码接口）