Add mypy to dircache

Author: alanhdu

fsspec/dircache.py

@@ -1,9 +1,17 @@
+from __future__ import annotations
 import time
-from collections.abc import MutableMapping
+from typing import MutableMapping, List, TypedDict, Iterator, Any
+
 from functools import lru_cache
 
 
-class DirCache(MutableMapping):
+class DirEntry(TypedDict):
+    name: str
+    size: int
+    type: str
+
+
+class DirCache(MutableMapping[str, List[DirEntry]]):
     """
     Caching of directory listings, in a structure like::
 
@@ -26,10 +34,10 @@ class DirCache(MutableMapping):
 
     def __init__(
         self,
-        use_listings_cache=True,
-        listings_expiry_time=None,
-        max_paths=None,
-        **kwargs,
+        use_listings_cache: bool = True,
+        listings_expiry_time: float | None = None,
+        max_paths: int | None = None,
+        **kwargs: Any,
     ):
         """
 
@@ -45,36 +53,36 @@ def __init__(
             The number of most recent listings that are considered valid; 'recent'
             refers to when the entry was set.
         """
-        self._cache = {}
-        self._times = {}
+        self._cache: dict[str, list[DirEntry]] = {}
+        self._times: dict[str, float] = {}
         if max_paths:
             self._q = lru_cache(max_paths + 1)(lambda key: self._cache.pop(key, None))
         self.use_listings_cache = use_listings_cache
         self.listings_expiry_time = listings_expiry_time
         self.max_paths = max_paths
 
-    def __getitem__(self, item):
+    def __getitem__(self, item: str) -> list[DirEntry]:
         if self.listings_expiry_time is not None:
             if self._times.get(item, 0) - time.time() < -self.listings_expiry_time:
                 del self._cache[item]
         if self.max_paths:
             self._q(item)
         return self._cache[item]  # maybe raises KeyError
 
-    def clear(self):
+    def clear(self) -> None:
         self._cache.clear()
 
-    def __len__(self):
+    def __len__(self) -> int:
         return len(self._cache)
 
-    def __contains__(self, item):
+    def __contains__(self, item: object) -> bool:
         try:
-            self[item]
+            self[item]  # type: ignore[index]
             return True
         except KeyError:
             return False
 
-    def __setitem__(self, key, value):
+    def __setitem__(self, key: str, value: List[DirEntry]) -> None:
         if not self.use_listings_cache:
             return
         if self.max_paths:
@@ -83,10 +91,10 @@ def __setitem__(self, key, value):
         if self.listings_expiry_time is not None:
             self._times[key] = time.time()
 
-    def __delitem__(self, key):
+    def __delitem__(self, key) -> None:
         del self._cache[key]
 
-    def __iter__(self):
+    def __iter__(self) -> Iterator[str]:
         entries = list(self._cache)
 
         return (k for k in entries if k in self)

setup.cfg

@@ -57,6 +57,9 @@ exclude = (test.*|conftest)\.py$
 [mypy-fsspec.caching]
 check_untyped_defs = True
 
+[mypy-fsspec.dircache]
+check_untyped_defs = True
+
 [mypy-fsspec.mapping]
 check_untyped_defs = True