权限缓存

说明

通过 Redis 实现用户权限和菜单的缓存，采用用户级缓存 + 全局版本号双层策略。权限查询函数位于 application/menu/services.py，供 PermissionRequired 中间件调用。

缓存策略

采用「用户级缓存 + 全局版本号」双层策略：

1. 每个用户的缓存 Key 带版本号后缀
2. 版本号存全局 Key（perm:version）
3. 菜单变更时递增全局版本号，所有用户缓存自然失效
4. 用户角色变更时，直接删除该用户缓存
5. 角色菜单变更时，查询该角色下所有用户 ID，批量删除缓存

缓存 Key 结构

_VERSION_KEY = 'perm:version'       # 全局版本号
_PERM_PREFIX = 'perm:list:'         # 用户权限列表
_MENU_PREFIX = 'perm:menu:'         # 用户菜单树

Key 格式示例：

perm:version → 5
perm:list:1001:5 → ["sys:user:page", "sys:user:add", ...]
perm:menu:1001:5 → [{id: 1, name: "系统管理", children: [...]}]

Redis 连接配置

Redis 通过 Django 缓存框架 django_redis 管理：

# application/settings.py
CACHES = {
    "default": {
        "BACKEND": "django_redis.cache.RedisCache",
        "LOCATION": f"redis://{env.REDIS_HOST}:{env.REDIS_PORT}/{env.REDIS_DB}",
        "OPTIONS": {
            "CLIENT_CLASS": "django_redis.client.DefaultClient",
            "PASSWORD": env.REDIS_PASSWORD if env.REDIS_PASSWORD else None,
            "CONNECTION_POOL_KWARGS": {"max_connections": 100},
            "IGNORE_EXCEPTIONS": True,
        },
        "KEY_PREFIX": "djangoadmin",
    }
}

获取 Redis 连接：

from django_redis import get_redis_connection

redis = get_redis_connection("default")

权限查询

get_user_permissions(user_id)

获取用户的权限列表，优先从缓存读取：

def get_user_permissions(user_id):
    """获取用户权限列表，带缓存"""
    redis_conn = get_redis_connection("default")
    ver = redis_conn.get(_VERSION_KEY) or 0
    cache_key = f'{_PERM_PREFIX}{user_id}:{ver}'

    # 尝试从缓存获取
    cached = redis_conn.get(cache_key)
    if cached:
        return json.loads(cached)

    # 缓存未命中，查询数据库
    permissions = _query_permissions_from_db(user_id)

    # 写入缓存（TTL 2小时）
    if permissions:
        redis_conn.set(cache_key, json.dumps(permissions), ex=7200)

    return permissions

缓存失效

菜单变更

菜单增删改时调用，递增全局版本号使所有用户缓存失效：

def invalidate_menu_change():
    """菜单变更时递增全局版本号"""
    redis_conn = get_redis_connection("default")
    redis_conn.incr(_VERSION_KEY)

递增版本号后，旧版本的缓存 Key 不再被访问，自然过期（TTL 2小时）。

用户角色变更

用户角色变更时调用，删除该用户的权限和菜单缓存：

def invalidate_user_perms(user_id):
    """用户角色变更时删除缓存"""
    redis_conn = get_redis_connection("default")
    ver = redis_conn.get(_VERSION_KEY) or 0
    keys = [
        f'{_PERM_PREFIX}{user_id}:{ver}',
        f'{_MENU_PREFIX}{user_id}:{ver}',
    ]
    redis_conn.delete(*keys)

角色菜单变更

角色菜单变更时调用，清理该角色下所有用户的权限和菜单缓存：

def invalidate_role_perms(role_id):
    """角色菜单变更时清理相关用户缓存"""
    # 查询该角色下所有用户 ID
    from application.user_role.models import UserRole
    user_ids = UserRole.objects.filter(
        role_id=role_id, is_delete=False
    ).values_list('user_id', flat=True)

    # 批量删除缓存
    redis_conn = get_redis_connection("default")
    ver = redis_conn.get(_VERSION_KEY) or 0
    keys = []
    for uid in user_ids:
        keys.append(f'{_PERM_PREFIX}{uid}:{ver}')
        keys.append(f'{_MENU_PREFIX}{uid}:{ver}')
    if keys:
        redis_conn.delete(*keys)

PermissionRequired 中间件

权限校验在 middleware/permission_middleware.py 中实现：

class PermissionRequired:
    """权限校验中间件，用户 ID=1 为超级管理员跳过校验"""

    def dispatch(self, request, *args, **kwargs):
        # 超级管理员跳过权限检查
        user_id = get_user_id(request)
        if user_id == 1:
            return super().dispatch(request, *args, **kwargs)

        # 获取用户权限列表
        permissions = get_user_permissions(user_id)

        # 检查是否包含所需权限
        for perm in self.permission_required:
            if perm not in permissions:
                return R.failed(msg="暂无操作权限", code=403)

        return super().dispatch(request, *args, **kwargs)

异常降级

所有缓存操作都包裹 try/except，Redis 不可用时降级直查 DB：

try:
    cached = redis_conn.get(cache_key)
    if cached:
        return json.loads(cached)
except Exception as e:
    logger.warning(f"读取权限缓存异常: {e}")
    # 降级查询数据库

温馨提示

权限缓存的 TTL 为 2小时，正常情况下缓存由业务操作即时失效，TTL 仅作为兜底防止 Redis 故障时缓存永久残留。

总结

权限缓存采用用户级缓存 + 全局版本号双层策略，菜单变更通过递增版本号使所有缓存自然失效，无需遍历删除。所有缓存操作都有异常降级机制，Redis 不可用时自动回源数据库查询。超级管理员（用户 ID=1）跳过权限检查。