权限缓存

说明

通过 Redis 实现用户权限和菜单的缓存，采用用户级缓存 + 全局版本号双层策略。源码位于 src/utils/perm_cache.py，供 src/modules/system/menu/service.py 中的权限查询函数调用。

缓存策略

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

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

缓存 Key 结构

_VERSION_KEY = 'perm:version' # 全局版本号
_PERM_PREFIX = 'perm:list:' # 用户权限列表
_MENU_PREFIX = 'perm:menu:' # 用户菜单树
_USER_ACTIVE_PREFIX = 'user:active:' # 用户有效状态

Key 格式示例：

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

权限缓存读写

get_cached_permissions()

async def get_cached_permissions(user_id: int):
    """获取缓存的用户权限列表，未命中返回 None"""
    redis = get_redis()
    ver = await _get_version()
    key = f'{_PERM_PREFIX}{user_id}:{ver}'
    data = await redis.get(key)
    if data is not None:
        return json.loads(data)
    return None

set_cached_permissions()

async def set_cached_permissions(user_id: int, perm_list: list):
    """缓存用户权限列表（空列表不缓存，避免缓存空壳后无法感知后续数据变更）"""
    if not perm_list:
        return
    redis = get_redis()
    ver = await _get_version()
    key = f'{_PERM_PREFIX}{user_id}:{ver}'
    await redis.set(key, json.dumps(perm_list), ex=_CACHE_TTL)

缓存失效

菜单变更

async def invalidate_menu_change():
    """菜单增删改时调用，递增全局版本号使所有用户缓存失效"""
    await _incr_version()

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

用户角色变更

async def invalidate_user_perms(user_id: int):
    """用户角色变更时调用，删除该用户的权限和菜单缓存"""
    redis = get_redis()
    ver = await _get_version()
    keys = [
        f'{_PERM_PREFIX}{user_id}:{ver}',
        f'{_MENU_PREFIX}{user_id}:{ver}',
    ]
    await redis.delete(*keys)

角色菜单变更

async def invalidate_role_perms(role_id: int):
    """角色菜单变更时调用，清理该角色下所有用户的权限和菜单缓存"""
    # 查询该角色下所有用户 ID
    user_ids = await asyncio.to_thread(_query_user_ids)

    # 批量删除缓存
    redis = get_redis()
    ver = await _get_version()
    keys = []
    for uid in user_ids:
        keys.append(f'{_PERM_PREFIX}{uid}:{ver}')
        keys.append(f'{_MENU_PREFIX}{uid}:{ver}')
    await redis.delete(*keys)

用户有效状态缓存

用于 login_required 每请求校验，避免每次请求都查库：

_USER_ACTIVE_TTL = 300 # 5分钟

async def get_cached_user_active(user_id: int):
 """读取用户有效状态缓存；未命中返回 None（调用方需回源 DB 校验）"""

async def set_cached_user_active(user_id: int, active: bool):
 """写入用户有效状态缓存"""

async def invalidate_user_active(user_id: int):
 """用户状态/删除变更时调用，删除有效状态缓存"""

异常降级

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

except Exception as e:
    logger.warning(f"读取权限缓存异常: {e}")
    return None  # 调用方回源 DB

删除失败时递增全局版本号孤立旧缓存（fail-closed）：

except Exception as e:
    logger.error(f"清理用户权限缓存异常，降级递增全局版本号: {e}")
    try:
        await _incr_version()
    except Exception as e2:
        logger.error(f"降级递增版本号失败: {e2}")

温馨提示

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

总结

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