feat: Add type annotations (#279)

* Add type annotations

* add typechecking to tox / github workflow

* code review

Author: dimbleby

.github/workflows/tests.yml

@@ -34,7 +34,10 @@ jobs:
           python -VV
           python -m site
           python -m pip install --upgrade pip setuptools wheel
-          python -m pip install --upgrade virtualenv tox tox-gh-actions          
+          python -m pip install --upgrade virtualenv tox tox-gh-actions
 
       - name: "Run tox targets for ${{ matrix.python-version }}"
         run: "python -m tox"
+
+      - name: "Run mypy for ${{ matrix.python-version }}"
+        run: "python -m tox -e mypy"

cachecontrol/__init__.py

@@ -10,9 +10,19 @@
 __email__ = "eric@ionrock.org"
 __version__ = "0.12.11"
 
-from .wrapper import CacheControl
 from .adapter import CacheControlAdapter
 from .controller import CacheController
+from .wrapper import CacheControl
+
+__all__ = [
+    "__author__",
+    "__email__",
+    "__version__",
+    "CacheControlAdapter",
+    "CacheController",
+    "CacheControl",
+]
 
 import logging
+
 logging.getLogger(__name__).addHandler(logging.NullHandler())

cachecontrol/_cmd.py

@@ -3,41 +3,46 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import logging
+from argparse import ArgumentParser
+from typing import TYPE_CHECKING
 
 import requests
 
 from cachecontrol.adapter import CacheControlAdapter
 from cachecontrol.cache import DictCache
 from cachecontrol.controller import logger
 
-from argparse import ArgumentParser
+if TYPE_CHECKING:
+    from argparse import Namespace
 
+    from .controller import CacheController
 
-def setup_logging():
+
+def setup_logging() -> None:
     logger.setLevel(logging.DEBUG)
     handler = logging.StreamHandler()
     logger.addHandler(handler)
 
 
-def get_session():
+def get_session() -> requests.Session:
     adapter = CacheControlAdapter(
         DictCache(), cache_etags=True, serializer=None, heuristic=None
     )
     sess = requests.Session()
     sess.mount("http://", adapter)
     sess.mount("https://", adapter)
 
-    sess.cache_controller = adapter.controller
+    sess.cache_controller = adapter.controller  # type: ignore[attr-defined]
     return sess
 
 
-def get_args():
+def get_args() -> "Namespace":
     parser = ArgumentParser()
     parser.add_argument("url", help="The URL to try and cache")
     return parser.parse_args()
 
 
-def main(args=None):
+def main() -> None:
     args = get_args()
     sess = get_session()
 
@@ -48,10 +53,13 @@ def main(args=None):
     setup_logging()
 
     # try setting the cache
-    sess.cache_controller.cache_response(resp.request, resp.raw)
+    cache_controller: "CacheController" = (
+        sess.cache_controller  # type: ignore[attr-defined]
+    )
+    cache_controller.cache_response(resp.request, resp.raw)
 
     # Now try to get it
-    if sess.cache_controller.cached_request(resp.request):
+    if cache_controller.cached_request(resp.request):
         print("Cached!")
     else:
         print("Not cached :(")

cachecontrol/adapter.py

@@ -2,31 +2,40 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
-import types
 import functools
+import types
 import zlib
+from typing import TYPE_CHECKING, Any, Collection, Mapping, Optional, Tuple, Type, Union
 
 from requests.adapters import HTTPAdapter
 
-from .controller import CacheController, PERMANENT_REDIRECT_STATUSES
 from .cache import DictCache
+from .controller import PERMANENT_REDIRECT_STATUSES, CacheController
 from .filewrapper import CallbackFileWrapper
 
+if TYPE_CHECKING:
+    from requests import PreparedRequest, Response
+
+    from .cache import BaseCache
+    from .compat import HTTPResponse
+    from .heuristics import BaseHeuristic
+    from .serialize import Serializer
+
 
 class CacheControlAdapter(HTTPAdapter):
     invalidating_methods = {"PUT", "PATCH", "DELETE"}
 
     def __init__(
         self,
-        cache=None,
-        cache_etags=True,
-        controller_class=None,
-        serializer=None,
-        heuristic=None,
-        cacheable_methods=None,
-        *args,
-        **kw
-    ):
+        cache: Optional["BaseCache"] = None,
+        cache_etags: bool = True,
+        controller_class: Optional[Type[CacheController]] = None,
+        serializer: Optional["Serializer"] = None,
+        heuristic: Optional["BaseHeuristic"] = None,
+        cacheable_methods: Optional[Collection[str]] = None,
+        *args: Any,
+        **kw: Any,
+    ) -> None:
         super(CacheControlAdapter, self).__init__(*args, **kw)
         self.cache = DictCache() if cache is None else cache
         self.heuristic = heuristic
@@ -37,7 +46,18 @@ def __init__(
             self.cache, cache_etags=cache_etags, serializer=serializer
         )
 
-    def send(self, request, cacheable_methods=None, **kw):
+    def send(
+        self,
+        request: "PreparedRequest",
+        stream: bool = False,
+        timeout: Union[None, float, Tuple[float, float], Tuple[float, None]] = None,
+        verify: Union[bool, str] = True,
+        cert: Union[
+            None, bytes, str, Tuple[Union[bytes, str], Union[bytes, str]]
+        ] = None,
+        proxies: Optional[Mapping[str, str]] = None,
+        cacheable_methods: Optional[Collection[str]] = None,
+    ) -> "Response":
         """
         Send a request. Use the request information to see if it
         exists in the cache and cache the response if we need to and can.
@@ -54,13 +74,19 @@ def send(self, request, cacheable_methods=None, **kw):
             # check for etags and add headers if appropriate
             request.headers.update(self.controller.conditional_headers(request))
 
-        resp = super(CacheControlAdapter, self).send(request, **kw)
+        resp = super(CacheControlAdapter, self).send(
+            request, stream, timeout, verify, cert, proxies
+        )
 
         return resp
 
     def build_response(
-        self, request, response, from_cache=False, cacheable_methods=None
-    ):
+        self,
+        request: "PreparedRequest",
+        response: "HTTPResponse",
+        from_cache: bool = False,
+        cacheable_methods: Optional[Collection[str]] = None,
+    ) -> "Response":
         """
         Build a response by making a request or using the cache.
 
@@ -111,7 +137,7 @@ def build_response(
                 if response.chunked:
                     super_update_chunk_length = response._update_chunk_length
 
-                    def _update_chunk_length(self):
+                    def _update_chunk_length(self: "HTTPResponse") -> None:
                         super_update_chunk_length()
                         if self.chunk_left == 0:
                             self._fp._close()
@@ -120,18 +146,21 @@ def _update_chunk_length(self):
                         _update_chunk_length, response
                     )
 
-        resp = super(CacheControlAdapter, self).build_response(request, response)
+        resp: "Response" = super(  # type: ignore[no-untyped-call]
+            CacheControlAdapter, self
+        ).build_response(request, response)
 
         # See if we should invalidate the cache.
         if request.method in self.invalidating_methods and resp.ok:
+            assert request.url is not None
             cache_url = self.controller.cache_url(request.url)
             self.cache.delete(cache_url)
 
         # Give the request a from_cache attr to let people use it
-        resp.from_cache = from_cache
+        resp.from_cache = from_cache  # type: ignore[attr-defined]
 
         return resp
 
-    def close(self):
+    def close(self) -> None:
         self.cache.close()
-        super(CacheControlAdapter, self).close()
+        super(CacheControlAdapter, self).close()  # type: ignore[no-untyped-call]

cachecontrol/cache.py

@@ -7,37 +7,43 @@
 safe in-memory dictionary.
 """
 from threading import Lock
+from typing import IO, TYPE_CHECKING, MutableMapping, Optional, Union
 
+if TYPE_CHECKING:
+    from datetime import datetime
 
-class BaseCache(object):
 
-    def get(self, key):
+class BaseCache(object):
+    def get(self, key: str) -> Optional[bytes]:
         raise NotImplementedError()
 
-    def set(self, key, value, expires=None):
+    def set(
+        self, key: str, value: bytes, expires: Optional[Union[int, "datetime"]] = None
+    ) -> None:
         raise NotImplementedError()
 
-    def delete(self, key):
+    def delete(self, key: str) -> None:
         raise NotImplementedError()
 
-    def close(self):
+    def close(self) -> None:
         pass
 
 
 class DictCache(BaseCache):
-
-    def __init__(self, init_dict=None):
+    def __init__(self, init_dict: Optional[MutableMapping[str, bytes]] = None) -> None:
         self.lock = Lock()
         self.data = init_dict or {}
 
-    def get(self, key):
+    def get(self, key: str) -> Optional[bytes]:
         return self.data.get(key, None)
 
-    def set(self, key, value, expires=None):
+    def set(
+        self, key: str, value: bytes, expires: Optional[Union[int, "datetime"]] = None
+    ) -> None:
         with self.lock:
             self.data.update({key: value})
 
-    def delete(self, key):
+    def delete(self, key: str) -> None:
         with self.lock:
             if key in self.data:
                 self.data.pop(key)
@@ -55,10 +61,11 @@ class SeparateBodyBaseCache(BaseCache):
 
     Similarly, the body should be loaded separately via ``get_body()``.
     """
-    def set_body(self, key, body):
+
+    def set_body(self, key: str, body: bytes) -> None:
         raise NotImplementedError()
 
-    def get_body(self, key):
+    def get_body(self, key: str) -> Optional["IO[bytes]"]:
         """
         Return the body as file-like object.
         """