Typing for ./pylint: typing for ./pylint/utils

pylint/checkers/similar.py

@@ -49,10 +49,11 @@
 import sys
 from collections import defaultdict
 from getopt import getopt
-from io import TextIOWrapper
+from io import BufferedReader, BytesIO
 from itertools import chain, groupby
 from typing import (
     Any,
+    Callable,
     Dict,
     FrozenSet,
     Generator,
@@ -62,6 +63,7 @@
     NewType,
     Set,
     Tuple,
+    Union,
 )
 
 import astroid
@@ -369,11 +371,11 @@ def __init__(
         self.linesets: List["LineSet"] = []
 
     def append_stream(
-        self, streamid: str, stream: TextIOWrapper, encoding=None
+        self, streamid: str, stream: Union[BufferedReader, BytesIO], encoding=None
     ) -> None:
         """append a file to search for similarities"""
         if encoding is None:
-            readlines = stream.readlines
+            readlines: Union[Callable[..., Union[str, bytes]], Any] = stream.readlines
         else:
             readlines = decoding_stream(stream, encoding).readlines
         try:

pylint/extensions/code_style.py

@@ -79,9 +79,8 @@ def __init__(self, linter: PyLinter) -> None:
     def open(self) -> None:
         py_version: Tuple[int, int] = get_global_option(self, "py-version")  # type: ignore
         self._py38_plus = py_version >= (3, 8)
-        self._max_length: int = (  # type: ignore
-            self.config.max_line_length_suggestions
-            or get_global_option(self, "max-line-length")
+        self._max_length: int = self.config.max_line_length_suggestions or get_global_option(  # type: ignore
+            self, "max-line-length"  # type: ignore # This is unnecessary, but black does not handle this correctly
         )
 
     @check_messages("consider-using-namedtuple-or-dataclass")

pylint/extensions/docparams.py

@@ -522,7 +522,8 @@ class constructor.
             expected_but_ignored_argument_names = {
                 arg
                 for arg in expected_argument_names
-                if ignored_argument_names.match(arg)
+                if isinstance(ignored_argument_names, re.Pattern)
+                and ignored_argument_names.match(arg)
             }
 
         if arguments_node.vararg is not None:

pylint/utils/ast_walker.py

@@ -3,28 +3,32 @@
 
 import collections
 import traceback
+from typing import TYPE_CHECKING, Any, Callable
 
 from astroid import nodes
 
+if TYPE_CHECKING:
+    from pylint.lint.pylinter import PyLinter
+
 
 class ASTWalker:
-    def __init__(self, linter):
+    def __init__(self, linter: "PyLinter") -> None:
         # callbacks per node types
         self.nbstatements = 0
-        self.visit_events = collections.defaultdict(list)
-        self.leave_events = collections.defaultdict(list)
+        self.visit_events: collections.defaultdict = collections.defaultdict(list)
+        self.leave_events: collections.defaultdict = collections.defaultdict(list)
         self.linter = linter
         self.exception_msg = False
 
-    def _is_method_enabled(self, method):
+    def _is_method_enabled(self, method: Callable) -> bool:
         if not hasattr(method, "checks_msgs"):
             return True
-        for msg_desc in method.checks_msgs:
+        for msg_desc in method.checks_msgs:  # type: ignore # mypy does not pick up that method always has checks_msgs
             if self.linter.is_message_enabled(msg_desc):
                 return True
         return False
 
-    def add_checker(self, checker):
+    def add_checker(self, checker: Any) -> None:
         """walk to the checker's dir and collect visit and leave methods"""
         vcids = set()
         lcids = set()
@@ -54,7 +58,7 @@ def add_checker(self, checker):
                     visits[cid].append(visit_default)
         # for now we have no "leave_default" method in Pylint
 
-    def walk(self, astroid):
+    def walk(self, astroid: Any) -> None:
         """call visit events of astroid checkers for the given node, recurse on
         its children, then leave events.
         """

pylint/utils/file_state.py

@@ -2,24 +2,35 @@
 # For details: https://github.com/PyCQA/pylint/blob/main/LICENSE
 
 import collections
+from typing import TYPE_CHECKING, Any, Dict, Iterator, Optional, Set, Tuple, Union
 
 from astroid import nodes
+from astroid.nodes.scoped_nodes import Module
 
 from pylint.constants import MSG_STATE_SCOPE_MODULE, WarningScope
+from pylint.interfaces import Confidence
+
+if TYPE_CHECKING:
+    from pylint.message.message_definition import MessageDefinition
+    from pylint.message.message_definition_store import MessageDefinitionStore
 
 
 class FileState:
     """Hold internal state specific to the currently analyzed file"""
 
-    def __init__(self, modname=None):
+    def __init__(self, modname: Optional[str] = None) -> None:
         self.base_name = modname
-        self._module_msgs_state = {}
-        self._raw_module_msgs_state = {}
-        self._ignored_msgs = collections.defaultdict(set)
-        self._suppression_mapping = {}
+        self._module_msgs_state: Dict[str, Dict[int, bool]] = {}
+        self._raw_module_msgs_state: Dict[str, Dict[int, bool]] = {}
+        self._ignored_msgs: collections.defaultdict[
+            Tuple[str, int], Set[int]
+        ] = collections.defaultdict(set)
+        self._suppression_mapping: Dict[Tuple[str, int], int] = {}
         self._effective_max_line_number = None
 
-    def collect_block_lines(self, msgs_store, module_node):
+    def collect_block_lines(
+        self, msgs_store: "MessageDefinitionStore", module_node: Module
+    ) -> None:
         """Walk the AST to collect block level options line numbers."""
         for msg, lines in self._module_msgs_state.items():
             self._raw_module_msgs_state[msg] = lines.copy()
@@ -29,7 +40,12 @@ def collect_block_lines(self, msgs_store, module_node):
         self._effective_max_line_number = module_node.tolineno
         self._collect_block_lines(msgs_store, module_node, orig_state)
 
-    def _collect_block_lines(self, msgs_store, node, msg_state):
+    def _collect_block_lines(
+        self,
+        msgs_store: "MessageDefinitionStore",
+        node: Any,
+        msg_state: Union[Dict[str, Dict[int, bool]], Dict[str, Dict]],
+    ) -> None:
         """Recursively walk (depth first) AST to collect block level options
         line numbers.
         """
@@ -90,7 +106,7 @@ def _collect_block_lines(self, msgs_store, node, msg_state):
                         self._module_msgs_state[msgid] = {line: state}
                 del lines[lineno]
 
-    def set_msg_status(self, msg, line, status):
+    def set_msg_status(self, msg: "MessageDefinition", line: int, status: bool) -> None:
         """Set status (enabled/disable) for a given message at a given line"""
         assert line > 0
         try:
@@ -99,8 +115,14 @@ def set_msg_status(self, msg, line, status):
             self._module_msgs_state[msg.msgid] = {line: status}
 
     def handle_ignored_message(
-        self, state_scope, msgid, line, node, args, confidence
-    ):  # pylint: disable=unused-argument
+        self,
+        state_scope: int,
+        msgid: str,
+        line: int,
+        _: nodes.NodeNG,
+        __: Any,
+        ___: Confidence,
+    ) -> None:  # pylint: disable=unused-argument
         """Report an ignored message.
 
         state_scope is either MSG_STATE_SCOPE_MODULE or MSG_STATE_SCOPE_CONFIG,
@@ -114,7 +136,13 @@ def handle_ignored_message(
             except KeyError:
                 pass
 
-    def iter_spurious_suppression_messages(self, msgs_store):
+    def iter_spurious_suppression_messages(
+        self, msgs_store: "MessageDefinitionStore"
+    ) -> Union[
+        Iterator,
+        Iterator[Tuple[str, int, Tuple[str, int]]],
+        Iterator[Tuple[str, int, Tuple[str]]],
+    ]:
         for warning, lines in self._raw_module_msgs_state.items():
             for line, enable in lines.items():
                 if not enable and (warning, line) not in self._ignored_msgs:
@@ -125,12 +153,12 @@ def iter_spurious_suppression_messages(self, msgs_store):
                             msgs_store.get_msg_display_string(warning),
                         )
         # don't use iteritems here, _ignored_msgs may be modified by add_message
-        for (warning, from_), lines in list(self._ignored_msgs.items()):
-            for line in lines:
+        for (warning, from_), ignored_lines in list(self._ignored_msgs.items()):
+            for line in ignored_lines:
                 yield "suppressed-message", line, (
                     msgs_store.get_msg_display_string(warning),
                     from_,
                 )
 
-    def get_effective_max_line_number(self):
+    def get_effective_max_line_number(self) -> Optional[int]:
         return self._effective_max_line_number

pylint/utils/pragma_parser.py

@@ -3,7 +3,7 @@
 
 import re
 from collections import namedtuple
-from typing import Generator, List
+from typing import Generator, List, Optional
 
 # Allow stopping after the first semicolon/hash encountered,
 # so that an option can be continued with the reasons
@@ -52,7 +52,7 @@
 )
 
 
-def emit_pragma_representer(action, messages):
+def emit_pragma_representer(action: str, messages: List[str]) -> PragmaRepresenter:
     if not messages and action in MESSAGE_KEYWORDS:
         raise InvalidPragmaError(
             "The keyword is not followed by message identifier", action
@@ -65,7 +65,7 @@ class PragmaParserError(Exception):
     A class for exceptions thrown by pragma_parser module
     """
 
-    def __init__(self, message, token):
+    def __init__(self, message: str, token: str) -> None:
         """
         :args message: explain the reason why the exception has been thrown
         :args token: token concerned by the exception
@@ -88,7 +88,7 @@ class InvalidPragmaError(PragmaParserError):
 
 
 def parse_pragma(pylint_pragma: str) -> Generator[PragmaRepresenter, None, None]:
-    action = None
+    action: Optional[str] = None
     messages: List[str] = []
     assignment_required = False
     previous_token = ""
@@ -112,7 +112,7 @@ def parse_pragma(pylint_pragma: str) -> Generator[PragmaRepresenter, None, None]
                 # Nothing at all detected before this assignment
                 raise InvalidPragmaError("Missing keyword before assignment", "")
             assignment_required = False
-        elif assignment_required:
+        elif assignment_required and action:
             raise InvalidPragmaError("The = sign is missing after the keyword", action)
         elif kind == "KEYWORD":
             if action: