Refactor cache key and prefix APIs (#1210)

Author: wu-clan

backend/app/admin/api/v1/monitor/online.py

@@ -18,7 +18,7 @@
 async def get_sessions(
     username: Annotated[str | None, Query(description='用户名')] = None,
 ) -> ResponseSchemaModel[list[GetTokenDetail]]:
-    token_keys = await redis_client.get_prefix(f'{settings.TOKEN_REDIS_PREFIX}:*')
+    token_keys = await redis_client.get_by_prefix(settings.TOKEN_REDIS_PREFIX)
     online_clients = await redis_client.smembers(settings.TOKEN_ONLINE_REDIS_PREFIX)
     data: list[GetTokenDetail] = []
     if not token_keys:

backend/app/admin/service/auth_service.py

@@ -218,7 +218,7 @@ async def refresh_token(*, db: AsyncSession, request: Request, response: Respons
             raise errors.NotFoundError(msg='用户不存在')
         if not user.status:
             raise errors.AuthorizationError(msg='用户已被锁定, 请联系统管理员')
-        token_keys = await redis_client.get_prefix(f'{settings.TOKEN_REDIS_PREFIX}:{user.id}:*')
+        token_keys = await redis_client.get_by_prefix(f'{settings.TOKEN_REDIS_PREFIX}:{user.id}')
         if not user.is_multi_login and [
             key for key in token_keys if not key.endswith(f':{token_payload.session_uuid}')
         ]:

backend/app/admin/service/plugin_service.py

@@ -27,7 +27,7 @@ async def get_all() -> list[dict[str, Any]]:
         """获取所有插件"""
 
         changed_key = f'{settings.PLUGIN_REDIS_PREFIX}:changed'
-        keys = [key for key in await redis_client.get_prefix(f'{settings.PLUGIN_REDIS_PREFIX}:') if key != changed_key]
+        keys = [key for key in await redis_client.get_by_prefix(settings.PLUGIN_REDIS_PREFIX) if key != changed_key]
         if not keys:
             return []
 

backend/app/admin/service/user_service.py

@@ -178,15 +178,15 @@ async def update_permission(*, db: AsyncSession, request: Request, pk: int, type
                     # 系统管理员修改自身时，除当前 token 外，其他 token 失效
                     if not new_multi_login:
                         key_prefix = f'{settings.TOKEN_REDIS_PREFIX}:{user.id}'
-                        await redis_client.delete_prefix(
+                        await redis_client.delete_by_prefix(
                             key_prefix,
-                            exclude=f'{key_prefix}:{token_payload.session_uuid}',
+                            exclude_keys=f'{key_prefix}:{token_payload.session_uuid}',
                         )
                 else:
                     # 系统管理员修改他人时，他人 token 全部失效
                     if not new_multi_login:
                         key_prefix = f'{settings.TOKEN_REDIS_PREFIX}:{user.id}'
-                        await redis_client.delete_prefix(key_prefix)
+                        await redis_client.delete_by_prefix(key_prefix)
             case _:
                 raise errors.RequestError(msg='权限类型不存在')
 
@@ -213,9 +213,9 @@ async def reset_password(*, db: AsyncSession, pk: int, password: str) -> int:
         history_obj = CreateUserPasswordHistoryParam(user_id=user.id, password=user.password)
         await password_security_service.save_password_history(db, history_obj)
         await user_dao.update_password_changed_time(db, user.id)
-        await redis_client.delete_prefix(f'{settings.TOKEN_REDIS_PREFIX}:{user.id}')
-        await redis_client.delete_prefix(f'{settings.TOKEN_REFRESH_REDIS_PREFIX}:{user.id}')
-        await redis_client.delete_prefix(f'{settings.JWT_USER_REDIS_PREFIX}:{user.id}')
+        await redis_client.delete_by_prefix(f'{settings.TOKEN_REDIS_PREFIX}:{user.id}')
+        await redis_client.delete_by_prefix(f'{settings.TOKEN_REFRESH_REDIS_PREFIX}:{user.id}')
+        await redis_client.delete_by_prefix(f'{settings.JWT_USER_REDIS_PREFIX}:{user.id}')
         return count
 
     @staticmethod
@@ -294,9 +294,9 @@ async def update_password(*, db: AsyncSession, user_id: int, obj: ResetPasswordP
         history_obj = CreateUserPasswordHistoryParam(user_id=user.id, password=user.password)
         await password_security_service.save_password_history(db, history_obj)
         await user_dao.update_password_changed_time(db, user.id)
-        await redis_client.delete_prefix(f'{settings.TOKEN_REDIS_PREFIX}:{user_id}')
-        await redis_client.delete_prefix(f'{settings.TOKEN_REFRESH_REDIS_PREFIX}:{user_id}')
-        await redis_client.delete_prefix(f'{settings.JWT_USER_REDIS_PREFIX}:{user_id}')
+        await redis_client.delete_by_prefix(f'{settings.TOKEN_REDIS_PREFIX}:{user_id}')
+        await redis_client.delete_by_prefix(f'{settings.TOKEN_REFRESH_REDIS_PREFIX}:{user_id}')
+        await redis_client.delete_by_prefix(f'{settings.JWT_USER_REDIS_PREFIX}:{user_id}')
         return count
 
     @staticmethod
@@ -312,9 +312,9 @@ async def delete(*, db: AsyncSession, pk: int) -> int:
         if not user:
             raise errors.NotFoundError(msg='用户不存在')
         count = await user_dao.delete(db, user.id)
-        await redis_client.delete_prefix(f'{settings.TOKEN_REDIS_PREFIX}:{user.id}')
-        await redis_client.delete_prefix(f'{settings.TOKEN_REFRESH_REDIS_PREFIX}:{user.id}')
-        await redis_client.delete_prefix(f'{settings.JWT_USER_REDIS_PREFIX}:{user.id}')
+        await redis_client.delete_by_prefix(f'{settings.TOKEN_REDIS_PREFIX}:{user.id}')
+        await redis_client.delete_by_prefix(f'{settings.TOKEN_REFRESH_REDIS_PREFIX}:{user.id}')
+        await redis_client.delete_by_prefix(f'{settings.JWT_USER_REDIS_PREFIX}:{user.id}')
         return count
 
 

backend/cli.py

@@ -264,7 +264,7 @@ async def init(db: AsyncSession, redis: RedisCli) -> None:
                 settings.TOKEN_REDIS_PREFIX,
                 settings.TOKEN_REFRESH_REDIS_PREFIX,
             ]:
-                await redis.delete_prefix(prefix)
+                await redis.delete_by_prefix(prefix)
 
             console.note('重建数据库表')
             conn = await db.connection()

backend/common/cache/decorator.py

@@ -1,6 +1,7 @@
 import functools
 
-from collections.abc import Callable, Sequence
+from collections.abc import Awaitable, Callable, Sequence
+from inspect import isawaitable
 from typing import Any, ParamSpec, TypeVar
 
 from msgspec import json
@@ -16,22 +17,23 @@
 
 P = ParamSpec('P')
 T = TypeVar('T')
+_MISSING = object()
 
 
-def _build_cache_key(
-    name: str,
+async def _build_cache_key(
+    namespace: str,
     key: str | None,
-    key_builder: Callable[..., str] | None,
+    key_builder: Callable[..., str | Awaitable[str]] | None,
     *args: Any,
     **kwargs: Any,
 ) -> str:
     """构建缓存 Key"""
     if key:
         if '.' in key:
             param, field = key.split('.', 1)
-            value = kwargs.get(param)
-            if value is None:
-                raise errors.ServerError(msg=f'缓存键构建失败，参数 "{param}" 不存在或值为空')
+            value = kwargs.get(param, _MISSING)
+            if value is _MISSING:
+                raise errors.ServerError(msg=f'缓存键构建失败，参数 "{param}" 不存在')
 
             if isinstance(value, list):
                 raise errors.ServerError(msg='缓存键构建失败：不支持从列表中提取字段，请使用 key_builder 处理列表参数')
@@ -43,16 +45,19 @@ def _build_cache_key(
             else:
                 raise errors.ServerError(msg=f'缓存键构建失败，对象中不存在字段 "{field}"')
         else:
-            value = kwargs.get(key)
-            if value is None:
-                raise errors.ServerError(msg=f'缓存键构建失败，参数 "{key}" 不存在或值为空')
+            value = kwargs.get(key, _MISSING)
+            if value is _MISSING:
+                raise errors.ServerError(msg=f'缓存键构建失败，参数 "{key}" 不存在')
 
-        return f'{name}:{value}'
+        return f'{namespace}:{value if value is not None else "none"}'
 
     if key_builder:
-        return f'{name}:{key_builder(*args, **kwargs)}'
+        value = key_builder(*args, **kwargs)
+        if isawaitable(value):
+            value = await value
+        return f'{namespace}:{value}'
 
-    return name
+    return namespace
 
 
 def _serialize_result(result: Any) -> bytes:
@@ -101,15 +106,15 @@ def user_key_builder() -> str:
 
 
 def cached(  # noqa: C901
-    name: str,
+    namespace: str,
     *,
     key: str | None = None,
-    key_builder: Callable[..., str] | None = None,
+    key_builder: Callable[..., str | Awaitable[str]] | None = None,
 ) -> Callable[[Callable[P, T]], Callable[P, T]]:
     """
     缓存装饰器
 
-    :param name: 缓存名称（通常为缓存 Key 前缀）
+    :param namespace: 缓存命名空间（通常为缓存 Key 前缀）
     :param key: 从方法参数中获取指定参数名的值作为缓存 Key，与 key_builder 互斥
     :param key_builder: 自定义 Key 生成函数，与 key 互斥
     :return:
@@ -120,7 +125,7 @@ def cached(  # noqa: C901
     def decorator(func: Callable[P, T]) -> Callable[P, T]:  # noqa: C901
         @functools.wraps(func)
         async def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
-            cache_key = _build_cache_key(name, key, key_builder, *args, **kwargs)
+            cache_key = await _build_cache_key(namespace, key, key_builder, *args, **kwargs)
 
             # L1: 本地缓存
             if settings.CACHE_LOCAL_ENABLED:
@@ -168,16 +173,16 @@ async def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
 
 
 def cache_invalidate(  # noqa: C901
-    name: str,
+    namespace: str,
     *,
     key: str | None = None,
-    key_builder: Callable[..., str] | None = None,
+    key_builder: Callable[..., str | Awaitable[str]] | None = None,
     atomic: bool = True,
 ) -> Callable[[Callable[P, T]], Callable[P, T]]:
     """
     缓存失效装饰器
 
-    :param name: 缓存名称（通常为缓存 Key 前缀）
+    :param namespace: 缓存命名空间（通常为缓存 Key 前缀）
     :param key: 从方法参数中获取指定参数名的值作为缓存 Key，与 key_builder 互斥
     :param key_builder: 自定义 Key 生成函数，与 key 互斥
     :param atomic: 是否保证缓存原子性
@@ -196,25 +201,25 @@ async def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
             invalidate_error = None
 
             try:
-                invalidate_key = _build_cache_key(name, key, key_builder, *args, **kwargs)
+                invalidate_key = await _build_cache_key(namespace, key, key_builder, *args, **kwargs)
 
                 # L1 缓存失效
                 if settings.CACHE_LOCAL_ENABLED:
-                    if invalidate_key == name:
-                        local_cache_manager.delete_prefix(invalidate_key)
+                    if invalidate_key == namespace:
+                        local_cache_manager.delete_by_prefix(invalidate_key)
                     else:
                         local_cache_manager.delete(invalidate_key)
 
                 # 广播失效消息（通知其他节点清除本地缓存）
                 if settings.CACHE_LOCAL_ENABLED:
-                    if invalidate_key == name:
-                        await cache_pubsub_manager.publish_invalidation(invalidate_key, is_delete_prefix=True)
+                    if invalidate_key == namespace:
+                        await cache_pubsub_manager.publish_invalidation(invalidate_key, delete_by_prefix=True)
                     else:
-                        await cache_pubsub_manager.publish_invalidation(invalidate_key, is_delete_prefix=False)
+                        await cache_pubsub_manager.publish_invalidation(invalidate_key, delete_by_prefix=False)
 
                 # L2 缓存失效
-                if invalidate_key == name:
-                    await redis_client.delete_prefix(invalidate_key)
+                if invalidate_key == namespace:
+                    await redis_client.delete_by_prefix(invalidate_key)
                 else:
                     await redis_client.delete(invalidate_key)
 

backend/common/cache/local.py

@@ -36,17 +36,23 @@ def clear(self) -> None:
         """清空缓存"""
         self.hot_cache.clear()
 
-    def delete_prefix(self, prefix: str, exclude: str | list[str] | None = None) -> None:
+    def delete_by_prefix(self, key_prefix: str, exclude_keys: str | list[str] | None = None) -> None:
         """
         删除指定前缀的缓存
 
-        :param prefix: 要删除的键前缀
-        :param exclude: 要排除的键或键列表
+        :param key_prefix: 要删除的键前缀
+        :param exclude_keys: 要排除的键或键列表
         :return:
         """
-        exclude_set = set(exclude) if isinstance(exclude, list) else {exclude} if isinstance(exclude, str) else set()
+        exclude_set = (
+            set(exclude_keys)
+            if isinstance(exclude_keys, list)
+            else {exclude_keys}
+            if isinstance(exclude_keys, str)
+            else set()
+        )
         for key in list(self.hot_cache.keys()):
-            if key.startswith(prefix) and key not in exclude_set:
+            if (key == key_prefix or key.startswith(f'{key_prefix}:')) and key not in exclude_set:
                 try:
                     del self.hot_cache[key]
                 except KeyError:

backend/common/cache/pubsub.py

@@ -13,16 +13,16 @@ class CachePubSubManager:
     _pubsub_task: asyncio.Task | None = None
 
     @staticmethod
-    async def publish_invalidation(key: str, *, is_delete_prefix: bool) -> None:
+    async def publish_invalidation(cache_key: str, *, delete_by_prefix: bool) -> None:
         """
         发布缓存失效通知
 
-        :param key: 缓存键
-        :param is_delete_prefix: 是否删除符合前缀的所有缓存
+        :param cache_key: 缓存键
+        :param delete_by_prefix: 是否删除符合前缀的所有缓存
         :return:
         """
         try:
-            message = json.dumps({'key': key, 'is_delete_prefix': is_delete_prefix})
+            message = json.dumps({'cache_key': cache_key, 'delete_by_prefix': delete_by_prefix})
             await redis_client.publish(settings.CACHE_PUBSUB_CHANNEL, message)
         except Exception as e:
             log.warning(f'[CachePubSub] 发布通知失败: {e}')
@@ -49,11 +49,11 @@ async def subscribe_and_listen() -> None:  # noqa: C901
                     if message['type'] == 'message':
                         try:
                             data = json.loads(message['data'])
-                            key = data['key']
-                            if not data['is_delete_prefix']:
-                                local_cache_manager.delete(key)
+                            cache_key = data['cache_key']
+                            if not data['delete_by_prefix']:
+                                local_cache_manager.delete(cache_key)
                             else:
-                                local_cache_manager.delete_prefix(key)
+                                local_cache_manager.delete_by_prefix(cache_key)
                         except json.JSONDecodeError as e:
                             log.warning(f'[CachePubSub] 消息格式错误 {e}')
                         except Exception as e:

backend/common/security/jwt.py

@@ -84,7 +84,7 @@ async def create_access_token(user_id: int, *, multi_login: bool, **kwargs) -> A
     })
 
     if not multi_login:
-        await redis_client.delete_prefix(f'{settings.TOKEN_REDIS_PREFIX}:{user_id}')
+        await redis_client.delete_by_prefix(f'{settings.TOKEN_REDIS_PREFIX}:{user_id}')
 
     await redis_client.set(
         f'{settings.TOKEN_REDIS_PREFIX}:{user_id}:{session_uuid}',
@@ -120,7 +120,7 @@ async def create_refresh_token(session_uuid: str, user_id: int, *, multi_login:
     })
 
     if not multi_login:
-        await redis_client.delete_prefix(f'{settings.TOKEN_REFRESH_REDIS_PREFIX}:{user_id}')
+        await redis_client.delete_by_prefix(f'{settings.TOKEN_REFRESH_REDIS_PREFIX}:{user_id}')
 
     await redis_client.set(
         f'{settings.TOKEN_REFRESH_REDIS_PREFIX}:{user_id}:{session_uuid}',

backend/database/redis.py

@@ -62,19 +62,33 @@ async def init(self) -> None:
             log.error('Redis 服务器连接异常 {}', e)
             sys.exit()
 
-    async def delete_prefix(self, prefix: str, exclude: str | list[str] | None = None, batch_size: int = 1000) -> None:
+    async def delete_by_prefix(
+        self,
+        key_prefix: str,
+        exclude_keys: str | list[str] | None = None,
+        batch_size: int = 1000,
+    ) -> None:
         """
         删除指定前缀的所有 key
 
-        :param prefix: 要删除的键前缀
-        :param exclude: 要排除的键或键列表
+        :param key_prefix: 要删除的键前缀
+        :param exclude_keys: 要排除的键或键列表
         :param batch_size: 批量删除的大小，避免一次性删除过多键导致 Redis 阻塞
         :return:
         """
-        exclude_set = set(exclude) if isinstance(exclude, list) else {exclude} if isinstance(exclude, str) else set()
+        exclude_set = (
+            set(exclude_keys)
+            if isinstance(exclude_keys, list)
+            else {exclude_keys}
+            if isinstance(exclude_keys, str)
+            else set()
+        )
         batch_keys = []
 
-        async for key in self.scan_iter(match=f'{prefix}*'):
+        if key_prefix not in exclude_set and await self.exists(key_prefix):
+            batch_keys.append(key_prefix)
+
+        async for key in self.scan_iter(match=f'{key_prefix}:*'):
             if key not in exclude_set:
                 batch_keys.append(key)
 
@@ -85,15 +99,15 @@ async def delete_prefix(self, prefix: str, exclude: str | list[str] | None = Non
         if batch_keys:
             await self.delete(*batch_keys)
 
-    async def get_prefix(self, prefix: str, count: int = 100) -> list[str]:
+    async def get_by_prefix(self, key_prefix: str, count: int = 100) -> list[str]:
         """
         获取指定前缀的所有 key
 
-        :param prefix: 要搜索的键前缀
+        :param key_prefix: 要搜索的键前缀
         :param count: 每次扫描批次的数量，值越大扫描速度越快，但会占用更多服务器资源
         :return:
         """
-        return [key async for key in self.scan_iter(match=f'{prefix}*', count=count)]
+        return [key async for key in self.scan_iter(match=f'{key_prefix}:*', count=count)]
 
 
 # 创建 redis 客户端单例