Typing fixes for the Cache controller

music_assistant/controllers/cache.py

@@ -11,7 +11,7 @@
 from collections.abc import AsyncGenerator, Awaitable, Callable, Coroutine, Iterator, MutableMapping
 from contextlib import asynccontextmanager
 from contextvars import ContextVar
-from typing import TYPE_CHECKING, Any, Concatenate, ParamSpec, TypeVar, get_type_hints
+from typing import TYPE_CHECKING, Any, Concatenate, ParamSpec, TypeVar, cast, get_type_hints
 
 from music_assistant_models.config_entries import ConfigEntry, ConfigValueType
 from music_assistant_models.enums import ConfigEntryType
@@ -41,7 +41,7 @@ class CacheController(CoreController):
 
     domain: str = "cache"
 
-    def __init__(self, *args, **kwargs) -> None:
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
         """Initialize core controller."""
         super().__init__(*args, **kwargs)
         self.database: DatabaseConnection | None = None
@@ -105,12 +105,11 @@ async def get(
                     cache object matches the checksum provided
         - default: value to return if no cache object is found
         """
+        assert self.database is not None
         assert key, "No key provided"
         if allow_bypass and BYPASS_CACHE.get():
             return default
         cur_time = int(time.time())
-        if checksum is not None and not isinstance(checksum, str):
-            checksum = str(checksum)
         # try memory cache first
         memory_key = f"{provider}/{category}/{key}"
         cache_data = self._mem_cache.get(memory_key)
@@ -162,10 +161,9 @@ async def set(
         - checksum: optional argument to store with the cache object
         - persistent: if True the cache object will not be deleted when clearing the cache
         """
+        assert self.database is not None
         if not key:
             return
-        if checksum is not None and not isinstance(checksum, str):
-            checksum = str(checksum)
         expires = int(time.time() + expiration)
         memory_key = f"{provider}/{category}/{key}"
         self._mem_cache[memory_key] = (data, checksum, expires)
@@ -190,6 +188,7 @@ async def delete(
         self, key: str | None, category: int | None = None, provider: str | None = None
     ) -> None:
         """Delete data from cache."""
+        assert self.database is not None
         match: dict[str, str | int] = {}
         if key is not None:
             match["key"] = key
@@ -211,6 +210,7 @@ async def clear(
         include_persistent: bool = False,
     ) -> None:
         """Clear all/partial items from cache."""
+        assert self.database is not None
         self._mem_cache.clear()
         self.logger.info("Clearing database...")
         query_parts: list[str] = []
@@ -228,6 +228,7 @@ async def clear(
 
     async def auto_cleanup(self) -> None:
         """Run scheduled auto cleanup task."""
+        assert self.database is not None
         self.logger.debug("Running automatic cleanup...")
         # simply reset the memory cache
         self._mem_cache.clear()
@@ -297,6 +298,7 @@ async def _setup_database(self) -> None:
 
     async def __create_database_tables(self) -> None:
         """Create database table(s)."""
+        assert self.database is not None
         await self.database.execute(
             f"""CREATE TABLE IF NOT EXISTS {DB_TABLE_SETTINGS}(
                     key TEXT PRIMARY KEY,
@@ -322,6 +324,7 @@ async def __create_database_tables(self) -> None:
 
     async def __create_database_indexes(self) -> None:
         """Create database indexes."""
+        assert self.database is not None
         await self.database.execute(
             f"CREATE INDEX IF NOT EXISTS {DB_TABLE_CACHE}_category_idx "
             f"ON {DB_TABLE_CACHE}(category);"
@@ -402,7 +405,7 @@ async def wrapper(self: ProviderT, *args: P.args, **kwargs: P.kwargs) -> R:
             )
             if cachedata is not None:
                 type_hints = get_type_hints(func)
-                return parse_value(func.__name__, cachedata, type_hints["return"])
+                return cast("R", parse_value(func.__name__, cachedata, type_hints["return"]))
             # get data from method/provider
             result = await func(self, *args, **kwargs)
             # store result in cache (but don't await)
@@ -424,13 +427,13 @@ async def wrapper(self: ProviderT, *args: P.args, **kwargs: P.kwargs) -> R:
     return _decorator
 
 
-class MemoryCache(MutableMapping):
+class MemoryCache(MutableMapping[str, Any]):
     """Simple limited in-memory cache implementation."""
 
     def __init__(self, maxlen: int) -> None:
         """Initialize."""
         self._maxlen = maxlen
-        self.d = OrderedDict()
+        self.d: OrderedDict[str, Any] = OrderedDict()
 
     @property
     def maxlen(self) -> int:
@@ -458,11 +461,11 @@ def __setitem__(self, key: str, value: Any) -> None:
             self.d.popitem(last=False)
         self.d[key] = value
 
-    def __delitem__(self, key) -> None:
+    def __delitem__(self, key: str) -> None:
         """Delete item."""
         del self.d[key]
 
-    def __iter__(self) -> Iterator:
+    def __iter__(self) -> Iterator[str]:
         """Iterate items."""
         return self.d.__iter__()
 

pyproject.toml

@@ -130,7 +130,6 @@ enable_error_code = [
   "truthy-iterable",
 ]
 exclude = [
-  '^music_assistant/controllers/cache.py$',
   '^music_assistant/controllers/config.py$',
   '^music_assistant/controllers/media/.*$',
   '^music_assistant/controllers/music.py$',