➕ expiringdict替换为expiringdictx

Author: AzideCupric

nonebot_bison/admin_page/token_manager.py

@@ -1,12 +1,13 @@
 import random
 import string
+from datetime import timedelta
 
-from expiringdict import ExpiringDict
+from expiringdictx import ExpiringDict
 
 
 class TokenManager:
     def __init__(self):
-        self.token_manager = ExpiringDict(max_len=100, max_age_seconds=60 * 10)
+        self.token_manager = ExpiringDict[str, tuple](capacity=100, default_age=timedelta(minutes=10))
 
     def get_user(self, token: str) -> tuple | None:
         res = self.token_manager.get(token)

nonebot_bison/platform/ceobecanteen/cache.py

@@ -1,10 +1,9 @@
 from functools import partial
+from datetime import timedelta
 from types import MappingProxyType
 from collections.abc import Callable
-from typing import Any, TypeVar, cast
-from datetime import datetime, timedelta
 
-from expiringdict import ExpiringDict
+from expiringdictx import ExpiringDict
 from hishel import Controller, AsyncCacheClient, AsyncFileStorage
 
 from .const import DATASOURCE_URL
@@ -18,47 +17,12 @@
     storage=AsyncFileStorage(ttl=300),
 )
 
-StoreType = TypeVar("StoreType")
-
-
-class SimpleCache:
-    """简单缓存"""
-
-    def __init__(self, default_lifetime: timedelta = timedelta(minutes=5)):
-        self.default_lifetime = default_lifetime
-        self._cache: dict[str, tuple[Any, datetime, timedelta]] = {}
-
-    def __setitem__(self, key: str, value: tuple[Any, timedelta | None]):
-        value, lifetime = value
-        self._cache[key] = (value, datetime.now(), lifetime or self.default_lifetime)
-
-    def __getitem__(self, key: str):
-        if item := self._cache.get(key):
-            value, create_time, lifetime = item
-            if datetime.now() - create_time <= lifetime:
-                return value
-            else:
-                del self._cache[key]
-        return None
-
-    def set(self, key: str, value: Any, lifetime: timedelta | None = None):
-        self[key] = (value, lifetime)
-
-    def get(
-        self, key: str, default: Any = None, value_convert_func: Callable[[Any], StoreType | None] | None = None
-    ) -> StoreType | None:
-        if value := self[key]:
-            if value_convert_func:
-                return value_convert_func(value)
-            return value
-        return default
-
 
 class CeobeDataSourceCache:
     """数据源缓存"""
 
     def __init__(self):
-        self._cache = ExpiringDict(max_len=100, max_age_seconds=60 * 60 * 24 * 7)
+        self._cache = ExpiringDict[str, CeobeTarget](capacity=100, default_age=timedelta(days=7))
         self.client = CeobeClient()
         self.url = DATASOURCE_URL
         self.init_requested = False
@@ -86,7 +50,7 @@ def select_one(self, cond_func: Callable[[CeobeTarget], bool]) -> CeobeTarget |
 
         不会刷新缓存
         """
-        cache = cast(list[CeobeTarget], self._cache.values())
+        cache = self._cache.values()
         return next(filter(cond_func, cache), None)
 
     async def get_by_unique_id(self, unique_id: str) -> CeobeTarget | None:
@@ -95,9 +59,9 @@ async def get_by_unique_id(self, unique_id: str) -> CeobeTarget | None:
         如果在缓存中找不到，会刷新缓存
         """
         if target := self._cache.get(unique_id):
-            return cast(CeobeTarget, target)
+            return target
         await self.refresh_data_sources()
-        return cast(CeobeTarget | None, self._cache.get(unique_id))
+        return self._cache.get(unique_id)
 
     async def get_by_nickname(self, nickname: str) -> CeobeTarget | None:
         """根据nickname获取数据源

nonebot_bison/platform/ceobecanteen/platform.py

@@ -3,14 +3,15 @@
 
 from nonebot import logger
 from hishel import AsyncCacheClient
+from expiringdictx import SimpleCache
 
 from ...post import Post
 from ..platform import NewMessage
 from .utils import process_response
 from ...utils import SchedulerConfig
 from ...types import Target, RawPost, Category
+from .cache import CeobeClient, CeobeDataSourceCache
 from .const import COMB_ID_URL, COOKIES_URL, COOKIE_ID_URL
-from .cache import CeobeClient, SimpleCache, CeobeDataSourceCache
 from .models import CeobeCookie, CombIdResponse, CookiesResponse, CookieIdResponse
 
 
@@ -57,7 +58,7 @@ async def get_comb_id(self, targets: list[Target] | None = None, force_refresh:
             )
             comb_id = process_response(resp, CombIdResponse).data["datasource_comb_id"]
             logger.trace(f"get comb_id: {comb_id}")
-            self.cache_store["comb_id"] = (comb_id, timedelta(hours=12))
+            self.cache_store["comb_id", timedelta(hours=12)] = comb_id
 
         return self.cache_store["comb_id"]
 
@@ -83,8 +84,8 @@ async def request():
 
         def update(cookie_id: str, cookies: list[CeobeCookie]):
             lifetime = timedelta(hours=1)
-            self.cache_store["cookie_id"] = (cookie_id, lifetime)
-            self.cache_store["cookies"] = (cookies, lifetime)
+            self.cache_store["cookie_id", lifetime] = cookie_id
+            self.cache_store["cookies", lifetime] = cookies
 
         if cookie_id != self.cache_store["cookie_id"] or force_refresh:
             logger.trace(f"cookie_id changed: {self.cache_store['cookie_id']} -> {cookie_id}, request cookies")