Skip to content

Commit

Permalink
Add typing to filepath
Browse files Browse the repository at this point in the history
  • Loading branch information
DanielNoord committed Sep 7, 2021
1 parent fab2de2 commit dc3696f
Show file tree
Hide file tree
Showing 9 changed files with 65 additions and 39 deletions.
3 changes: 2 additions & 1 deletion pylint/config/option_manager_mixin.py
Expand Up @@ -10,6 +10,7 @@
import optparse # pylint: disable=deprecated-module
import os
import sys
from typing import List

import toml

Expand Down Expand Up @@ -326,7 +327,7 @@ def load_configuration_from_config(self, config):
provider = self._all_options[opt]
provider.set_option(opt, opt_value)

def load_command_line_configuration(self, args=None):
def load_command_line_configuration(self, args=None) -> List[str]:
"""Override configuration according to command line parameters
return additional arguments
Expand Down
27 changes: 23 additions & 4 deletions pylint/lint/expand_modules.py
@@ -1,9 +1,28 @@
import os
import sys
from typing import List, Pattern, Tuple
from typing import List, Pattern, Tuple, Union

from astroid import modutils

if sys.version_info >= (3, 8):
from typing import Literal, TypedDict
else:
from typing_extensions import Literal, TypedDict


class MODULE_DESCRIPTION_DICT(TypedDict):
path: str
name: str
isarg: bool
basepath: str
basename: str


class ERROR_DESCRIPTION_DICT(TypedDict):
key: Literal["fatal"]
mod: str
ex: Union[ImportError, SyntaxError]


def _modpath_from_file(filename, is_namespace, path=None):
def _is_package_cb(path, parts):
Expand Down Expand Up @@ -42,12 +61,12 @@ def expand_modules(
ignore_list: List[str],
ignore_list_re: List[Pattern],
ignore_list_paths_re: List[Pattern],
) -> Tuple[List[dict], List[dict]]:
) -> Tuple[List[MODULE_DESCRIPTION_DICT], List[ERROR_DESCRIPTION_DICT]]:
"""take a list of files/modules/packages and return the list of tuple
(file, module name) which have to be actually checked
"""
result = []
errors = []
result: List[MODULE_DESCRIPTION_DICT] = []
errors: List[ERROR_DESCRIPTION_DICT] = []
path = sys.path.copy()

for something in files_or_modules:
Expand Down
11 changes: 8 additions & 3 deletions pylint/lint/parallel.py
Expand Up @@ -3,6 +3,7 @@

import collections
import functools
from typing import Any, DefaultDict, Iterator, List, Tuple

from pylint import reporters
from pylint.lint.utils import _patch_sys_path
Expand Down Expand Up @@ -62,9 +63,13 @@ def _worker_initialize(linter, arguments=None):
_patch_sys_path(arguments or ())


def _worker_check_single_file(file_item):
def _worker_check_single_file(
file_item: Tuple[str, str, str]
) -> Tuple[int, Any, str, Any, List[Tuple[Any, ...]], Any, Any, DefaultDict[Any, List]]:
name, filepath, modname = file_item

if not _worker_linter:
raise Exception("Worker linter not yet initialised")
_worker_linter.open()
_worker_linter.check_single_file(name, filepath, modname)
mapreduce_data = collections.defaultdict(list)
Expand Down Expand Up @@ -109,7 +114,7 @@ def _merge_mapreduce_data(linter, all_mapreduce_data):
checker.reduce_map_data(linter, collated_map_reduce_data[checker.name])


def check_parallel(linter, jobs, files, arguments=None):
def check_parallel(linter, jobs, files: Iterator[Tuple[str, str, str]], arguments=None):
"""Use the given linter to lint the files with given amount of workers (jobs)
This splits the work filestream-by-filestream. If you need to do work across
multiple files, as in the similarity-checker, then inherit from MapReduceMixin and
Expand Down Expand Up @@ -151,7 +156,7 @@ def check_parallel(linter, jobs, files, arguments=None):
linter.set_current_module(module, file_path)
for msg in messages:
msg = Message(*msg)
linter.reporter.handle_message(msg)
linter.reporter.handle_message(msg) # type: ignore # L134 makes linter always have a reporter attribute
all_stats.append(stats)
all_mapreduce_data[worker_idx].append(mapreduce_data)
linter.msg_status |= msg_status
Expand Down
34 changes: 19 additions & 15 deletions pylint/lint/pylinter.py
Expand Up @@ -11,13 +11,14 @@
import traceback
import warnings
from io import TextIOWrapper
from typing import Iterator, List, Optional, Tuple, Union

import astroid
from astroid import AstroidError

from pylint import checkers, config, exceptions, interfaces, reporters
from pylint.constants import MAIN_CHECKER_NAME, MSG_TYPES
from pylint.lint.expand_modules import expand_modules
from pylint.lint.expand_modules import MODULE_DESCRIPTION_DICT, expand_modules
from pylint.lint.parallel import check_parallel
from pylint.lint.report_functions import (
report_messages_by_module_stats,
Expand Down Expand Up @@ -935,16 +936,13 @@ def initialize(self):
if not msg.may_be_emitted():
self._msgs_state[msg.msgid] = False

def check(self, files_or_modules):
def check(self, files_or_modules: List[str]) -> None:
"""main checking entry: check a list of files or modules from their name.
files_or_modules is either a string or list of strings presenting modules to check.
files_or_modules is a list of strings presenting modules to check.
"""
self.initialize()

if not isinstance(files_or_modules, (list, tuple)):
files_or_modules = (files_or_modules,)

if self.config.from_stdin:
if len(files_or_modules) != 1:
raise exceptions.InvalidArgsError(
Expand All @@ -970,7 +968,7 @@ def check(self, files_or_modules):
files_or_modules,
)

def check_single_file(self, name, filepath, modname):
def check_single_file(self, name: str, filepath: str, modname: str) -> None:
"""Check single file
The arguments are the same that are documented in _check_files
Expand All @@ -982,7 +980,11 @@ def check_single_file(self, name, filepath, modname):
self.get_ast, check_astroid_module, name, filepath, modname
)

def _check_files(self, get_ast, file_descrs):
def _check_files(
self,
get_ast,
file_descrs: Union[List[Tuple[str, str, str]], Iterator[Tuple[str, str, str]]],
):
"""Check all files from file_descrs
The file_descrs should be iterable of tuple (name, filepath, modname)
Expand All @@ -1004,12 +1006,14 @@ def _check_files(self, get_ast, file_descrs):
msg = get_fatal_error_message(filepath, template_path)
if isinstance(ex, AstroidError):
symbol = "astroid-error"
msg = (filepath, msg)
self.add_message(symbol, args=(filepath, msg))
else:
symbol = "fatal"
self.add_message(symbol, args=msg)
self.add_message(symbol, args=msg)

def _check_file(self, get_ast, check_astroid_module, name, filepath, modname):
def _check_file(
self, get_ast, check_astroid_module, name: str, filepath: str, modname: str
):
"""Check a file using the passed utility functions (get_ast and check_astroid_module)
:param callable get_ast: callable returning AST from defined file taking the following arguments
Expand Down Expand Up @@ -1042,7 +1046,7 @@ def _check_file(self, get_ast, check_astroid_module, name, filepath, modname):
self.add_message(msgid, line, None, args)

@staticmethod
def _get_file_descr_from_stdin(filepath):
def _get_file_descr_from_stdin(filepath: str) -> Tuple[str, str, str]:
"""Return file description (tuple of module name, file path, base name) from given file path
This method is used for creating suitable file description for _check_files when the
Expand All @@ -1058,7 +1062,7 @@ def _get_file_descr_from_stdin(filepath):

return (modname, filepath, filepath)

def _iterate_file_descrs(self, files_or_modules):
def _iterate_file_descrs(self, files_or_modules) -> Iterator[Tuple[str, str, str]]:
"""Return generator yielding file descriptions (tuples of module name, file path, base name)
The returned generator yield one item for each Python module that should be linted.
Expand All @@ -1068,7 +1072,7 @@ def _iterate_file_descrs(self, files_or_modules):
if self.should_analyze_file(name, filepath, is_argument=is_arg):
yield (name, filepath, descr["basename"])

def _expand_files(self, modules):
def _expand_files(self, modules) -> List[MODULE_DESCRIPTION_DICT]:
"""get modules and errors from a list of modules and handle errors"""
result, errors = expand_modules(
modules,
Expand All @@ -1085,7 +1089,7 @@ def _expand_files(self, modules):
self.add_message(key, args=message)
return result

def set_current_module(self, modname, filepath=None):
def set_current_module(self, modname, filepath: Optional[str] = None):
"""set the name of the currently analyzed module and
init statistics for it
"""
Expand Down
9 changes: 3 additions & 6 deletions pylint/lint/utils.py
Expand Up @@ -5,8 +5,7 @@
import sys
import traceback
from datetime import datetime
from pathlib import Path, PosixPath
from typing import Union
from pathlib import Path

from pylint.config import PYLINT_HOME
from pylint.lint.expand_modules import get_python_path
Expand All @@ -16,9 +15,7 @@ class ArgumentPreprocessingError(Exception):
"""Raised if an error occurs during argument preprocessing."""


def prepare_crash_report(
ex: Exception, filepath: PosixPath, crash_file_path: Union[Path, str]
) -> Path:
def prepare_crash_report(ex: Exception, filepath: str, crash_file_path: str) -> Path:
issue_template_path = (
Path(PYLINT_HOME) / datetime.now().strftime(str(crash_file_path))
).resolve()
Expand Down Expand Up @@ -63,7 +60,7 @@ def prepare_crash_report(
return issue_template_path


def get_fatal_error_message(filepath: str, issue_template_path: str) -> str:
def get_fatal_error_message(filepath: str, issue_template_path: Path) -> str:
return (
f"Fatal error while checking '{filepath}'. "
f"Please open an issue in our bug tracker so we address this. "
Expand Down
4 changes: 2 additions & 2 deletions pylint/reporters/base_reporter.py
Expand Up @@ -3,7 +3,7 @@

import os
import sys
from typing import List
from typing import List, Optional

from pylint.message import Message

Expand Down Expand Up @@ -62,7 +62,7 @@ def display_messages(self, layout):

# Event callbacks

def on_set_current_module(self, module, filepath):
def on_set_current_module(self, module: str, filepath: Optional[str]) -> None:
"""Hook called when a module starts to be analysed."""

def on_close(self, stats, previous_stats):
Expand Down
5 changes: 2 additions & 3 deletions pylint/reporters/multi_reporter.py
Expand Up @@ -3,15 +3,14 @@


import os
from typing import IO, Any, AnyStr, Callable, List, Mapping, Optional, Union
from typing import IO, Any, AnyStr, Callable, List, Mapping, Optional

from pylint.interfaces import IReporter
from pylint.message import Message
from pylint.reporters.base_reporter import BaseReporter
from pylint.reporters.ureports.nodes import BaseLayout

AnyFile = IO[AnyStr]
AnyPath = Union[str, bytes, os.PathLike]
PyLinter = Any


Expand Down Expand Up @@ -88,7 +87,7 @@ def display_messages(self, layout: BaseLayout) -> None:
for rep in self._sub_reporters:
rep.display_messages(layout)

def on_set_current_module(self, module: str, filepath: Optional[AnyPath]) -> None:
def on_set_current_module(self, module: str, filepath: Optional[str]) -> None:
"""hook called when a module starts to be analysed"""
for rep in self._sub_reporters:
rep.on_set_current_module(module, filepath)
Expand Down
3 changes: 2 additions & 1 deletion pylint/reporters/text.py
Expand Up @@ -25,6 +25,7 @@
import os
import sys
import warnings
from typing import Optional

from pylint import utils
from pylint.interfaces import IReporter
Expand Down Expand Up @@ -136,7 +137,7 @@ def __init__(self, output=None):
self._modules = set()
self._template = self.line_format

def on_set_current_module(self, module, filepath):
def on_set_current_module(self, module: str, filepath: Optional[str]) -> None:
self._template = str(self.linter.config.msg_template or self._template)

def write_message(self, msg):
Expand Down
8 changes: 4 additions & 4 deletions pylint/testutils/reporter_for_tests.py
Expand Up @@ -3,7 +3,7 @@

from io import StringIO
from os import getcwd, linesep, sep
from typing import Dict, List
from typing import Dict, List, Optional

from pylint import interfaces
from pylint.message import Message
Expand Down Expand Up @@ -51,7 +51,7 @@ def finalize(self):
return result

# pylint: disable=unused-argument
def on_set_current_module(self, module, filepath):
def on_set_current_module(self, module: str, filepath: Optional[str]) -> None:
pass

# pylint: enable=unused-argument
Expand All @@ -63,14 +63,14 @@ def display_reports(self, layout):


class MinimalTestReporter(BaseReporter):
def on_set_current_module(self, module, filepath):
def on_set_current_module(self, module: str, filepath: Optional[str]) -> None:
self.messages = []

_display = None


class FunctionalTestReporter(BaseReporter):
def on_set_current_module(self, module, filepath):
def on_set_current_module(self, module: str, filepath: Optional[str]) -> None:
self.messages = []

def display_reports(self, layout):
Expand Down

0 comments on commit dc3696f

Please sign in to comment.