# 权限缓存 :::tip 说明 通过 Redis 实现用户权限和菜单的缓存,采用用户级缓存 + 全局版本号双层策略。权限查询函数位于 `application/menu/services.py`,供 `PermissionRequired` 中间件调用。 ::: ## 缓存策略 采用「用户级缓存 + 全局版本号」双层策略: ``` 1. 每个用户的缓存 Key 带版本号后缀 2. 版本号存全局 Key(perm:version) 3. 菜单变更时递增全局版本号,所有用户缓存自然失效 4. 用户角色变更时,直接删除该用户缓存 5. 角色菜单变更时,查询该角色下所有用户 ID,批量删除缓存 ``` ## 缓存 Key 结构 ```python _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` 管理: ```python # 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 连接: ```python from django_redis import get_redis_connection redis = get_redis_connection("default") ``` ## 权限查询 ### get_user_permissions(user_id) 获取用户的权限列表,优先从缓存读取: ```python 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 ``` ## 缓存失效 ### 菜单变更 菜单增删改时调用,递增全局版本号使所有用户缓存失效: ```python def invalidate_menu_change(): """菜单变更时递增全局版本号""" redis_conn = get_redis_connection("default") redis_conn.incr(_VERSION_KEY) ``` 递增版本号后,旧版本的缓存 Key 不再被访问,自然过期(TTL 2小时)。 ### 用户角色变更 用户角色变更时调用,删除该用户的权限和菜单缓存: ```python 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) ``` ### 角色菜单变更 角色菜单变更时调用,清理该角色下所有用户的权限和菜单缓存: ```python 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` 中实现: ```python 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: ```python try: cached = redis_conn.get(cache_key) if cached: return json.loads(cached) except Exception as e: logger.warning(f"读取权限缓存异常: {e}") # 降级查询数据库 ``` :::tip 温馨提示 权限缓存的 TTL 为 2小时,正常情况下缓存由业务操作即时失效,TTL 仅作为兜底防止 Redis 故障时缓存永久残留。 ::: ## 总结 权限缓存采用用户级缓存 + 全局版本号双层策略,菜单变更通过递增版本号使所有缓存自然失效,无需遍历删除。所有缓存操作都有异常降级机制,Redis 不可用时自动回源数据库查询。超级管理员(用户 ID=1)跳过权限检查。