Add typing to references of stream and stream()

examples/custom_raw.py

@@ -1,3 +1,5 @@
+from astroid import nodes
+
 from pylint.checkers import BaseChecker
 from pylint.interfaces import IRawChecker
 
@@ -22,7 +24,7 @@ class MyRawChecker(BaseChecker):
     }
     options = ()
 
-    def process_module(self, node):
+    def process_module(self, node: nodes.Module) -> None:
         """process a module
 
         the module's content is accessible via node.stream() function

pylint/checkers/format.py

@@ -351,7 +351,7 @@ def new_line(self, tokens, line_end, line_start):
             self._lines[line_num] = line.split("\n")[0]
         self.check_lines(line, line_num)
 
-    def process_module(self, _module):
+    def process_module(self, _node: nodes.Module) -> None:
         pass
 
     def _check_keyword_parentheses(

pylint/checkers/misc.py

@@ -28,6 +28,9 @@
 
 import re
 import tokenize
+from typing import Optional
+
+from astroid import nodes
 
 from pylint.checkers import BaseChecker
 from pylint.interfaces import IRawChecker, ITokenChecker
@@ -50,11 +53,11 @@ class ByIdManagedMessagesChecker(BaseChecker):
     }
     options = ()
 
-    def process_module(self, module):
+    def process_module(self, node: nodes.Module) -> None:
         """Inspect the source file to find messages activated or deactivated by id."""
         managed_msgs = MessagesHandlerMixIn.get_by_id_managed_msgs()
         for (mod_name, msgid, symbol, lineno, is_disabled) in managed_msgs:
-            if mod_name == module.name:
+            if mod_name == node.name:
                 verb = "disable" if is_disabled else "enable"
                 txt = f"'{msgid}' is cryptic: use '# pylint: {verb}={symbol}' instead"
                 self.add_message("use-symbolic-message-instead", line=lineno, args=txt)
@@ -114,22 +117,28 @@ def open(self):
 
         self._fixme_pattern = re.compile(regex_string, re.I)
 
-    def _check_encoding(self, lineno, line, file_encoding):
+    def _check_encoding(
+        self, lineno: int, line: bytes, file_encoding: str
+    ) -> Optional[str]:
         try:
             return line.decode(file_encoding)
         except UnicodeDecodeError:
             pass
         except LookupError:
-            if line.startswith("#") and "coding" in line and file_encoding in line:
+            if (
+                line.startswith(b"#")
+                and "coding" in str(line)
+                and file_encoding in str(line)
+            ):
                 msg = f"Cannot decode using encoding '{file_encoding}', bad encoding"
                 self.add_message("syntax-error", line=lineno, args=msg)
         return None
 
-    def process_module(self, module):
+    def process_module(self, node: nodes.Module) -> None:
         """inspect the source file to find encoding problem"""
-        encoding = module.file_encoding if module.file_encoding else "ascii"
+        encoding = node.file_encoding if node.file_encoding else "ascii"
 
-        with module.stream() as stream:
+        with node.stream() as stream:
             for lineno, line in enumerate(stream):
                 self._check_encoding(lineno + 1, line, encoding)
 

pylint/checkers/similar.py

@@ -49,6 +49,7 @@
 import sys
 from collections import defaultdict
 from getopt import getopt
+from io import BufferedIOBase, BufferedReader, BytesIO
 from itertools import chain, groupby
 from typing import (
     Any,
@@ -59,9 +60,11 @@
     List,
     NamedTuple,
     NewType,
+    Optional,
     Set,
     TextIO,
     Tuple,
+    Union,
 )
 
 import astroid
@@ -96,6 +99,9 @@ class LineSpecifs(NamedTuple):
 # Links index in the lineset's stripped lines to the real lines in the file
 IndexToLines_T = Dict[Index, "SuccessiveLinesLimits"]
 
+# The types the streams read by pylint can take. Originating from astroid.nodes.Module.stream() and open()
+STREAM_TYPES = Union[TextIO, BufferedReader, BytesIO]
+
 
 class CplSuccessiveLinesLimits:
     """
@@ -368,12 +374,16 @@ def __init__(
         self.ignore_signatures = ignore_signatures
         self.linesets: List["LineSet"] = []
 
-    def append_stream(self, streamid: str, stream: TextIO, encoding=None) -> None:
+    def append_stream(
+        self, streamid: str, stream: STREAM_TYPES, encoding: Optional[str] = None
+    ) -> None:
         """append a file to search for similarities"""
-        if encoding is None:
-            readlines = stream.readlines
-        else:
+        if isinstance(stream, BufferedIOBase):
+            if encoding is None:
+                raise ValueError
             readlines = decoding_stream(stream, encoding).readlines
+        else:
+            readlines = stream.readlines  # type: ignore # hint parameter is incorrectly typed as non-optional
         try:
             self.linesets.append(
                 LineSet(
@@ -658,12 +668,12 @@ class LineSet:
 
     def __init__(
         self,
-        name,
-        lines,
-        ignore_comments=False,
-        ignore_docstrings=False,
-        ignore_imports=False,
-        ignore_signatures=False,
+        name: str,
+        lines: List[str],
+        ignore_comments: bool = False,
+        ignore_docstrings: bool = False,
+        ignore_imports: bool = False,
+        ignore_signatures: bool = False,
     ) -> None:
         self.name = name
         self._real_lines = lines
@@ -820,7 +830,7 @@ def open(self):
             nb_duplicated_lines=0, percent_duplicated_lines=0
         )
 
-    def process_module(self, node):
+    def process_module(self, node: nodes.Module) -> None:
         """process a module
 
         the module's content is accessible via the stream object

pylint/checkers/strings.py

@@ -710,8 +710,8 @@ def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.string_tokens = {}  # token position -> (token value, next token)
 
-    def process_module(self, module):
-        self._unicode_literals = "unicode_literals" in module.future_imports
+    def process_module(self, node: nodes.Module) -> None:
+        self._unicode_literals = "unicode_literals" in node.future_imports
 
     def process_tokens(self, tokens):
         encoding = "ascii"

pylint/config/man_help_formatter.py

@@ -14,6 +14,7 @@ def __init__(
         optparse.HelpFormatter.__init__(
             self, indent_increment, max_help_position, width, short_first
         )
+        self.output_level: int
 
     def format_heading(self, heading):
         return f".SH {heading.upper()}\n"

pylint/config/option_manager_mixin.py

@@ -10,6 +10,8 @@
 import optparse  # pylint: disable=deprecated-module
 import os
 import sys
+from types import ModuleType
+from typing import Dict, List, Optional, TextIO, Tuple
 
 import toml
 
@@ -186,11 +188,13 @@ def global_set_option(self, opt, value):
         """set option on the correct option provider"""
         self._all_options[opt].set_option(opt, value)
 
-    def generate_config(self, stream=None, skipsections=()):
+    def generate_config(
+        self, stream: Optional[TextIO] = None, skipsections: Tuple[str, ...] = ()
+    ) -> None:
         """write a configuration file according to the current configuration
         into the given stream or stdout
         """
-        options_by_section = {}
+        options_by_section: Dict[str, List[Tuple]] = {}
         sections = []
         for provider in self.options_providers:
             for section, options in provider.options_by_section():
@@ -219,7 +223,9 @@ def generate_config(self, stream=None, skipsections=()):
             )
             printed = True
 
-    def generate_manpage(self, pkginfo, section=1, stream=sys.stdout):
+    def generate_manpage(
+        self, pkginfo: ModuleType, section: int = 1, stream: TextIO = sys.stdout
+    ) -> None:
         with _patch_optparse():
             formatter = _ManHelpFormatter()
             formatter.output_level = self._maxlevel

pylint/extensions/empty_comment.py

@@ -1,3 +1,5 @@
+from astroid import nodes
+
 from pylint.checkers import BaseChecker
 from pylint.interfaces import IRawChecker
 
@@ -43,7 +45,7 @@ class CommentChecker(BaseChecker):
     options = ()
     priority = -1  # low priority
 
-    def process_module(self, node):
+    def process_module(self, node: nodes.Module) -> None:
         with node.stream() as stream:
             for (line_num, line) in enumerate(stream):
                 line = line.rstrip()

pylint/interfaces.py

@@ -18,6 +18,8 @@
 """Interfaces for Pylint objects"""
 from collections import namedtuple
 
+from astroid import nodes
+
 Confidence = namedtuple("Confidence", ["name", "description"])
 # Warning Certainties
 HIGH = Confidence("HIGH", "No false positive possible.")
@@ -66,7 +68,7 @@ def close(self):
 class IRawChecker(IChecker):
     """interface for checker which need to parse the raw file"""
 
-    def process_module(self, astroid):
+    def process_module(self, node: nodes.Module) -> None:
         """process a module
 
         the module's content is accessible via astroid.stream

pylint/lint/pylinter.py

@@ -13,7 +13,7 @@
 from io import TextIOWrapper
 
 import astroid
-from astroid import AstroidError
+from astroid import AstroidError, nodes
 
 from pylint import checkers, config, exceptions, interfaces, reporters
 from pylint.constants import MAIN_CHECKER_NAME, MSG_TYPES
@@ -1180,10 +1180,12 @@ def check_astroid_module(self, ast_node, walker, rawcheckers, tokencheckers):
 
         return retval
 
-    def _check_astroid_module(self, ast_node, walker, rawcheckers, tokencheckers):
+    def _check_astroid_module(
+        self, node: nodes.Module, walker, rawcheckers, tokencheckers
+    ):
         """Check given AST node with given walker and checkers
 
-        :param astroid.nodes.Module ast_node: AST node of the module to check
+        :param astroid.nodes.Module node: AST node of the module to check
         :param pylint.utils.ast_walker.ASTWalker walker: AST walker
         :param list rawcheckers: List of token checkers to use
         :param list tokencheckers: List of raw checkers to use
@@ -1193,13 +1195,13 @@ def _check_astroid_module(self, ast_node, walker, rawcheckers, tokencheckers):
         :rtype: bool
         """
         try:
-            tokens = utils.tokenize_module(ast_node)
+            tokens = utils.tokenize_module(node)
         except tokenize.TokenError as ex:
             self.add_message("syntax-error", line=ex.args[1][0], args=ex.args[0])
             return None
 
-        if not ast_node.pure_python:
-            self.add_message("raw-checker-failed", args=ast_node.name)
+        if not node.pure_python:
+            self.add_message("raw-checker-failed", args=node.name)
         else:
             # assert astroid.file.endswith('.py')
             # invoke ITokenChecker interface on self to fetch module/block
@@ -1208,14 +1210,14 @@ def _check_astroid_module(self, ast_node, walker, rawcheckers, tokencheckers):
             if self._ignore_file:
                 return False
             # walk ast to collect line numbers
-            self.file_state.collect_block_lines(self.msgs_store, ast_node)
+            self.file_state.collect_block_lines(self.msgs_store, node)
             # run raw and tokens checkers
             for checker in rawcheckers:
-                checker.process_module(ast_node)
+                checker.process_module(node)
             for checker in tokencheckers:
                 checker.process_tokens(tokens)
         # generate events to astroid checkers
-        walker.walk(ast_node)
+        walker.walk(node)
         return True
 
     # IAstroidChecker interface #################################################

pylint/message/message_handler_mix_in.py

@@ -2,7 +2,8 @@
 # For details: https://github.com/PyCQA/pylint/blob/main/LICENSE
 
 import sys
-from typing import List, Tuple, Union
+from io import TextIOWrapper
+from typing import List, TextIO, Tuple, Union
 
 from pylint.constants import (
     _SCOPE_EXEMPT,
@@ -390,20 +391,13 @@ def get_checkers_documentation(self):
             result += checker.get_full_documentation(**information)
         return result
 
-    def print_full_documentation(self, stream=None):
+    def print_full_documentation(self, stream: TextIO = sys.stdout) -> None:
         """output a full documentation in ReST format"""
-        if not stream:
-            stream = sys.stdout
         print(self.get_checkers_documentation()[:-1], file=stream)
 
     @staticmethod
-    def _print_checker_doc(information, stream=None):
-        """Helper method for print_full_documentation.
-
-        Also used by doc/exts/pylint_extensions.py.
-        """
-        if not stream:
-            stream = sys.stdout
+    def _print_checker_doc(information, stream: TextIOWrapper) -> None:
+        """Helper method used by doc/exts/pylint_extensions.py."""
         checker = information["checker"]
         del information["checker"]
         print(checker.get_full_documentation(**information)[:-1], file=stream)

pylint/reporters/ureports/__init__.py

@@ -18,20 +18,19 @@
 import os
 import sys
 from io import StringIO
+from typing import Iterator, TextIO
 
 
 class BaseWriter:
     """base class for ureport writers"""
 
-    def format(self, layout, stream=None, encoding=None):
+    def format(self, layout, stream: TextIO = sys.stdout, encoding=None) -> None:
         """format and write the given layout into the stream object
 
         unicode policy: unicode strings may be found in the layout;
         try to call stream.write with it, but give it back encoded using
         the given encoding if it fails
         """
-        if stream is None:
-            stream = sys.stdout
         if not encoding:
             encoding = getattr(stream, "encoding", "UTF-8")
         self.encoding = encoding or "UTF-8"
@@ -79,7 +78,7 @@ def get_table_content(self, table):
         result[-1] += [""] * (cols - len(result[-1]))
         return result
 
-    def compute_content(self, layout):
+    def compute_content(self, layout) -> Iterator[str]:
         """trick to compute the formatting of children layout before actually
         writing it