Implementation:Spotify Luigi Mypy Plugin

Knowledge Sources	Spotify_Luigi
Domains	Static_Analysis, Type_Checking
Last Updated	2026-02-10 08:00 GMT

Overview

The Luigi Mypy Plugin is a mypy plugin that provides proper type inference for Luigi Task classes, automatically generating typed __init__ signatures from Parameter declarations and correctly resolving Parameter return types.

Description

The Luigi mypy plugin (luigi.mypy) enhances static type checking for Luigi Task classes by bridging the gap between Luigi's runtime parameter system and mypy's static analysis. Without this plugin, mypy cannot understand that class-level Parameter() assignments become typed constructor arguments at runtime. The plugin contains three core classes:

• TaskPlugin (extends mypy.plugin.Plugin) -- The main plugin entry point. It implements two hooks:
  • get_base_class_hook() -- Triggered for any class in the MRO chain of luigi.task.Task. When a Task subclass is encountered, it invokes the TaskTransformer to analyze Parameter declarations and synthesize a properly typed __init__ method.
  • get_function_hook() -- Triggered when a Luigi Parameter constructor is called (e.g., luigi.IntParameter(), luigi.Parameter()). It infers the return type of the Parameter from either the __new__ method return type, the generic type argument of the Parameter base class, or the default argument type. Special handling exists for ChoiceListParameter, EnumListParameter, ChoiceParameter, and EnumParameter.
  • check_parameter() -- Utility method that checks whether a fully qualified name refers to a class in the luigi.parameter.Parameter MRO.

• TaskAttribute -- Represents a single Parameter declaration on a Task class. It stores the attribute name, type, default status, source location, and owning TypeInfo. It can serialize/deserialize itself for cross-module metadata propagation, expand type variables when inherited by subclasses, and generate mypy Argument objects for the synthesized __init__ method. All generated arguments are keyword-only and optional (ARG_NAMED_OPT).

• TaskTransformer -- Performs the actual transformation of a Task class definition. Its transform() method collects attributes from the entire MRO (respecting inheritance and overrides), filters for Parameter calls, infers init types (including support for __set__ descriptor signatures), and adds a synthesized __init__ method to the class using add_method_to_class(). Attribute metadata is stored in TypeInfo.metadata under the task key for consumption by subclass transformers.

The module-level plugin(version: str) function is the standard mypy plugin entry point that returns the TaskPlugin class. The plugin requires Python 3.8+ due to use of the walrus operator.

Usage

Use the Luigi mypy plugin when you want full static type checking support for Luigi Task classes in your codebase. It is configured via mypy.ini or pyproject.toml and requires no code changes. It is particularly valuable for large Luigi codebases where type safety helps catch parameter misuse, missing required parameters, and type mismatches at development time rather than runtime. The plugin correctly handles Task inheritance hierarchies, ensuring that parameters defined in parent classes appear in child class constructors.

Code Reference

Source Location

• Repository: Spotify_Luigi
• File: luigi/mypy.py
• Lines: 1-482

Signature

METADATA_TAG: Final[str] = "task"

class TaskPlugin(Plugin):
    def get_base_class_hook(
        self, fullname: str
    ) -> Callable[[ClassDefContext], None] | None: ...
    def get_function_hook(
        self, fullname: str
    ) -> Callable[[FunctionContext], Type] | None: ...
    def check_parameter(self, fullname: str) -> bool: ...

class TaskAttribute:
    def __init__(
        self, name: str, has_default: bool, line: int, column: int,
        type: Type | None, info: TypeInfo,
        api: SemanticAnalyzerPluginInterface,
    ) -> None: ...
    def to_argument(
        self, current_info: TypeInfo, *, of: Literal["__init__"]
    ) -> Argument: ...
    def expand_type(self, current_info: TypeInfo) -> Type | None: ...
    def to_var(self, current_info: TypeInfo) -> Var: ...
    def serialize(self) -> JsonDict: ...
    @classmethod
    def deserialize(
        cls, info: TypeInfo, data: JsonDict,
        api: SemanticAnalyzerPluginInterface
    ) -> TaskAttribute: ...
    def expand_typevar_from_subtype(self, sub_type: TypeInfo) -> None: ...

class TaskTransformer:
    def __init__(
        self, cls: ClassDef, reason: Expression | Statement,
        api: SemanticAnalyzerPluginInterface,
        task_plugin: TaskPlugin,
    ) -> None: ...
    def transform(self) -> bool: ...
    def collect_attributes(self) -> Optional[List[TaskAttribute]]: ...
    def is_parameter_call(self, expr: Expression) -> bool: ...

def plugin(version: str) -> type[Plugin]: ...

Import

# Configured via mypy.ini or pyproject.toml (not imported directly in user code):

# mypy.ini
[mypy]
plugins = luigi.mypy

# pyproject.toml
[tool.mypy]
plugins = ["luigi.mypy"]

I/O Contract

Inputs

Name	Type	Required	Description
version	str	Yes	Mypy version string passed by the mypy plugin system to the plugin() entry point
Task class definitions	ClassDef	Yes	Any class in the MRO of luigi.task.Task triggers the base class hook
Parameter calls	CallExpr	Yes	Calls to luigi.Parameter(), luigi.IntParameter(), etc. trigger the function hook

Outputs

Name	Type	Description
Synthesized __init__	FuncDef	A typed __init__ method added to Task classes with keyword-only optional arguments for each Parameter
Parameter return types	Type	Inferred types for Parameter declarations (e.g., int for IntParameter, str for Parameter)
Metadata	JsonDict	Serialized attribute information stored in TypeInfo.metadata["task"] for cross-module inheritance

Usage Examples

Basic Usage

# mypy.ini configuration:
# [mypy]
# plugins = luigi.mypy

import luigi


class MyTask(luigi.Task):
    name = luigi.Parameter(default='world')
    count = luigi.IntParameter(default=1)
    rate = luigi.FloatParameter()

    def run(self):
        # mypy now knows:
        #   self.name is str
        #   self.count is int
        #   self.rate is float
        print(f"Hello {self.name} x{self.count} at rate {self.rate}")


# mypy understands the constructor signature:
task = MyTask(name='test', count=5, rate=0.5)

# mypy will flag type errors:
# task = MyTask(count='not_an_int')  # Error: Argument "count" has incompatible type "str"; expected "int"

Inheritance

import luigi


class BaseTask(luigi.Task):
    date = luigi.DateParameter()
    environment = luigi.Parameter(default='production')


class DerivedTask(BaseTask):
    extra_param = luigi.IntParameter(default=10)

    def run(self):
        # mypy knows all three parameters are available:
        #   self.date, self.environment, self.extra_param
        pass


# Constructor includes all inherited parameters:
task = DerivedTask(date='2024-01-01', environment='staging', extra_param=20)

Related Pages

• Principle:Spotify_Luigi_Static_Type_Checking