saving dependency work

This commit is contained in:
2022-12-10 21:17:54 -06:00
parent bf923c3406
commit c2c19e5382
5636 changed files with 666283 additions and 8587 deletions
@@ -0,0 +1,55 @@
# Licensed under the GPL: https://www.gnu.org/licenses/old-licenses/gpl-2.0.html
# For details: https://github.com/PyCQA/pylint/blob/main/LICENSE
# Copyright (c) https://github.com/PyCQA/pylint/blob/main/CONTRIBUTORS.txt
"""Some various utilities and helper classes, most of them used in the
main pylint class.
"""
from pylint.utils.ast_walker import ASTWalker
from pylint.utils.docs import print_full_documentation
from pylint.utils.file_state import FileState
from pylint.utils.linterstats import LinterStats, ModuleStats, merge_stats
from pylint.utils.utils import (
HAS_ISORT_5,
IsortDriver,
_check_csv,
_format_option_value,
_splitstrip,
_unquote,
decoding_stream,
diff_string,
format_section,
get_global_option,
get_module_and_frameid,
get_rst_section,
get_rst_title,
normalize_text,
register_plugins,
tokenize_module,
)
__all__ = [
"ASTWalker",
"HAS_ISORT_5",
"IsortDriver",
"_check_csv",
"_format_option_value",
"_splitstrip",
"_unquote",
"decoding_stream",
"diff_string",
"FileState",
"format_section",
"get_global_option",
"get_module_and_frameid",
"get_rst_section",
"get_rst_title",
"normalize_text",
"register_plugins",
"tokenize_module",
"merge_stats",
"LinterStats",
"ModuleStats",
"print_full_documentation",
]
@@ -0,0 +1,105 @@
# Licensed under the GPL: https://www.gnu.org/licenses/old-licenses/gpl-2.0.html
# For details: https://github.com/PyCQA/pylint/blob/main/LICENSE
# Copyright (c) https://github.com/PyCQA/pylint/blob/main/CONTRIBUTORS.txt
from __future__ import annotations
import sys
import traceback
from collections import defaultdict
from collections.abc import Sequence
from typing import TYPE_CHECKING, Callable
from astroid import nodes
if TYPE_CHECKING:
from pylint.checkers.base_checker import BaseChecker
from pylint.lint import PyLinter
# Callable parameter type NodeNG not completely correct.
# Due to contravariance of Callable parameter types,
# it should be a Union of all NodeNG subclasses.
# However, since the methods are only retrieved with
# getattr(checker, member) and thus are inferred as Any,
# NodeNG will work too.
AstCallback = Callable[[nodes.NodeNG], None]
class ASTWalker:
def __init__(self, linter: PyLinter) -> None:
# callbacks per node types
self.nbstatements = 0
self.visit_events: defaultdict[str, list[AstCallback]] = defaultdict(list)
self.leave_events: defaultdict[str, list[AstCallback]] = defaultdict(list)
self.linter = linter
self.exception_msg = False
def _is_method_enabled(self, method: AstCallback) -> bool:
if not hasattr(method, "checks_msgs"):
return True
return any(self.linter.is_message_enabled(m) for m in method.checks_msgs) # type: ignore[attr-defined]
def add_checker(self, checker: BaseChecker) -> None:
"""Walk to the checker's dir and collect visit and leave methods."""
vcids: set[str] = set()
lcids: set[str] = set()
visits = self.visit_events
leaves = self.leave_events
for member in dir(checker):
cid = member[6:]
if cid == "default":
continue
if member.startswith("visit_"):
v_meth = getattr(checker, member)
# don't use visit_methods with no activated message:
if self._is_method_enabled(v_meth):
visits[cid].append(v_meth)
vcids.add(cid)
elif member.startswith("leave_"):
l_meth = getattr(checker, member)
# don't use leave_methods with no activated message:
if self._is_method_enabled(l_meth):
leaves[cid].append(l_meth)
lcids.add(cid)
visit_default = getattr(checker, "visit_default", None)
if visit_default:
for cls in nodes.ALL_NODE_CLASSES:
cid = cls.__name__.lower()
if cid not in vcids:
visits[cid].append(visit_default)
# For now, we have no "leave_default" method in Pylint
def walk(self, astroid: nodes.NodeNG) -> None:
"""Call visit events of astroid checkers for the given node, recurse on
its children, then leave events.
"""
cid = astroid.__class__.__name__.lower()
# Detect if the node is a new name for a deprecated alias.
# In this case, favour the methods for the deprecated
# alias if any, in order to maintain backwards
# compatibility.
visit_events: Sequence[AstCallback] = self.visit_events.get(cid, ())
leave_events: Sequence[AstCallback] = self.leave_events.get(cid, ())
try:
if astroid.is_statement:
self.nbstatements += 1
# generate events for this node on each checker
for callback in visit_events:
callback(astroid)
# recurse on children
for child in astroid.get_children():
self.walk(child)
for callback in leave_events:
callback(astroid)
except Exception:
if self.exception_msg is False:
file = getattr(astroid.root(), "file", None)
print(
f"Exception on node {repr(astroid)} in file '{file}'",
file=sys.stderr,
)
traceback.print_exc()
self.exception_msg = True
raise
@@ -0,0 +1,103 @@
# Licensed under the GPL: https://www.gnu.org/licenses/old-licenses/gpl-2.0.html
# For details: https://github.com/PyCQA/pylint/blob/main/LICENSE
# Copyright (c) https://github.com/PyCQA/pylint/blob/main/CONTRIBUTORS.txt
"""Various helper functions to create the docs of a linter object."""
from __future__ import annotations
import sys
import warnings
from typing import TYPE_CHECKING, Any, TextIO
from pylint.constants import MAIN_CHECKER_NAME
from pylint.utils.utils import get_rst_section, get_rst_title
if TYPE_CHECKING:
from pylint.lint.pylinter import PyLinter
def _get_checkers_infos(linter: PyLinter) -> dict[str, dict[str, Any]]:
"""Get info from a checker and handle KeyError."""
by_checker: dict[str, dict[str, Any]] = {}
for checker in linter.get_checkers():
name = checker.name
if name != MAIN_CHECKER_NAME:
try:
by_checker[name]["checker"] = checker
with warnings.catch_warnings():
warnings.filterwarnings("ignore", category=DeprecationWarning)
by_checker[name]["options"] += checker.options_and_values()
by_checker[name]["msgs"].update(checker.msgs)
by_checker[name]["reports"] += checker.reports
except KeyError:
with warnings.catch_warnings():
warnings.filterwarnings("ignore", category=DeprecationWarning)
by_checker[name] = {
"checker": checker,
"options": list(checker.options_and_values()),
"msgs": dict(checker.msgs),
"reports": list(checker.reports),
}
return by_checker
def _get_global_options_documentation(linter: PyLinter) -> str:
"""Get documentation for the main checker."""
result = get_rst_title("Pylint global options and switches", "-")
result += """
Pylint provides global options and switches.
"""
for checker in linter.get_checkers():
if checker.name == MAIN_CHECKER_NAME and checker.options:
with warnings.catch_warnings():
warnings.filterwarnings("ignore", category=DeprecationWarning)
for section, options in checker.options_by_section():
if section is None:
title = "General options"
else:
title = f"{section.capitalize()} options"
result += get_rst_title(title, "~")
assert isinstance(options, list)
result += f"{get_rst_section(None, options)}\n"
return result
def _get_checkers_documentation(linter: PyLinter, show_options: bool = True) -> str:
"""Get documentation for individual checkers."""
if show_options:
result = _get_global_options_documentation(linter)
else:
result = ""
result += get_rst_title("Pylint checkers' options and switches", "-")
result += """\
Pylint checkers can provide three set of features:
* options that control their execution,
* messages that they can raise,
* reports that they can generate.
Below is a list of all checkers and their features.
"""
by_checker = _get_checkers_infos(linter)
for checker_name in sorted(by_checker):
information = by_checker[checker_name]
checker = information["checker"]
del information["checker"]
result += checker.get_full_documentation(
**information, show_options=show_options
)
return result
def print_full_documentation(
linter: PyLinter, stream: TextIO = sys.stdout, show_options: bool = True
) -> None:
"""Output a full documentation in ReST format."""
print(
_get_checkers_documentation(linter, show_options=show_options)[:-3], file=stream
)
@@ -0,0 +1,295 @@
# Licensed under the GPL: https://www.gnu.org/licenses/old-licenses/gpl-2.0.html
# For details: https://github.com/PyCQA/pylint/blob/main/LICENSE
# Copyright (c) https://github.com/PyCQA/pylint/blob/main/CONTRIBUTORS.txt
from __future__ import annotations
import collections
import sys
import warnings
from collections import defaultdict
from collections.abc import Iterator
from typing import TYPE_CHECKING, Dict
from astroid import nodes
from pylint.constants import (
INCOMPATIBLE_WITH_USELESS_SUPPRESSION,
MSG_STATE_SCOPE_MODULE,
WarningScope,
)
if sys.version_info >= (3, 8):
from typing import Literal
else:
from typing_extensions import Literal
if TYPE_CHECKING:
from pylint.message import MessageDefinition, MessageDefinitionStore
MessageStateDict = Dict[str, Dict[int, bool]]
class FileState:
"""Hold internal state specific to the currently analyzed file."""
def __init__(
self,
modname: str | None = None,
msg_store: MessageDefinitionStore | None = None,
node: nodes.Module | None = None,
*,
is_base_filestate: bool = False,
) -> None:
if modname is None:
warnings.warn(
"FileState needs a string as modname argument. "
"This argument will be required in pylint 3.0",
DeprecationWarning,
)
if msg_store is None:
warnings.warn(
"FileState needs a 'MessageDefinitionStore' as msg_store argument. "
"This argument will be required in pylint 3.0",
DeprecationWarning,
)
self.base_name = modname
self._module_msgs_state: MessageStateDict = {}
self._raw_module_msgs_state: MessageStateDict = {}
self._ignored_msgs: defaultdict[
tuple[str, int], set[int]
] = collections.defaultdict(set)
self._suppression_mapping: dict[tuple[str, int], int] = {}
self._module = node
if node:
self._effective_max_line_number = node.tolineno
else:
self._effective_max_line_number = None
self._msgs_store = msg_store
self._is_base_filestate = is_base_filestate
"""If this FileState is the base state made during initialization of
PyLinter.
"""
def collect_block_lines(
self, msgs_store: MessageDefinitionStore, module_node: nodes.Module
) -> None:
"""Walk the AST to collect block level options line numbers."""
warnings.warn(
"'collect_block_lines' has been deprecated and will be removed in pylint 3.0.",
DeprecationWarning,
)
for msg, lines in self._module_msgs_state.items():
self._raw_module_msgs_state[msg] = lines.copy()
orig_state = self._module_msgs_state.copy()
self._module_msgs_state = {}
self._suppression_mapping = {}
self._effective_max_line_number = module_node.tolineno
for msgid, lines in orig_state.items():
for msgdef in msgs_store.get_message_definitions(msgid):
self._set_state_on_block_lines(msgs_store, module_node, msgdef, lines)
def _set_state_on_block_lines(
self,
msgs_store: MessageDefinitionStore,
node: nodes.NodeNG,
msg: MessageDefinition,
msg_state: dict[int, bool],
) -> None:
"""Recursively walk (depth first) AST to collect block level options
line numbers and set the state correctly.
"""
for child in node.get_children():
self._set_state_on_block_lines(msgs_store, child, msg, msg_state)
# first child line number used to distinguish between disable
# which are the first child of scoped node with those defined later.
# For instance in the code below:
#
# 1. def meth8(self):
# 2. """test late disabling"""
# 3. pylint: disable=not-callable, useless-suppression
# 4. print(self.blip)
# 5. pylint: disable=no-member, useless-suppression
# 6. print(self.bla)
#
# E1102 should be disabled from line 1 to 6 while E1101 from line 5 to 6
#
# this is necessary to disable locally messages applying to class /
# function using their fromlineno
if (
isinstance(node, (nodes.Module, nodes.ClassDef, nodes.FunctionDef))
and node.body
):
firstchildlineno = node.body[0].fromlineno
else:
firstchildlineno = node.tolineno
self._set_message_state_in_block(msg, msg_state, node, firstchildlineno)
def _set_message_state_in_block(
self,
msg: MessageDefinition,
lines: dict[int, bool],
node: nodes.NodeNG,
firstchildlineno: int,
) -> None:
"""Set the state of a message in a block of lines."""
first = node.fromlineno
last = node.tolineno
for lineno, state in list(lines.items()):
original_lineno = lineno
if first > lineno or last < lineno:
continue
# Set state for all lines for this block, if the
# warning is applied to nodes.
if msg.scope == WarningScope.NODE:
if lineno > firstchildlineno:
state = True
first_, last_ = node.block_range(lineno)
# pylint: disable=useless-suppression
# For block nodes first_ is their definition line. For example, we
# set the state of line zero for a module to allow disabling
# invalid-name for the module. For example:
# 1. # pylint: disable=invalid-name
# 2. ...
# OR
# 1. """Module docstring"""
# 2. # pylint: disable=invalid-name
# 3. ...
#
# But if we already visited line 0 we don't need to set its state again
# 1. # pylint: disable=invalid-name
# 2. # pylint: enable=invalid-name
# 3. ...
# The state should come from line 1, not from line 2
# Therefore, if the 'fromlineno' is already in the states we just start
# with the lineno we were originally visiting.
# pylint: enable=useless-suppression
if (
first_ == node.fromlineno
and first_ >= firstchildlineno
and node.fromlineno in self._module_msgs_state.get(msg.msgid, ())
):
first_ = lineno
else:
first_ = lineno
last_ = last
for line in range(first_, last_ + 1):
# Do not override existing entries. This is especially important
# when parsing the states for a scoped node where some line-disables
# have already been parsed.
if (
(
isinstance(node, nodes.Module)
and node.fromlineno <= line < lineno
)
or (
not isinstance(node, nodes.Module)
and node.fromlineno < line < lineno
)
) and line in self._module_msgs_state.get(msg.msgid, ()):
continue
if line in lines: # state change in the same block
state = lines[line]
original_lineno = line
self._set_message_state_on_line(msg, line, state, original_lineno)
del lines[lineno]
def _set_message_state_on_line(
self,
msg: MessageDefinition,
line: int,
state: bool,
original_lineno: int,
) -> None:
"""Set the state of a message on a line."""
# Update suppression mapping
if not state:
self._suppression_mapping[(msg.msgid, line)] = original_lineno
else:
self._suppression_mapping.pop((msg.msgid, line), None)
# Update message state for respective line
try:
self._module_msgs_state[msg.msgid][line] = state
except KeyError:
self._module_msgs_state[msg.msgid] = {line: state}
def set_msg_status(
self,
msg: MessageDefinition,
line: int,
status: bool,
scope: str = "package",
) -> None:
"""Set status (enabled/disable) for a given message at a given line."""
assert line > 0
assert self._module
# TODO: 3.0: Remove unnecessary assertion
assert self._msgs_store
if scope != "line":
# Expand the status to cover all relevant block lines
self._set_state_on_block_lines(
self._msgs_store, self._module, msg, {line: status}
)
else:
self._set_message_state_on_line(msg, line, status, line)
# Store the raw value
try:
self._raw_module_msgs_state[msg.msgid][line] = status
except KeyError:
self._raw_module_msgs_state[msg.msgid] = {line: status}
def handle_ignored_message(
self, state_scope: Literal[0, 1, 2] | None, msgid: str, line: int | None
) -> None:
"""Report an ignored message.
state_scope is either MSG_STATE_SCOPE_MODULE or MSG_STATE_SCOPE_CONFIG,
depending on whether the message was disabled locally in the module,
or globally.
"""
if state_scope == MSG_STATE_SCOPE_MODULE:
assert isinstance(line, int) # should always be int inside module scope
try:
orig_line = self._suppression_mapping[(msgid, line)]
self._ignored_msgs[(msgid, orig_line)].add(line)
except KeyError:
pass
def iter_spurious_suppression_messages(
self,
msgs_store: MessageDefinitionStore,
) -> Iterator[
tuple[
Literal["useless-suppression", "suppressed-message"],
int,
tuple[str] | tuple[str, int],
]
]:
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
and warning not in INCOMPATIBLE_WITH_USELESS_SUPPRESSION
):
yield "useless-suppression", line, (
msgs_store.get_msg_display_string(warning),
)
# don't use iteritems here, _ignored_msgs may be modified by add_message
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) -> int | None:
return self._effective_max_line_number
@@ -0,0 +1,387 @@
# Licensed under the GPL: https://www.gnu.org/licenses/old-licenses/gpl-2.0.html
# For details: https://github.com/PyCQA/pylint/blob/main/LICENSE
# Copyright (c) https://github.com/PyCQA/pylint/blob/main/CONTRIBUTORS.txt
from __future__ import annotations
import sys
from typing import cast
from pylint.typing import MessageTypesFullName
if sys.version_info >= (3, 8):
from typing import Literal, TypedDict
else:
from typing_extensions import Literal, TypedDict
class BadNames(TypedDict):
"""TypedDict to store counts of node types with bad names."""
argument: int
attr: int
klass: int
class_attribute: int
class_const: int
const: int
inlinevar: int
function: int
method: int
module: int
variable: int
typevar: int
class CodeTypeCount(TypedDict):
"""TypedDict to store counts of lines of code types."""
code: int
comment: int
docstring: int
empty: int
total: int
class DuplicatedLines(TypedDict):
"""TypedDict to store counts of lines of duplicated code."""
nb_duplicated_lines: int
percent_duplicated_lines: float
class NodeCount(TypedDict):
"""TypedDict to store counts of different types of nodes."""
function: int
klass: int
method: int
module: int
class UndocumentedNodes(TypedDict):
"""TypedDict to store counts of undocumented node types."""
function: int
klass: int
method: int
module: int
class ModuleStats(TypedDict):
"""TypedDict to store counts of types of messages and statements."""
convention: int
error: int
fatal: int
info: int
refactor: int
statement: int
warning: int
# pylint: disable-next=too-many-instance-attributes
class LinterStats:
"""Class used to linter stats."""
def __init__(
self,
bad_names: BadNames | None = None,
by_module: dict[str, ModuleStats] | None = None,
by_msg: dict[str, int] | None = None,
code_type_count: CodeTypeCount | None = None,
dependencies: dict[str, set[str]] | None = None,
duplicated_lines: DuplicatedLines | None = None,
node_count: NodeCount | None = None,
undocumented: UndocumentedNodes | None = None,
) -> None:
self.bad_names = bad_names or BadNames(
argument=0,
attr=0,
klass=0,
class_attribute=0,
class_const=0,
const=0,
inlinevar=0,
function=0,
method=0,
module=0,
variable=0,
typevar=0,
)
self.by_module: dict[str, ModuleStats] = by_module or {}
self.by_msg: dict[str, int] = by_msg or {}
self.code_type_count = code_type_count or CodeTypeCount(
code=0, comment=0, docstring=0, empty=0, total=0
)
self.dependencies: dict[str, set[str]] = dependencies or {}
self.duplicated_lines = duplicated_lines or DuplicatedLines(
nb_duplicated_lines=0, percent_duplicated_lines=0.0
)
self.node_count = node_count or NodeCount(
function=0, klass=0, method=0, module=0
)
self.undocumented = undocumented or UndocumentedNodes(
function=0, klass=0, method=0, module=0
)
self.convention = 0
self.error = 0
self.fatal = 0
self.info = 0
self.refactor = 0
self.statement = 0
self.warning = 0
self.global_note = 0
self.nb_duplicated_lines = 0
self.percent_duplicated_lines = 0.0
def __repr__(self) -> str:
return str(self)
def __str__(self) -> str:
return f"""{self.bad_names}
{sorted(self.by_module.items())}
{sorted(self.by_msg.items())}
{self.code_type_count}
{sorted(self.dependencies.items())}
{self.duplicated_lines}
{self.undocumented}
{self.convention}
{self.error}
{self.fatal}
{self.info}
{self.refactor}
{self.statement}
{self.warning}
{self.global_note}
{self.nb_duplicated_lines}
{self.percent_duplicated_lines}"""
def init_single_module(self, module_name: str) -> None:
"""Use through PyLinter.set_current_module so PyLinter.current_name is
consistent.
"""
self.by_module[module_name] = ModuleStats(
convention=0, error=0, fatal=0, info=0, refactor=0, statement=0, warning=0
)
def get_bad_names(
self,
node_name: Literal[
"argument",
"attr",
"class",
"class_attribute",
"class_const",
"const",
"inlinevar",
"function",
"method",
"module",
"variable",
"typevar",
],
) -> int:
"""Get a bad names node count."""
if node_name == "class":
return self.bad_names.get("klass", 0)
return self.bad_names.get(node_name, 0)
def increase_bad_name(self, node_name: str, increase: int) -> None:
"""Increase a bad names node count."""
if node_name not in {
"argument",
"attr",
"class",
"class_attribute",
"class_const",
"const",
"inlinevar",
"function",
"method",
"module",
"variable",
"typevar",
}:
raise ValueError("Node type not part of the bad_names stat")
node_name = cast(
Literal[
"argument",
"attr",
"class",
"class_attribute",
"class_const",
"const",
"inlinevar",
"function",
"method",
"module",
"variable",
"typevar",
],
node_name,
)
if node_name == "class":
self.bad_names["klass"] += increase
else:
self.bad_names[node_name] += increase
def reset_bad_names(self) -> None:
"""Resets the bad_names attribute."""
self.bad_names = BadNames(
argument=0,
attr=0,
klass=0,
class_attribute=0,
class_const=0,
const=0,
inlinevar=0,
function=0,
method=0,
module=0,
variable=0,
typevar=0,
)
def get_code_count(
self, type_name: Literal["code", "comment", "docstring", "empty", "total"]
) -> int:
"""Get a code type count."""
return self.code_type_count.get(type_name, 0)
def reset_code_count(self) -> None:
"""Resets the code_type_count attribute."""
self.code_type_count = CodeTypeCount(
code=0, comment=0, docstring=0, empty=0, total=0
)
def reset_duplicated_lines(self) -> None:
"""Resets the duplicated_lines attribute."""
self.duplicated_lines = DuplicatedLines(
nb_duplicated_lines=0, percent_duplicated_lines=0.0
)
def get_node_count(
self, node_name: Literal["function", "class", "method", "module"]
) -> int:
"""Get a node count while handling some extra conditions."""
if node_name == "class":
return self.node_count.get("klass", 0)
return self.node_count.get(node_name, 0)
def reset_node_count(self) -> None:
"""Resets the node count attribute."""
self.node_count = NodeCount(function=0, klass=0, method=0, module=0)
def get_undocumented(
self, node_name: Literal["function", "class", "method", "module"]
) -> float:
"""Get a undocumented node count."""
if node_name == "class":
return self.undocumented["klass"]
return self.undocumented[node_name]
def reset_undocumented(self) -> None:
"""Resets the undocumented attribute."""
self.undocumented = UndocumentedNodes(function=0, klass=0, method=0, module=0)
def get_global_message_count(self, type_name: str) -> int:
"""Get a global message count."""
return getattr(self, type_name, 0)
def get_module_message_count(self, modname: str, type_name: str) -> int:
"""Get a module message count."""
return getattr(self.by_module[modname], type_name, 0)
def increase_single_message_count(self, type_name: str, increase: int) -> None:
"""Increase the message type count of an individual message type."""
setattr(self, type_name, getattr(self, type_name) + increase)
def increase_single_module_message_count(
self, modname: str, type_name: MessageTypesFullName, increase: int
) -> None:
"""Increase the message type count of an individual message type of a
module.
"""
self.by_module[modname][type_name] += increase
def reset_message_count(self) -> None:
"""Resets the message type count of the stats object."""
self.convention = 0
self.error = 0
self.fatal = 0
self.info = 0
self.refactor = 0
self.warning = 0
def merge_stats(stats: list[LinterStats]) -> LinterStats:
"""Used to merge multiple stats objects into a new one when pylint is run in
parallel mode.
"""
merged = LinterStats()
for stat in stats:
merged.bad_names["argument"] += stat.bad_names["argument"]
merged.bad_names["attr"] += stat.bad_names["attr"]
merged.bad_names["klass"] += stat.bad_names["klass"]
merged.bad_names["class_attribute"] += stat.bad_names["class_attribute"]
merged.bad_names["class_const"] += stat.bad_names["class_const"]
merged.bad_names["const"] += stat.bad_names["const"]
merged.bad_names["inlinevar"] += stat.bad_names["inlinevar"]
merged.bad_names["function"] += stat.bad_names["function"]
merged.bad_names["method"] += stat.bad_names["method"]
merged.bad_names["module"] += stat.bad_names["module"]
merged.bad_names["variable"] += stat.bad_names["variable"]
merged.bad_names["typevar"] += stat.bad_names["typevar"]
for mod_key, mod_value in stat.by_module.items():
merged.by_module[mod_key] = mod_value
for msg_key, msg_value in stat.by_msg.items():
try:
merged.by_msg[msg_key] += msg_value
except KeyError:
merged.by_msg[msg_key] = msg_value
merged.code_type_count["code"] += stat.code_type_count["code"]
merged.code_type_count["comment"] += stat.code_type_count["comment"]
merged.code_type_count["docstring"] += stat.code_type_count["docstring"]
merged.code_type_count["empty"] += stat.code_type_count["empty"]
merged.code_type_count["total"] += stat.code_type_count["total"]
for dep_key, dep_value in stat.dependencies.items():
try:
merged.dependencies[dep_key].update(dep_value)
except KeyError:
merged.dependencies[dep_key] = dep_value
merged.duplicated_lines["nb_duplicated_lines"] += stat.duplicated_lines[
"nb_duplicated_lines"
]
merged.duplicated_lines["percent_duplicated_lines"] += stat.duplicated_lines[
"percent_duplicated_lines"
]
merged.node_count["function"] += stat.node_count["function"]
merged.node_count["klass"] += stat.node_count["klass"]
merged.node_count["method"] += stat.node_count["method"]
merged.node_count["module"] += stat.node_count["module"]
merged.undocumented["function"] += stat.undocumented["function"]
merged.undocumented["klass"] += stat.undocumented["klass"]
merged.undocumented["method"] += stat.undocumented["method"]
merged.undocumented["module"] += stat.undocumented["module"]
merged.convention += stat.convention
merged.error += stat.error
merged.fatal += stat.fatal
merged.info += stat.info
merged.refactor += stat.refactor
merged.statement += stat.statement
merged.warning += stat.warning
merged.global_note += stat.global_note
return merged
@@ -0,0 +1,133 @@
# Licensed under the GPL: https://www.gnu.org/licenses/old-licenses/gpl-2.0.html
# For details: https://github.com/PyCQA/pylint/blob/main/LICENSE
# Copyright (c) https://github.com/PyCQA/pylint/blob/main/CONTRIBUTORS.txt
from __future__ import annotations
import re
from collections import namedtuple
from collections.abc import Generator
# Allow stopping after the first semicolon/hash encountered,
# so that an option can be continued with the reasons
# why it is active or disabled.
OPTION_RGX = r"""
(?:^\s*\#.*|\s*| # Comment line, or whitespaces,
\s*\#.*(?=\#.*?\bpylint:)) # or a beginning of an inline comment
# followed by "pylint:" pragma
(\# # Beginning of comment
.*? # Anything (as little as possible)
\bpylint: # pylint word and column
\s* # Any number of whitespaces
([^;#\n]+)) # Anything except semicolon or hash or
# newline (it is the second matched group)
# and end of the first matched group
[;#]{0,1} # From 0 to 1 repetition of semicolon or hash
"""
OPTION_PO = re.compile(OPTION_RGX, re.VERBOSE)
PragmaRepresenter = namedtuple("PragmaRepresenter", "action messages")
ATOMIC_KEYWORDS = frozenset(("disable-all", "skip-file"))
MESSAGE_KEYWORDS = frozenset(
("disable-next", "disable-msg", "enable-msg", "disable", "enable")
)
# sorted is necessary because sets are unordered collections and ALL_KEYWORDS
# string should not vary between executions
# reverse is necessary in order to have the longest keywords first, so that, for example,
# 'disable' string should not be matched instead of 'disable-all'
ALL_KEYWORDS = "|".join(
sorted(ATOMIC_KEYWORDS | MESSAGE_KEYWORDS, key=len, reverse=True)
)
TOKEN_SPECIFICATION = [
("KEYWORD", rf"\b({ALL_KEYWORDS:s})\b"),
("MESSAGE_STRING", r"[0-9A-Za-z\-\_]{2,}"), # Identifiers
("ASSIGN", r"="), # Assignment operator
("MESSAGE_NUMBER", r"[CREIWF]{1}\d*"),
]
TOK_REGEX = "|".join(
f"(?P<{token_name:s}>{token_rgx:s})"
for token_name, token_rgx in TOKEN_SPECIFICATION
)
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
)
return PragmaRepresenter(action, messages)
class PragmaParserError(Exception):
"""A class for exceptions thrown by pragma_parser module."""
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.
"""
self.message = message
self.token = token
super().__init__(self.message)
class UnRecognizedOptionError(PragmaParserError):
"""Thrown in case the of a valid but unrecognized option."""
class InvalidPragmaError(PragmaParserError):
"""Thrown in case the pragma is invalid."""
def parse_pragma(pylint_pragma: str) -> Generator[PragmaRepresenter, None, None]:
action: str | None = None
messages: list[str] = []
assignment_required = False
previous_token = ""
for mo in re.finditer(TOK_REGEX, pylint_pragma):
kind = mo.lastgroup
value = mo.group()
if kind == "ASSIGN":
if not assignment_required:
if action:
# A keyword has been found previously but doesn't support assignment
raise UnRecognizedOptionError(
"The keyword doesn't support assignment", action
)
if previous_token:
# Something found previously but not a known keyword
raise UnRecognizedOptionError(
"The keyword is unknown", previous_token
)
# Nothing at all detected before this assignment
raise InvalidPragmaError("Missing keyword before assignment", "")
assignment_required = False
elif assignment_required:
raise InvalidPragmaError(
"The = sign is missing after the keyword", action or ""
)
elif kind == "KEYWORD":
if action:
yield emit_pragma_representer(action, messages)
action = value
messages = []
assignment_required = action in MESSAGE_KEYWORDS
elif kind in {"MESSAGE_STRING", "MESSAGE_NUMBER"}:
messages.append(value)
assignment_required = False
else:
raise RuntimeError("Token not recognized")
previous_token = value
if action:
yield emit_pragma_representer(action, messages)
else:
raise UnRecognizedOptionError("The keyword is unknown", previous_token)
@@ -0,0 +1,435 @@
# Licensed under the GPL: https://www.gnu.org/licenses/old-licenses/gpl-2.0.html
# For details: https://github.com/PyCQA/pylint/blob/main/LICENSE
# Copyright (c) https://github.com/PyCQA/pylint/blob/main/CONTRIBUTORS.txt
from __future__ import annotations
try:
import isort.api
HAS_ISORT_5 = True
except ImportError: # isort < 5
import isort
HAS_ISORT_5 = False
import argparse
import codecs
import os
import re
import sys
import textwrap
import tokenize
import warnings
from collections.abc import Sequence
from io import BufferedReader, BytesIO
from typing import (
TYPE_CHECKING,
Any,
List,
Pattern,
TextIO,
Tuple,
TypeVar,
Union,
overload,
)
from astroid import Module, modutils, nodes
from pylint.constants import PY_EXTS
from pylint.typing import OptionDict
if sys.version_info >= (3, 8):
from typing import Literal
else:
from typing_extensions import Literal
if TYPE_CHECKING:
from pylint.checkers.base_checker import BaseChecker
from pylint.lint import PyLinter
DEFAULT_LINE_LENGTH = 79
# These are types used to overload get_global_option() and refer to the options type
GLOBAL_OPTION_BOOL = Literal[
"suggestion-mode",
"analyse-fallback-blocks",
"allow-global-unused-variables",
]
GLOBAL_OPTION_INT = Literal["max-line-length", "docstring-min-length"]
GLOBAL_OPTION_LIST = Literal["ignored-modules"]
GLOBAL_OPTION_PATTERN = Literal[
"no-docstring-rgx",
"dummy-variables-rgx",
"ignored-argument-names",
"mixin-class-rgx",
]
GLOBAL_OPTION_PATTERN_LIST = Literal["exclude-too-few-public-methods", "ignore-paths"]
GLOBAL_OPTION_TUPLE_INT = Literal["py-version"]
GLOBAL_OPTION_NAMES = Union[
GLOBAL_OPTION_BOOL,
GLOBAL_OPTION_INT,
GLOBAL_OPTION_LIST,
GLOBAL_OPTION_PATTERN,
GLOBAL_OPTION_PATTERN_LIST,
GLOBAL_OPTION_TUPLE_INT,
]
T_GlobalOptionReturnTypes = TypeVar(
"T_GlobalOptionReturnTypes",
bool,
int,
List[str],
Pattern[str],
List[Pattern[str]],
Tuple[int, ...],
)
def normalize_text(
text: str, line_len: int = DEFAULT_LINE_LENGTH, indent: str = ""
) -> str:
"""Wrap the text on the given line length."""
return "\n".join(
textwrap.wrap(
text, width=line_len, initial_indent=indent, subsequent_indent=indent
)
)
CMPS = ["=", "-", "+"]
# py3k has no more cmp builtin
def cmp(a: int | float, b: int | float) -> int:
return (a > b) - (a < b)
def diff_string(old: int | float, new: int | float) -> str:
"""Given an old and new int value, return a string representing the
difference.
"""
diff = abs(old - new)
diff_str = f"{CMPS[cmp(old, new)]}{diff and f'{diff:.2f}' or ''}"
return diff_str
def get_module_and_frameid(node: nodes.NodeNG) -> tuple[str, str]:
"""Return the module name and the frame id in the module."""
frame = node.frame(future=True)
module, obj = "", []
while frame:
if isinstance(frame, Module):
module = frame.name
else:
obj.append(getattr(frame, "name", "<lambda>"))
try:
frame = frame.parent.frame(future=True)
except AttributeError:
break
obj.reverse()
return module, ".".join(obj)
def get_rst_title(title: str, character: str) -> str:
"""Permit to get a title formatted as ReStructuredText test (underlined with a
chosen character).
"""
return f"{title}\n{character * len(title)}\n"
def get_rst_section(
section: str | None,
options: list[tuple[str, OptionDict, Any]],
doc: str | None = None,
) -> str:
"""Format an option's section using as a ReStructuredText formatted output."""
result = ""
if section:
result += get_rst_title(section, "'")
if doc:
formatted_doc = normalize_text(doc)
result += f"{formatted_doc}\n\n"
for optname, optdict, value in options:
help_opt = optdict.get("help")
result += f":{optname}:\n"
if help_opt:
assert isinstance(help_opt, str)
formatted_help = normalize_text(help_opt, indent=" ")
result += f"{formatted_help}\n"
if value and optname != "py-version":
value = str(_format_option_value(optdict, value))
result += f"\n Default: ``{value.replace('`` ', '```` ``')}``\n"
return result
def decoding_stream(
stream: BufferedReader | BytesIO,
encoding: str,
errors: Literal["strict"] = "strict",
) -> codecs.StreamReader:
try:
reader_cls = codecs.getreader(encoding or sys.getdefaultencoding())
except LookupError:
reader_cls = codecs.getreader(sys.getdefaultencoding())
return reader_cls(stream, errors)
def tokenize_module(node: nodes.Module) -> list[tokenize.TokenInfo]:
with node.stream() as stream:
readline = stream.readline
return list(tokenize.tokenize(readline))
def register_plugins(linter: PyLinter, directory: str) -> None:
"""Load all module and package in the given directory, looking for a
'register' function in each one, used to register pylint checkers.
"""
imported = {}
for filename in os.listdir(directory):
base, extension = os.path.splitext(filename)
if base in imported or base == "__pycache__":
continue
if (
extension in PY_EXTS
and base != "__init__"
or (
not extension
and os.path.isdir(os.path.join(directory, base))
and not filename.startswith(".")
)
):
try:
module = modutils.load_module_from_file(
os.path.join(directory, filename)
)
except ValueError:
# empty module name (usually Emacs auto-save files)
continue
except ImportError as exc:
print(f"Problem importing module {filename}: {exc}", file=sys.stderr)
else:
if hasattr(module, "register"):
module.register(linter)
imported[base] = 1
@overload
def get_global_option(
checker: BaseChecker, option: GLOBAL_OPTION_BOOL, default: bool | None = ...
) -> bool:
...
@overload
def get_global_option(
checker: BaseChecker, option: GLOBAL_OPTION_INT, default: int | None = ...
) -> int:
...
@overload
def get_global_option(
checker: BaseChecker,
option: GLOBAL_OPTION_LIST,
default: list[str] | None = ...,
) -> list[str]:
...
@overload
def get_global_option(
checker: BaseChecker,
option: GLOBAL_OPTION_PATTERN,
default: Pattern[str] | None = ...,
) -> Pattern[str]:
...
@overload
def get_global_option(
checker: BaseChecker,
option: GLOBAL_OPTION_PATTERN_LIST,
default: list[Pattern[str]] | None = ...,
) -> list[Pattern[str]]:
...
@overload
def get_global_option(
checker: BaseChecker,
option: GLOBAL_OPTION_TUPLE_INT,
default: tuple[int, ...] | None = ...,
) -> tuple[int, ...]:
...
def get_global_option(
checker: BaseChecker,
option: GLOBAL_OPTION_NAMES,
default: T_GlobalOptionReturnTypes | None = None, # pylint: disable=unused-argument
) -> T_GlobalOptionReturnTypes | None | Any:
"""DEPRECATED: Retrieve an option defined by the given *checker* or
by all known option providers.
It will look in the list of all options providers
until the given *option* will be found.
If the option wasn't found, the *default* value will be returned.
"""
warnings.warn(
"get_global_option has been deprecated. You can use "
"checker.linter.config to get all global options instead.",
DeprecationWarning,
)
return getattr(checker.linter.config, option.replace("-", "_"))
def _splitstrip(string: str, sep: str = ",") -> list[str]:
"""Return a list of stripped string by splitting the string given as
argument on `sep` (',' by default), empty strings are discarded.
>>> _splitstrip('a, b, c , 4,,')
['a', 'b', 'c', '4']
>>> _splitstrip('a')
['a']
>>> _splitstrip('a,\nb,\nc,')
['a', 'b', 'c']
:type string: str or unicode
:param string: a csv line
:type sep: str or unicode
:param sep: field separator, default to the comma (',')
:rtype: str or unicode
:return: the unquoted string (or the input string if it wasn't quoted)
"""
return [word.strip() for word in string.split(sep) if word.strip()]
def _unquote(string: str) -> str:
"""Remove optional quotes (simple or double) from the string.
:param string: an optionally quoted string
:return: the unquoted string (or the input string if it wasn't quoted)
"""
if not string:
return string
if string[0] in "\"'":
string = string[1:]
if string[-1] in "\"'":
string = string[:-1]
return string
def _check_csv(value: list[str] | tuple[str] | str) -> Sequence[str]:
if isinstance(value, (list, tuple)):
return value
return _splitstrip(value)
def _comment(string: str) -> str:
"""Return string as a comment."""
lines = [line.strip() for line in string.splitlines()]
sep = "\n"
return "# " + f"{sep}# ".join(lines)
def _format_option_value(optdict: OptionDict, value: Any) -> str:
"""Return the user input's value from a 'compiled' value.
TODO: 3.0: Remove deprecated function
"""
if optdict.get("type", None) == "py_version":
value = ".".join(str(item) for item in value)
elif isinstance(value, (list, tuple)):
value = ",".join(_format_option_value(optdict, item) for item in value)
elif isinstance(value, dict):
value = ",".join(f"{k}:{v}" for k, v in value.items())
elif hasattr(value, "match"): # optdict.get('type') == 'regexp'
# compiled regexp
value = value.pattern
elif optdict.get("type") == "yn":
value = "yes" if value else "no"
elif isinstance(value, str) and value.isspace():
value = f"'{value}'"
return str(value)
def format_section(
stream: TextIO,
section: str,
options: list[tuple[str, OptionDict, Any]],
doc: str | None = None,
) -> None:
"""Format an option's section using the INI format."""
warnings.warn(
"format_section has been deprecated. It will be removed in pylint 3.0.",
DeprecationWarning,
)
if doc:
print(_comment(doc), file=stream)
print(f"[{section}]", file=stream)
with warnings.catch_warnings():
warnings.filterwarnings("ignore", category=DeprecationWarning)
_ini_format(stream, options)
def _ini_format(stream: TextIO, options: list[tuple[str, OptionDict, Any]]) -> None:
"""Format options using the INI format."""
warnings.warn(
"_ini_format has been deprecated. It will be removed in pylint 3.0.",
DeprecationWarning,
)
for optname, optdict, value in options:
# Skip deprecated option
if "kwargs" in optdict:
assert isinstance(optdict["kwargs"], dict)
if "new_names" in optdict["kwargs"]:
continue
value = _format_option_value(optdict, value)
help_opt = optdict.get("help")
if help_opt:
assert isinstance(help_opt, str)
help_opt = normalize_text(help_opt, indent="# ")
print(file=stream)
print(help_opt, file=stream)
else:
print(file=stream)
if value in {"None", "False"}:
print(f"#{optname}=", file=stream)
else:
value = str(value).strip()
if re.match(r"^([\w-]+,)+[\w-]+$", str(value)):
separator = "\n " + " " * len(optname)
value = separator.join(x + "," for x in str(value).split(","))
# remove trailing ',' from last element of the list
value = value[:-1]
print(f"{optname}={value}", file=stream)
class IsortDriver:
"""A wrapper around isort API that changed between versions 4 and 5."""
def __init__(self, config: argparse.Namespace) -> None:
if HAS_ISORT_5:
self.isort5_config = isort.api.Config(
# There is no typo here. EXTRA_standard_library is
# what most users want. The option has been named
# KNOWN_standard_library for ages in pylint, and we
# don't want to break compatibility.
extra_standard_library=config.known_standard_library,
known_third_party=config.known_third_party,
)
else:
# pylint: disable-next=no-member
self.isort4_obj = isort.SortImports(
file_contents="",
known_standard_library=config.known_standard_library,
known_third_party=config.known_third_party,
)
def place_module(self, package: str) -> str:
if HAS_ISORT_5:
return isort.api.place_module(package, self.isort5_config)
return self.isort4_obj.place_module(package)