Add typing to Celery

This is a simple bootstrap of the process, adding some types to a
few selected functions, based on comment annotations. MyPy is chosen as the
default static analyzer for the types.

.github/workflows/lint_python.yml

@@ -9,9 +9,22 @@ jobs:
       - uses: pre-commit/action@v2.0.3
       - run: pip install --upgrade pip wheel
       - run: pip install bandit codespell flake8 isort pytest pyupgrade tox
-      - run: bandit -r . || true
-      - run: codespell --ignore-words-list="brane,gool,ist,sherif,wil" --quiet-level=2 --skip="*.key" || true
+
+      - name: bandit
+        run: bandit -r . || true
+
+      - name: Run CodeSpell
+        run: codespell --ignore-words-list="brane,gool,ist,sherif,wil" --quiet-level=2 --skip="*.key" || true
       - run: pip install -r requirements.txt || true
-      - run: tox || true
-      - run: pytest . || true
-      - run: pytest --doctest-modules . || true
+
+      - name: Run tox
+        run: tox || true
+
+      - name: Run pytest
+        run: pytest . || true
+
+      - name: Test pytest with doctest
+        run: pytest --doctest-modules . || true
+
+      - name: MyPy
+        run: tox -e mypy

celery/contrib/testing/worker.py

@@ -90,16 +90,15 @@ def start_worker(
 
 
 @contextmanager
-def _start_worker_thread(app,
-                         concurrency=1,
-                         pool='solo',
+def _start_worker_thread(app: Celery,
+                         concurrency: int = 1,
+                         pool: str = 'solo',
                          loglevel=WORKER_LOGLEVEL,
-                         logfile=None,
-                         WorkController=TestWorkController,
-                         perform_ping_check=True,
-                         shutdown_timeout=10.0,
-                         **kwargs):
-    # type: (Celery, int, str, Union[str, int], str, Any, **Any) -> Iterable
+                         logfile: Union[str, int] = None,
+                         WorkController: Any = TestWorkController,
+                         perform_ping_check: bool = True,
+                         shutdown_timeout: float = 10.0,
+                         **kwargs: Any) -> Iterable:
     """Start Celery worker in a thread.
 
     Yields:
@@ -167,8 +166,7 @@ def _start_worker_process(app,
     cluster.stopwait()
 
 
-def setup_app_for_worker(app, loglevel, logfile):
-    # type: (Celery, Union[str, int], str) -> None
+def setup_app_for_worker(app: Celery, loglevel: Union[str, int], logfile: str) -> None:
     """Setup the app to be used for starting an embedded worker."""
     app.finalize()
     app.set_current()

celery/events/state.py

@@ -22,7 +22,7 @@
 from itertools import islice
 from operator import itemgetter
 from time import time
-from typing import Mapping
+from typing import Mapping, Optional
 from weakref import WeakSet, ref
 
 from kombu.clocks import timetuple
@@ -457,7 +457,7 @@ def clear_tasks(self, ready=True):
         with self._mutex:
             return self._clear_tasks(ready)
 
-    def _clear_tasks(self, ready=True):
+    def _clear_tasks(self, ready: bool = True):
         if ready:
             in_progress = {
                 uuid: task for uuid, task in self.itertasks()
@@ -475,7 +475,7 @@ def _clear(self, ready=True):
         self.event_count = 0
         self.task_count = 0
 
-    def clear(self, ready=True):
+    def clear(self, ready: bool = True):
         with self._mutex:
             return self._clear(ready)
 
@@ -495,7 +495,7 @@ def get_or_create_worker(self, hostname, **kwargs):
                 hostname, **kwargs)
             return worker, True
 
-    def get_or_create_task(self, uuid):
+    def get_or_create_task(self, uuid: str):
         """Get or create task by uuid."""
         try:
             return self.tasks[uuid], False
@@ -652,13 +652,13 @@ def rebuild_taskheap(self, timetuple=timetuple):
         ]
         heap.sort()
 
-    def itertasks(self, limit=None):
+    def itertasks(self, limit: Optional[int] = None):
         for index, row in enumerate(self.tasks.items()):
             yield row
             if limit and index + 1 >= limit:
                 break
 
-    def tasks_by_time(self, limit=None, reverse=True):
+    def tasks_by_time(self, limit=None, reverse: bool = True):
         """Generator yielding tasks ordered by time.
 
         Yields:

celery/utils/collections.py

@@ -113,8 +113,7 @@ def __getattr__(self, k):
             raise AttributeError(
                 f'{type(self).__name__!r} object has no attribute {k!r}')
 
-    def __setattr__(self, key, value):
-        # type: (str, Any) -> None
+    def __setattr__(self, key: str, value: Any) -> None:
         """`d[key] = value -> d.key = value`."""
         self[key] = value
 
@@ -595,8 +594,7 @@ def purge(self, now=None):
                     break  # oldest item hasn't expired yet
                 self.pop()
 
-    def pop(self, default=None):
-        # type: (Any) -> Any
+    def pop(self, default: Any = None) -> Any:
         """Remove and return the oldest item, or :const:`None` when empty."""
         while self._heap:
             _, item = heappop(self._heap)
@@ -608,8 +606,7 @@ def pop(self, default=None):
                 return item
         return default
 
-    def as_dict(self):
-        # type: () -> Dict
+    def as_dict(self) -> Dict:
         """Whole set as serializable dictionary.
 
         Example:
@@ -675,20 +672,17 @@ class Evictable:
 
     Empty = Empty
 
-    def evict(self):
-        # type: () -> None
+    def evict(self) -> None:
         """Force evict until maxsize is enforced."""
         self._evict(range=count)
 
-    def _evict(self, limit=100, range=range):
-        # type: (int) -> None
+    def _evict(self, limit: int = 100, range=range) -> None:
         try:
             [self._evict1() for _ in range(limit)]
         except IndexError:
             pass
 
-    def _evict1(self):
-        # type: () -> None
+    def _evict1(self) -> None:
         if self._evictcount <= self.maxsize:
             raise IndexError()
         try:

mypy.ini

@@ -0,0 +1,10 @@
+[mypy]
+warn_unused_configs = True
+follow_imports = skip
+strict = False
+warn_return_any = True
+show_error_codes = True
+# True is the goal for disallow_untyped_defs
+disallow_untyped_defs = False
+ignore_missing_imports = True
+allowlist_externals = True

requirements/test.txt

@@ -4,6 +4,8 @@ pytest-subtests
 pytest-timeout~=1.4.2
 boto3>=1.9.178
 moto>=2.2.6
+# typing extensions
+mypy
 pre-commit
 -r extras/yaml.txt
 -r extras/msgpack.txt

tox.ini

@@ -78,9 +78,12 @@ basepython =
     3.9: python3.9
     3.10: python3.10
     pypy3: pypy3
+    mypy: python3.8
     lint,apicheck,linkcheck,configcheck,bandit: python3.9
 usedevelop = True
 
+[testenv:mypy]
+commands = mypy celery
 
 [testenv:apicheck]
 setenv =