Implement mypy plugin for luigi.Task

.github/workflows/pythonbuild.yml

@@ -205,9 +205,9 @@ jobs:
             tox-env: py312-azureblob
 
 
-          - python-version: 3.9
+          - python-version: "3.10"
             tox-env: flake8
-          - python-version: 3.9
+          - python-version: "3.10"
             tox-env: docs
 
     steps:

doc/conf.py

@@ -338,6 +338,8 @@ def _warn_node(self, msg, node, *args, **kwargs):
 # If true, do not generate a @detailmenu in the "Top" node's menu.
 #texinfo_no_detailmenu = False
 
+autodoc_mock_imports = ["mypy"]
+
 # Some regression introduced
 # https://github.com/sphinx-doc/sphinx/issues/2330
 # https://github.com/spotify/luigi/pull/1555

doc/index.rst

@@ -22,6 +22,7 @@ Table of Contents
    configuration.rst
    logging.rst
    design_and_limitations.rst
+   mypy.rst
 
 API Reference
 -------------

doc/mypy.rst

@@ -0,0 +1,42 @@
+Mypy plugin
+--------------
+
+Mypy plugin provides type checking for ``luigi.Task`` using Mypy.
+
+Require Python 3.8 or later.
+
+How to use
+~~~~~~~~~~
+
+Configure Mypy to use this plugin by adding the following to your ``mypy.ini`` file:
+
+.. code:: ini
+
+    [mypy]
+    plugins = luigi.mypy
+
+or by adding the following to your ``pyproject.toml`` file:
+
+.. code:: toml
+
+    [tool.mypy]
+    plugins = ["luigi.mypy"]
+
+Then, run Mypy as usual.
+
+Examples
+~~~~~~~~
+
+For example the following code linted by Mypy:
+
+.. code:: python
+
+    import luigi
+
+
+    class MyTask(luigi.Task):
+        foo: int = luigi.IntParameter()
+        bar: str = luigi.Parameter()
+
+    MyTask(foo=1, bar='2')   # OK
+    MyTask(foo='1', bar='2')  # Error: Argument 1 to "Foo" has incompatible type "str"; expected "int"

luigi/configuration/cfg_parser.py

@@ -117,7 +117,7 @@ def before_write(self, parser, section, option, value):
 class LuigiConfigParser(BaseParser, ConfigParser):
     NO_DEFAULT = object()
     enabled = True
-    optionxform = str
+    optionxform = str  # type: ignore
     _instance = None
     _config_paths = [
         '/etc/luigi/client.cfg',  # Deprecated old-style global luigi config

luigi/configuration/toml_parser.py

@@ -16,20 +16,22 @@
 #
 import os.path
 from configparser import ConfigParser
+from typing import Any, Dict
 
 try:
     import toml
+    toml_enabled = True
 except ImportError:
-    toml = False
+    toml_enabled = False
 
 from .base_parser import BaseParser
 from ..freezing import recursively_freeze
 
 
 class LuigiTomlParser(BaseParser, ConfigParser):
     NO_DEFAULT = object()
-    enabled = bool(toml)
-    data = dict()
+    enabled = bool(toml_enabled)
+    data: Dict[str, Any] = dict()
     _instance = None
     _config_paths = [
         '/etc/luigi/luigi.toml',

luigi/db_task_history.py

@@ -188,7 +188,7 @@ def find_task_by_id(self, id, session=None):
             return session.query(TaskRecord).get(id)
 
 
-class TaskParameter(Base):
+class TaskParameter(Base):  # type: ignore
     """
     Table to track luigi.Parameter()s of a Task.
     """
@@ -201,7 +201,7 @@ def __repr__(self):
         return "TaskParameter(task_id=%d, name=%s, value=%s)" % (self.task_id, self.name, self.value)
 
 
-class TaskEvent(Base):
+class TaskEvent(Base):  # type: ignore
     """
     Table to track when a task is scheduled, starts, finishes, and fails.
     """
@@ -215,7 +215,7 @@ def __repr__(self):
         return "TaskEvent(task_id=%s, event_name=%s, ts=%s" % (self.task_id, self.event_name, self.ts)
 
 
-class TaskRecord(Base):
+class TaskRecord(Base):  # type: ignore
     """
     Base table to track information about a luigi.Task.
 

luigi/freezing.py

@@ -8,7 +8,7 @@
 try:
     from collections.abc import Mapping
 except ImportError:
-    from collections import Mapping
+    from collections import Mapping  # type: ignore
 import operator
 import functools