Add strict type hinting

.github/workflows/build.yaml

@@ -34,7 +34,7 @@ jobs:
       - name: Run isort
         run: isort --check --profile=black .
       - name: Run mypy
-        run: mypy .
+        run: mypy kq
       - name: Run pytest
         run: py.test --cov=./ --cov-report=xml
       - name: Run Sphinx doctest

.pre-commit-config.yaml

@@ -24,6 +24,7 @@ repos:
     rev: v0.790
     hooks:
       - id: mypy
+        args: [ kq ]
   - repo: https://gitlab.com/pycqa/flake8
     rev: 3.8.4
     hooks:

CONTRIBUTING.md

@@ -10,6 +10,7 @@ pre-commit install         # Install git pre-commit hooks
 Run unit tests with coverage:
 
 ```shell
+docker run -d -p 9092:9092 -e ADV_HOST=127.0.0.1 lensesio/fast-data-dev  # Start Kafka docker.
 py.test --cov=./ --cov-report=html  # Open htmlcov/index.html in your browser
 ```
 

kq/job.py

@@ -1,5 +1,5 @@
 from dataclasses import dataclass
-from typing import Callable, Dict, Optional, Sequence, Union
+from typing import Any, Callable, Dict, Optional, Sequence, Union
 
 
 @dataclass(frozen=True)
@@ -15,13 +15,13 @@ class Job:
     topic: Optional[str] = None
 
     # Function to execute.
-    func: Optional[Callable] = None
+    func: Optional[Callable[..., Any]] = None
 
     # Positional arguments for the function.
-    args: Optional[Sequence] = None
+    args: Optional[Sequence[Any]] = None
 
     # Keyword arguments for the function.
-    kwargs: Optional[Dict] = None
+    kwargs: Optional[Dict[str, Any]] = None
 
     # Job timeout threshold in seconds.
     timeout: Optional[Union[float, int]] = None

kq/queue.py

@@ -35,7 +35,7 @@ def __init__(
         self,
         topic: str,
         producer: KafkaProducer,
-        serializer: Callable,
+        serializer: Callable[[Any], bytes],
         logger: logging.Logger,
         timeout: Union[float, int],
         key: Optional[bytes],
@@ -53,7 +53,9 @@ def __init__(
         self._key = key
         self._partition = partition
 
-    def enqueue(self, obj: Union[Callable, Job], *args: Any, **kwargs: Any) -> Job:
+    def enqueue(
+        self, obj: Union[Callable[..., Any], Job], *args: Any, **kwargs: Any
+    ) -> Job:
         """Enqueue a function call or :doc:`job` instance.
 
         :param obj: Function or :doc:`job <job>`. Must be serializable and
@@ -175,8 +177,8 @@ def __init__(
         self,
         topic: str,
         producer: KafkaProducer,
-        serializer: Optional[Callable] = None,
-        timeout=0,
+        serializer: Optional[Callable[..., bytes]] = None,
+        timeout: int = 0,
         logger: Optional[logging.Logger] = None,
     ) -> None:
         assert is_str(topic), "topic must be a str"
@@ -187,7 +189,7 @@ def __init__(
         assert is_none_or_logger(logger), "bad logger instance"
 
         self._topic = topic
-        self._hosts = producer.config["bootstrap_servers"]
+        self._hosts: str = producer.config["bootstrap_servers"]
         self._producer = producer
         self._serializer = serializer or dill.dumps
         self._timeout = timeout
@@ -245,7 +247,7 @@ def producer(self) -> KafkaProducer:
         return self._producer
 
     @property
-    def serializer(self) -> Callable:
+    def serializer(self) -> Callable[..., bytes]:
         """Return the serializer function.
 
         :return: Serializer function.
@@ -262,7 +264,7 @@ def timeout(self) -> Union[float, int]:
         """
         return self._timeout
 
-    def enqueue(self, func: Callable, *args: Any, **kwargs: Any) -> Job:
+    def enqueue(self, func: Callable[..., bytes], *args: Any, **kwargs: Any) -> Job:
         """Enqueue a function call or a :doc:`job <job>`.
 
         :param func: Function or a :doc:`job <job>` object. Must be

kq/utils.py

@@ -1,9 +1,9 @@
 import logging
 from inspect import isbuiltin, isclass, isfunction, ismethod
-from typing import Any
+from typing import Any, Callable
 
 
-def get_call_repr(func: Any, *args: Any, **kwargs: Any) -> str:
+def get_call_repr(func: Callable[..., Any], *args: Any, **kwargs: Any) -> str:
     """Return the string representation of the function call.
 
     :param func: A callable (e.g. function, method).

kq/worker.py

@@ -7,7 +7,7 @@
 import dill
 from kafka import KafkaConsumer
 
-from kq import Job
+from kq.job import Job
 from kq.message import Message
 from kq.utils import get_call_repr, is_none_or_func, is_none_or_logger, is_str
 
@@ -62,8 +62,8 @@ def __init__(
         self,
         topic: str,
         consumer: KafkaConsumer,
-        callback: Optional[Callable] = None,
-        deserializer: Optional[Callable] = None,
+        callback: Optional[Callable[..., Any]] = None,
+        deserializer: Optional[Callable[[bytes], Any]] = None,
         logger: Optional[logging.Logger] = None,
     ):
         assert is_str(topic), "topic must be a str"
@@ -74,8 +74,8 @@ def __init__(
         assert is_none_or_logger(logger), "bad logger instance"
 
         self._topic = topic
-        self._hosts = consumer.config["bootstrap_servers"]
-        self._group = consumer.config["group_id"]
+        self._hosts: str = consumer.config["bootstrap_servers"]
+        self._group: str = consumer.config["group_id"]
         self._consumer = consumer
         self._callback = callback
         self._deserializer = deserializer or dill.loads
@@ -212,7 +212,7 @@ def consumer(self) -> KafkaConsumer:
         return self._consumer
 
     @property
-    def deserializer(self) -> Callable:
+    def deserializer(self) -> Callable[[bytes], Any]:
         """Return the deserializer function.
 
         :return: Deserializer function.
@@ -221,7 +221,7 @@ def deserializer(self) -> Callable:
         return self._deserializer
 
     @property
-    def callback(self) -> Optional[Callable]:
+    def callback(self) -> Optional[Callable[..., Any]]:
         """Return the callback function.
 
         :return: Callback function, or None if not set.

setup.cfg

@@ -6,3 +6,4 @@ per-file-ignores = __init__.py:F401 conf.py:E402
 
 [mypy]
 ignore_missing_imports = True
+strict = True