1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
|
# 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 argparse
import collections
import contextlib
import functools
import os
import sys
import tokenize
import traceback
import warnings
from collections import defaultdict
from collections.abc import Callable, Iterator, Sequence
from io import TextIOWrapper
from pathlib import Path
from re import Pattern
from typing import Any
import astroid
from astroid import nodes
from pylint import checkers, exceptions, interfaces, reporters
from pylint.checkers.base_checker import BaseChecker
from pylint.config.arguments_manager import _ArgumentsManager
from pylint.constants import (
MAIN_CHECKER_NAME,
MSG_TYPES,
MSG_TYPES_STATUS,
WarningScope,
)
from pylint.interfaces import HIGH
from pylint.lint.base_options import _make_linter_options
from pylint.lint.caching import load_results, save_results
from pylint.lint.expand_modules import _is_ignored_file, expand_modules
from pylint.lint.message_state_handler import _MessageStateHandler
from pylint.lint.parallel import check_parallel
from pylint.lint.report_functions import (
report_messages_by_module_stats,
report_messages_stats,
report_total_messages_stats,
)
from pylint.lint.utils import (
_is_relative_to,
fix_import_path,
get_fatal_error_message,
prepare_crash_report,
)
from pylint.message import Message, MessageDefinition, MessageDefinitionStore
from pylint.reporters.base_reporter import BaseReporter
from pylint.reporters.text import TextReporter
from pylint.reporters.ureports import nodes as report_nodes
from pylint.typing import (
DirectoryNamespaceDict,
FileItem,
ManagedMessage,
MessageDefinitionTuple,
MessageLocationTuple,
ModuleDescriptionDict,
Options,
)
from pylint.utils import ASTWalker, FileState, LinterStats, utils
if sys.version_info >= (3, 8):
from typing import Protocol
else:
from typing_extensions import Protocol
MANAGER = astroid.MANAGER
class GetAstProtocol(Protocol):
def __call__(
self, filepath: str, modname: str, data: str | None = None
) -> nodes.Module:
...
def _read_stdin() -> str:
# See https://github.com/python/typeshed/pull/5623 for rationale behind assertion
assert isinstance(sys.stdin, TextIOWrapper)
sys.stdin = TextIOWrapper(sys.stdin.detach(), encoding="utf-8")
return sys.stdin.read()
def _load_reporter_by_class(reporter_class: str) -> type[BaseReporter]:
qname = reporter_class
module_part = astroid.modutils.get_module_part(qname)
module = astroid.modutils.load_module_from_name(module_part)
class_name = qname.split(".")[-1]
klass = getattr(module, class_name)
assert issubclass(klass, BaseReporter), f"{klass} is not a BaseReporter"
return klass
# Python Linter class #########################################################
# pylint: disable-next=consider-using-namedtuple-or-dataclass
MSGS: dict[str, MessageDefinitionTuple] = {
"F0001": (
"%s",
"fatal",
"Used when an error occurred preventing the analysis of a \
module (unable to find it for instance).",
{"scope": WarningScope.LINE},
),
"F0002": (
"%s: %s",
"astroid-error",
"Used when an unexpected error occurred while building the "
"Astroid representation. This is usually accompanied by a "
"traceback. Please report such errors !",
{"scope": WarningScope.LINE},
),
"F0010": (
"error while code parsing: %s",
"parse-error",
"Used when an exception occurred while building the Astroid "
"representation which could be handled by astroid.",
{"scope": WarningScope.LINE},
),
"F0011": (
"error while parsing the configuration: %s",
"config-parse-error",
"Used when an exception occurred while parsing a pylint configuration file.",
{"scope": WarningScope.LINE},
),
"I0001": (
"Unable to run raw checkers on built-in module %s",
"raw-checker-failed",
"Used to inform that a built-in module has not been checked "
"using the raw checkers.",
{"scope": WarningScope.LINE},
),
"I0010": (
"Unable to consider inline option %r",
"bad-inline-option",
"Used when an inline option is either badly formatted or can't "
"be used inside modules.",
{"scope": WarningScope.LINE},
),
"I0011": (
"Locally disabling %s (%s)",
"locally-disabled",
"Used when an inline option disables a message or a messages category.",
{"scope": WarningScope.LINE},
),
"I0013": (
"Ignoring entire file",
"file-ignored",
"Used to inform that the file will not be checked",
{"scope": WarningScope.LINE},
),
"I0020": (
"Suppressed %s (from line %d)",
"suppressed-message",
"A message was triggered on a line, but suppressed explicitly "
"by a disable= comment in the file. This message is not "
"generated for messages that are ignored due to configuration "
"settings.",
{"scope": WarningScope.LINE},
),
"I0021": (
"Useless suppression of %s",
"useless-suppression",
"Reported when a message is explicitly disabled for a line or "
"a block of code, but never triggered.",
{"scope": WarningScope.LINE},
),
"I0022": (
'Pragma "%s" is deprecated, use "%s" instead',
"deprecated-pragma",
"Some inline pylint options have been renamed or reworked, "
"only the most recent form should be used. "
"NOTE:skip-all is only available with pylint >= 0.26",
{
"old_names": [("I0014", "deprecated-disable-all")],
"scope": WarningScope.LINE,
},
),
"E0001": (
"%s",
"syntax-error",
"Used when a syntax error is raised for a module.",
{"scope": WarningScope.LINE},
),
"E0011": (
"Unrecognized file option %r",
"unrecognized-inline-option",
"Used when an unknown inline option is encountered.",
{"scope": WarningScope.LINE},
),
"W0012": (
"Unknown option value for '%s', expected a valid pylint message and got '%s'",
"unknown-option-value",
"Used when an unknown value is encountered for an option.",
{
"scope": WarningScope.LINE,
"old_names": [("E0012", "bad-option-value")],
},
),
"R0022": (
"Useless option value for '%s', %s",
"useless-option-value",
"Used when a value for an option that is now deleted from pylint"
" is encountered.",
{
"scope": WarningScope.LINE,
"old_names": [("E0012", "bad-option-value")],
},
),
"E0013": (
"Plugin '%s' is impossible to load, is it installed ? ('%s')",
"bad-plugin-value",
"Used when a bad value is used in 'load-plugins'.",
{"scope": WarningScope.LINE},
),
"E0014": (
"Out-of-place setting encountered in top level configuration-section '%s' : '%s'",
"bad-configuration-section",
"Used when we detect a setting in the top level of a toml configuration that shouldn't be there.",
{"scope": WarningScope.LINE},
),
"E0015": (
"Unrecognized option found: %s",
"unrecognized-option",
"Used when we detect an option that we do not recognize.",
{"scope": WarningScope.LINE},
),
}
# pylint: disable=too-many-instance-attributes,too-many-public-methods
class PyLinter(
_ArgumentsManager,
_MessageStateHandler,
reporters.ReportsHandlerMixIn,
checkers.BaseChecker,
):
"""Lint Python modules using external checkers.
This is the main checker controlling the other ones and the reports
generation. It is itself both a raw checker and an astroid checker in order
to:
* handle message activation / deactivation at the module level
* handle some basic but necessary stats' data (number of classes, methods...)
IDE plugin developers: you may have to call
`astroid.MANAGER.clear_cache()` across runs if you want
to ensure the latest code version is actually checked.
This class needs to support pickling for parallel linting to work. The exception
is reporter member; see check_parallel function for more details.
"""
name = MAIN_CHECKER_NAME
msgs = MSGS
# Will be used like this : datetime.now().strftime(crash_file_path)
crash_file_path: str = "pylint-crash-%Y-%m-%d-%H-%M-%S.txt"
option_groups_descs = {
"Messages control": "Options controlling analysis messages",
"Reports": "Options related to output formatting and reporting",
}
def __init__(
self,
options: Options = (),
reporter: reporters.BaseReporter | reporters.MultiReporter | None = None,
option_groups: tuple[tuple[str, str], ...] = (),
# TODO: Deprecate passing the pylintrc parameter
pylintrc: str | None = None, # pylint: disable=unused-argument
) -> None:
_ArgumentsManager.__init__(self, prog="pylint")
_MessageStateHandler.__init__(self, self)
# Some stuff has to be done before initialization of other ancestors...
# messages store / checkers / reporter / astroid manager
# Attributes for reporters
self.reporter: reporters.BaseReporter | reporters.MultiReporter
if reporter:
self.set_reporter(reporter)
else:
self.set_reporter(TextReporter())
self._reporters: dict[str, type[reporters.BaseReporter]] = {}
"""Dictionary of possible but non-initialized reporters."""
# Attributes for checkers and plugins
self._checkers: defaultdict[
str, list[checkers.BaseChecker]
] = collections.defaultdict(list)
"""Dictionary of registered and initialized checkers."""
self._dynamic_plugins: set[str] = set()
"""Set of loaded plugin names."""
# Attributes related to registering messages and their handling
self.msgs_store = MessageDefinitionStore()
self.msg_status = 0
self._by_id_managed_msgs: list[ManagedMessage] = []
# Attributes related to visiting files
self.file_state = FileState("", self.msgs_store, is_base_filestate=True)
self.current_name: str | None = None
self.current_file: str | None = None
self._ignore_file = False
self._ignore_paths: list[Pattern[str]] = []
# Attributes related to stats
self.stats = LinterStats()
# Attributes related to (command-line) options and their parsing
self.options: Options = options + _make_linter_options(self)
for opt_group in option_groups:
self.option_groups_descs[opt_group[0]] = opt_group[1]
self._option_groups: tuple[tuple[str, str], ...] = option_groups + (
("Messages control", "Options controlling analysis messages"),
("Reports", "Options related to output formatting and reporting"),
)
self.fail_on_symbols: list[str] = []
"""List of message symbols on which pylint should fail, set by --fail-on."""
self._error_mode = False
reporters.ReportsHandlerMixIn.__init__(self)
checkers.BaseChecker.__init__(self, self)
# provided reports
self.reports = (
("RP0001", "Messages by category", report_total_messages_stats),
(
"RP0002",
"% errors / warnings by module",
report_messages_by_module_stats,
),
("RP0003", "Messages", report_messages_stats),
)
self.register_checker(self)
@property
def option_groups(self) -> tuple[tuple[str, str], ...]:
# TODO: 3.0: Remove deprecated attribute
warnings.warn(
"The option_groups attribute has been deprecated and will be removed in pylint 3.0",
DeprecationWarning,
)
return self._option_groups
@option_groups.setter
def option_groups(self, value: tuple[tuple[str, str], ...]) -> None:
warnings.warn(
"The option_groups attribute has been deprecated and will be removed in pylint 3.0",
DeprecationWarning,
)
self._option_groups = value
def load_default_plugins(self) -> None:
checkers.initialize(self)
reporters.initialize(self)
def load_plugin_modules(self, modnames: list[str]) -> None:
"""Check a list pylint plugins modules, load and register them."""
for modname in modnames:
if modname in self._dynamic_plugins:
continue
self._dynamic_plugins.add(modname)
try:
module = astroid.modutils.load_module_from_name(modname)
module.register(self)
except ModuleNotFoundError:
pass
def load_plugin_configuration(self) -> None:
"""Call the configuration hook for plugins.
This walks through the list of plugins, grabs the "load_configuration"
hook, if exposed, and calls it to allow plugins to configure specific
settings.
"""
for modname in self._dynamic_plugins:
try:
module = astroid.modutils.load_module_from_name(modname)
if hasattr(module, "load_configuration"):
module.load_configuration(self)
except ModuleNotFoundError as e:
self.add_message("bad-plugin-value", args=(modname, e), line=0)
def _load_reporters(self, reporter_names: str) -> None:
"""Load the reporters if they are available on _reporters."""
if not self._reporters:
return
sub_reporters = []
output_files = []
with contextlib.ExitStack() as stack:
for reporter_name in reporter_names.split(","):
reporter_name, *reporter_output = reporter_name.split(":", 1)
reporter = self._load_reporter_by_name(reporter_name)
sub_reporters.append(reporter)
if reporter_output:
output_file = stack.enter_context(
open(reporter_output[0], "w", encoding="utf-8")
)
reporter.out = output_file
output_files.append(output_file)
# Extend the lifetime of all opened output files
close_output_files = stack.pop_all().close
if len(sub_reporters) > 1 or output_files:
self.set_reporter(
reporters.MultiReporter(
sub_reporters,
close_output_files,
)
)
else:
self.set_reporter(sub_reporters[0])
def _load_reporter_by_name(self, reporter_name: str) -> reporters.BaseReporter:
name = reporter_name.lower()
if name in self._reporters:
return self._reporters[name]()
try:
reporter_class = _load_reporter_by_class(reporter_name)
except (ImportError, AttributeError, AssertionError) as e:
raise exceptions.InvalidReporterError(name) from e
else:
return reporter_class()
def set_reporter(
self, reporter: reporters.BaseReporter | reporters.MultiReporter
) -> None:
"""Set the reporter used to display messages and reports."""
self.reporter = reporter
reporter.linter = self
def register_reporter(self, reporter_class: type[reporters.BaseReporter]) -> None:
"""Registers a reporter class on the _reporters attribute."""
self._reporters[reporter_class.name] = reporter_class
def report_order(self) -> list[BaseChecker]:
reports = sorted(self._reports, key=lambda x: getattr(x, "name", ""))
try:
# Remove the current reporter and add it
# at the end of the list.
reports.pop(reports.index(self))
except ValueError:
pass
else:
reports.append(self)
return reports
# checkers manipulation methods ############################################
def register_checker(self, checker: checkers.BaseChecker) -> None:
"""This method auto registers the checker."""
self._checkers[checker.name].append(checker)
for r_id, r_title, r_cb in checker.reports:
self.register_report(r_id, r_title, r_cb, checker)
if hasattr(checker, "msgs"):
self.msgs_store.register_messages_from_checker(checker)
# Register the checker, but disable all of its messages.
if not getattr(checker, "enabled", True):
self.disable(checker.name)
def enable_fail_on_messages(self) -> None:
"""Enable 'fail on' msgs.
Convert values in config.fail_on (which might be msg category, msg id,
or symbol) to specific msgs, then enable and flag them for later.
"""
fail_on_vals = self.config.fail_on
if not fail_on_vals:
return
fail_on_cats = set()
fail_on_msgs = set()
for val in fail_on_vals:
# If value is a category, add category, else add message
if val in MSG_TYPES:
fail_on_cats.add(val)
else:
fail_on_msgs.add(val)
# For every message in every checker, if cat or msg flagged, enable check
for all_checkers in self._checkers.values():
for checker in all_checkers:
for msg in checker.messages:
if msg.msgid in fail_on_msgs or msg.symbol in fail_on_msgs:
# message id/symbol matched, enable and flag it
self.enable(msg.msgid)
self.fail_on_symbols.append(msg.symbol)
elif msg.msgid[0] in fail_on_cats:
# message starts with a category value, flag (but do not enable) it
self.fail_on_symbols.append(msg.symbol)
def any_fail_on_issues(self) -> bool:
return any(x in self.fail_on_symbols for x in self.stats.by_msg.keys())
def disable_reporters(self) -> None:
"""Disable all reporters."""
for _reporters in self._reports.values():
for report_id, _, _ in _reporters:
self.disable_report(report_id)
def _parse_error_mode(self) -> None:
"""Parse the current state of the error mode.
Error mode: enable only errors; no reports, no persistent.
"""
if not self._error_mode:
return
self.disable_noerror_messages()
self.disable("miscellaneous")
self.set_option("reports", False)
self.set_option("persistent", False)
self.set_option("score", False)
# code checking methods ###################################################
def get_checkers(self) -> list[BaseChecker]:
"""Return all available checkers as an ordered list."""
return sorted(c for _checkers in self._checkers.values() for c in _checkers)
def get_checker_names(self) -> list[str]:
"""Get all the checker names that this linter knows about."""
return sorted(
{
checker.name
for checker in self.get_checkers()
if checker.name != MAIN_CHECKER_NAME
}
)
def prepare_checkers(self) -> list[BaseChecker]:
"""Return checkers needed for activated messages and reports."""
if not self.config.reports:
self.disable_reporters()
# get needed checkers
needed_checkers: list[BaseChecker] = [self]
for checker in self.get_checkers()[1:]:
messages = {msg for msg in checker.msgs if self.is_message_enabled(msg)}
if messages or any(self.report_is_enabled(r[0]) for r in checker.reports):
needed_checkers.append(checker)
return needed_checkers
# pylint: disable=unused-argument
@staticmethod
def should_analyze_file(modname: str, path: str, is_argument: bool = False) -> bool:
"""Returns whether a module should be checked.
This implementation returns True for all python source file, indicating
that all files should be linted.
Subclasses may override this method to indicate that modules satisfying
certain conditions should not be linted.
:param str modname: The name of the module to be checked.
:param str path: The full path to the source code of the module.
:param bool is_argument: Whether the file is an argument to pylint or not.
Files which respect this property are always
checked, since the user requested it explicitly.
:returns: True if the module should be checked.
"""
if is_argument:
return True
return path.endswith(".py")
# pylint: enable=unused-argument
def initialize(self) -> None:
"""Initialize linter for linting.
This method is called before any linting is done.
"""
self._ignore_paths = self.config.ignore_paths
# initialize msgs_state now that all messages have been registered into
# the store
for msg in self.msgs_store.messages:
if not msg.may_be_emitted():
self._msgs_state[msg.msgid] = False
def _discover_files(self, files_or_modules: Sequence[str]) -> Iterator[str]:
"""Discover python modules and packages in sub-directory.
Returns iterator of paths to discovered modules and packages.
"""
for something in files_or_modules:
if os.path.isdir(something) and not os.path.isfile(
os.path.join(something, "__init__.py")
):
skip_subtrees: list[str] = []
for root, _, files in os.walk(something):
if any(root.startswith(s) for s in skip_subtrees):
# Skip subtree of already discovered package.
continue
if _is_ignored_file(
root,
self.config.ignore,
self.config.ignore_patterns,
self.config.ignore_paths,
):
skip_subtrees.append(root)
continue
if "__init__.py" in files:
skip_subtrees.append(root)
yield root
else:
yield from (
os.path.join(root, file)
for file in files
if file.endswith(".py")
)
else:
yield something
def check(self, files_or_modules: Sequence[str] | 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.
"""
# 1) Initialize
self.initialize()
# 2) Gather all files
if not isinstance(files_or_modules, (list, tuple)):
# TODO: 3.0: Remove deprecated typing and update docstring
warnings.warn(
"In pylint 3.0, the checkers check function will only accept sequence of string",
DeprecationWarning,
)
files_or_modules = (files_or_modules,) # type: ignore[assignment]
if self.config.recursive:
files_or_modules = tuple(self._discover_files(files_or_modules))
if self.config.from_stdin:
if len(files_or_modules) != 1:
raise exceptions.InvalidArgsError(
"Missing filename required for --from-stdin"
)
# TODO: Move the parallel invocation into step 5 of the checking process
if not self.config.from_stdin and self.config.jobs > 1:
original_sys_path = sys.path[:]
check_parallel(
self,
self.config.jobs,
self._iterate_file_descrs(files_or_modules),
files_or_modules, # this argument patches sys.path
)
sys.path = original_sys_path
return
# 3) Get all FileItems
with fix_import_path(files_or_modules):
if self.config.from_stdin:
fileitems = iter(
(self._get_file_descr_from_stdin(files_or_modules[0]),)
)
data: str | None = _read_stdin()
else:
fileitems = self._iterate_file_descrs(files_or_modules)
data = None
# The contextmanager also opens all checkers and sets up the PyLinter class
with self._astroid_module_checker() as check_astroid_module:
with fix_import_path(files_or_modules):
# 4) Get the AST for each FileItem
ast_per_fileitem = self._get_asts(fileitems, data)
# 5) Lint each ast
self._lint_files(ast_per_fileitem, check_astroid_module)
def _get_asts(
self, fileitems: Iterator[FileItem], data: str | None
) -> dict[FileItem, nodes.Module | None]:
"""Get the AST for all given FileItems."""
ast_per_fileitem: dict[FileItem, nodes.Module | None] = {}
for fileitem in fileitems:
self.set_current_module(fileitem.name, fileitem.filepath)
try:
ast_per_fileitem[fileitem] = self.get_ast(
fileitem.filepath, fileitem.name, data
)
except astroid.AstroidBuildingError as ex:
template_path = prepare_crash_report(
ex, fileitem.filepath, self.crash_file_path
)
msg = get_fatal_error_message(fileitem.filepath, template_path)
self.add_message(
"astroid-error",
args=(fileitem.filepath, msg),
confidence=HIGH,
)
return ast_per_fileitem
def check_single_file(self, name: str, filepath: str, modname: str) -> None:
warnings.warn(
"In pylint 3.0, the checkers check_single_file function will be removed. "
"Use check_single_file_item instead.",
DeprecationWarning,
)
self.check_single_file_item(FileItem(name, filepath, modname))
def check_single_file_item(self, file: FileItem) -> None:
"""Check single file item.
The arguments are the same that are documented in _check_files
initialize() should be called before calling this method
"""
with self._astroid_module_checker() as check_astroid_module:
self._check_file(self.get_ast, check_astroid_module, file)
def _lint_files(
self,
ast_mapping: dict[FileItem, nodes.Module | None],
check_astroid_module: Callable[[nodes.Module], bool | None],
) -> None:
"""Lint all AST modules from a mapping.."""
for fileitem, module in ast_mapping.items():
if module is None:
continue
try:
self._lint_file(fileitem, module, check_astroid_module)
except Exception as ex: # pylint: disable=broad-except
template_path = prepare_crash_report(
ex, fileitem.filepath, self.crash_file_path
)
msg = get_fatal_error_message(fileitem.filepath, template_path)
if isinstance(ex, astroid.AstroidError):
self.add_message(
"astroid-error", args=(fileitem.filepath, msg), confidence=HIGH
)
else:
self.add_message("fatal", args=msg, confidence=HIGH)
def _lint_file(
self,
file: FileItem,
module: nodes.Module,
check_astroid_module: Callable[[nodes.Module], bool | None],
) -> None:
"""Lint a file using the passed utility function check_astroid_module).
:param FileItem file: data about the file
:param nodes.Module module: the ast module to lint
:param Callable check_astroid_module: callable checking an AST taking the following arguments
- ast: AST of the module
:raises AstroidError: for any failures stemming from astroid
"""
self.set_current_module(file.name, file.filepath)
self._ignore_file = False
self.file_state = FileState(file.modpath, self.msgs_store, module)
# fix the current file (if the source file was not available or
# if it's actually a c extension)
self.current_file = module.file
try:
check_astroid_module(module)
except Exception as e:
raise astroid.AstroidError from e
# warn about spurious inline messages handling
spurious_messages = self.file_state.iter_spurious_suppression_messages(
self.msgs_store
)
for msgid, line, args in spurious_messages:
self.add_message(msgid, line, None, args)
def _check_file(
self,
get_ast: GetAstProtocol,
check_astroid_module: Callable[[nodes.Module], bool | None],
file: FileItem,
) -> None:
"""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
- filepath: path to the file to check
- name: Python module name
:param callable check_astroid_module: callable checking an AST taking the following arguments
- ast: AST of the module
:param FileItem file: data about the file
:raises AstroidError: for any failures stemming from astroid
"""
self.set_current_module(file.name, file.filepath)
# get the module representation
ast_node = get_ast(file.filepath, file.name)
if ast_node is None:
return
self._ignore_file = False
self.file_state = FileState(file.modpath, self.msgs_store, ast_node)
# fix the current file (if the source file was not available or
# if it's actually a c extension)
self.current_file = ast_node.file
try:
check_astroid_module(ast_node)
except Exception as e: # pragma: no cover
raise astroid.AstroidError from e
# warn about spurious inline messages handling
spurious_messages = self.file_state.iter_spurious_suppression_messages(
self.msgs_store
)
for msgid, line, args in spurious_messages:
self.add_message(msgid, line, None, args)
@staticmethod
def _get_file_descr_from_stdin(filepath: str) -> FileItem:
"""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
source is standard input.
"""
try:
# Note that this function does not really perform an
# __import__ but may raise an ImportError exception, which
# we want to catch here.
modname = ".".join(astroid.modutils.modpath_from_file(filepath))
except ImportError:
modname = os.path.splitext(os.path.basename(filepath))[0]
return FileItem(modname, filepath, filepath)
def _iterate_file_descrs(
self, files_or_modules: Sequence[str]
) -> Iterator[FileItem]:
"""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.
"""
for descr in self._expand_files(files_or_modules):
name, filepath, is_arg = descr["name"], descr["path"], descr["isarg"]
if self.should_analyze_file(name, filepath, is_argument=is_arg):
yield FileItem(name, filepath, descr["basename"])
def _expand_files(self, modules: Sequence[str]) -> list[ModuleDescriptionDict]:
"""Get modules and errors from a list of modules and handle errors."""
result, errors = expand_modules(
modules,
self.config.ignore,
self.config.ignore_patterns,
self._ignore_paths,
)
for error in errors:
message = modname = error["mod"]
key = error["key"]
self.set_current_module(modname)
if key == "fatal":
message = str(error["ex"]).replace(os.getcwd() + os.sep, "")
self.add_message(key, args=message)
return result
def set_current_module(
self, modname: str | None, filepath: str | None = None
) -> None:
"""Set the name of the currently analyzed module and
init statistics for it.
"""
if not modname and filepath is None:
return
self.reporter.on_set_current_module(modname or "", filepath)
if modname is None:
# TODO: 3.0: Remove all modname or ""'s in this method
warnings.warn(
(
"In pylint 3.0 modname should be a string so that it can be used to "
"correctly set the current_name attribute of the linter instance. "
"If unknown it should be initialized as an empty string."
),
DeprecationWarning,
)
self.current_name = modname
self.current_file = filepath or modname
self.stats.init_single_module(modname or "")
# If there is an actual filepath we might need to update the config attribute
if filepath:
namespace = self._get_namespace_for_file(
Path(filepath), self._directory_namespaces
)
if namespace:
self.config = namespace or self._base_config
def _get_namespace_for_file(
self, filepath: Path, namespaces: DirectoryNamespaceDict
) -> argparse.Namespace | None:
for directory in namespaces:
if _is_relative_to(filepath, directory):
namespace = self._get_namespace_for_file(
filepath, namespaces[directory][1]
)
if namespace is None:
return namespaces[directory][0]
return None
@contextlib.contextmanager
def _astroid_module_checker(
self,
) -> Iterator[Callable[[nodes.Module], bool | None]]:
"""Context manager for checking ASTs.
The value in the context is callable accepting AST as its only argument.
"""
walker = ASTWalker(self)
_checkers = self.prepare_checkers()
tokencheckers = [
c
for c in _checkers
if isinstance(c, checkers.BaseTokenChecker) and c is not self
]
# TODO: 3.0: Remove deprecated for-loop
for c in _checkers:
with warnings.catch_warnings():
warnings.filterwarnings("ignore", category=DeprecationWarning)
if (
interfaces.implements(c, interfaces.ITokenChecker)
and c not in tokencheckers
and c is not self
):
tokencheckers.append(c) # type: ignore[arg-type] # pragma: no cover
warnings.warn( # pragma: no cover
"Checkers should subclass BaseTokenChecker "
"instead of using the __implements__ mechanism. Use of __implements__ "
"will no longer be supported in pylint 3.0",
DeprecationWarning,
)
rawcheckers = [
c for c in _checkers if isinstance(c, checkers.BaseRawFileChecker)
]
# TODO: 3.0: Remove deprecated if-statement
for c in _checkers:
with warnings.catch_warnings():
warnings.filterwarnings("ignore", category=DeprecationWarning)
if (
interfaces.implements(c, interfaces.IRawChecker)
and c not in rawcheckers
):
rawcheckers.append(c) # type: ignore[arg-type] # pragma: no cover
warnings.warn( # pragma: no cover
"Checkers should subclass BaseRawFileChecker "
"instead of using the __implements__ mechanism. Use of __implements__ "
"will no longer be supported in pylint 3.0",
DeprecationWarning,
)
# notify global begin
for checker in _checkers:
checker.open()
walker.add_checker(checker)
yield functools.partial(
self.check_astroid_module,
walker=walker,
tokencheckers=tokencheckers,
rawcheckers=rawcheckers,
)
# notify global end
self.stats.statement = walker.nbstatements
for checker in reversed(_checkers):
checker.close()
def get_ast(
self, filepath: str, modname: str, data: str | None = None
) -> nodes.Module | None:
"""Return an ast(roid) representation of a module or a string.
:param filepath: path to checked file.
:param str modname: The name of the module to be checked.
:param str data: optional contents of the checked file.
:returns: the AST
:rtype: astroid.nodes.Module
:raises AstroidBuildingError: Whenever we encounter an unexpected exception
"""
try:
if data is None:
return MANAGER.ast_from_file(filepath, modname, source=True)
return astroid.builder.AstroidBuilder(MANAGER).string_build(
data, modname, filepath
)
except astroid.AstroidSyntaxError as ex:
self.add_message(
"syntax-error",
line=getattr(ex.error, "lineno", 0),
col_offset=getattr(ex.error, "offset", None),
args=f"Parsing failed: '{ex.error}'",
confidence=HIGH,
)
except astroid.AstroidBuildingError as ex:
self.add_message("parse-error", args=ex)
except Exception as ex:
traceback.print_exc()
# We raise BuildingError here as this is essentially an astroid issue
# Creating an issue template and adding the 'astroid-error' message is handled
# by caller: _check_files
raise astroid.AstroidBuildingError(
"Building error when trying to create ast representation of module '{modname}'",
modname=modname,
) from ex
return None
def check_astroid_module(
self,
ast_node: nodes.Module,
walker: ASTWalker,
rawcheckers: list[checkers.BaseRawFileChecker],
tokencheckers: list[checkers.BaseTokenChecker],
) -> bool | None:
"""Check a module from its astroid representation.
For return value see _check_astroid_module
"""
before_check_statements = walker.nbstatements
retval = self._check_astroid_module(
ast_node, walker, rawcheckers, tokencheckers
)
# TODO: 3.0: Remove unnecessary assertion
assert self.current_name
self.stats.by_module[self.current_name]["statement"] = (
walker.nbstatements - before_check_statements
)
return retval
def _check_astroid_module(
self,
node: nodes.Module,
walker: ASTWalker,
rawcheckers: list[checkers.BaseRawFileChecker],
tokencheckers: list[checkers.BaseTokenChecker],
) -> bool | None:
"""Check given AST node with given walker and checkers.
: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
:returns: True if the module was checked, False if ignored,
None if the module contents could not be parsed
"""
try:
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 node.pure_python:
self.add_message("raw-checker-failed", args=node.name)
else:
# assert astroid.file.endswith('.py')
# Parse module/block level option pragma's
self.process_tokens(tokens)
if self._ignore_file:
return False
# run raw and tokens checkers
for raw_checker in rawcheckers:
raw_checker.process_module(node)
for token_checker in tokencheckers:
token_checker.process_tokens(tokens)
# generate events to astroid checkers
walker.walk(node)
return True
def open(self) -> None:
"""Initialize counters."""
self.stats = LinterStats()
MANAGER.always_load_extensions = self.config.unsafe_load_any_extension
MANAGER.max_inferable_values = self.config.limit_inference_results
MANAGER.extension_package_whitelist.update(self.config.extension_pkg_allow_list)
if self.config.extension_pkg_whitelist:
MANAGER.extension_package_whitelist.update(
self.config.extension_pkg_whitelist
)
self.stats.reset_message_count()
def generate_reports(self) -> int | None:
"""Close the whole package /module, it's time to make reports !
if persistent run, pickle results for later comparison
"""
# Display whatever messages are left on the reporter.
self.reporter.display_messages(report_nodes.Section())
# TODO: 3.0: Remove second half of if-statement
if (
not self.file_state._is_base_filestate
and self.file_state.base_name is not None
):
# load previous results if any
previous_stats = load_results(self.file_state.base_name)
self.reporter.on_close(self.stats, previous_stats)
if self.config.reports:
sect = self.make_reports(self.stats, previous_stats)
else:
sect = report_nodes.Section()
if self.config.reports:
self.reporter.display_reports(sect)
score_value = self._report_evaluation()
# save results if persistent run
if self.config.persistent:
save_results(self.stats, self.file_state.base_name)
else:
self.reporter.on_close(self.stats, LinterStats())
score_value = None
return score_value
def _report_evaluation(self) -> int | None:
"""Make the global evaluation report."""
# check with at least check 1 statements (usually 0 when there is a
# syntax error preventing pylint from further processing)
note = None
# TODO: 3.0: Remove assertion
assert self.file_state.base_name is not None
previous_stats = load_results(self.file_state.base_name)
if self.stats.statement == 0:
return note
# get a global note for the code
evaluation = self.config.evaluation
try:
stats_dict = {
"fatal": self.stats.fatal,
"error": self.stats.error,
"warning": self.stats.warning,
"refactor": self.stats.refactor,
"convention": self.stats.convention,
"statement": self.stats.statement,
"info": self.stats.info,
}
note = eval(evaluation, {}, stats_dict) # pylint: disable=eval-used
except Exception as ex: # pylint: disable=broad-except
msg = f"An exception occurred while rating: {ex}"
else:
self.stats.global_note = note
msg = f"Your code has been rated at {note:.2f}/10"
if previous_stats:
pnote = previous_stats.global_note
if pnote is not None:
msg += f" (previous run: {pnote:.2f}/10, {note - pnote:+.2f})"
if self.config.score:
sect = report_nodes.EvaluationSection(msg)
self.reporter.display_reports(sect)
return note
def _add_one_message(
self,
message_definition: MessageDefinition,
line: int | None,
node: nodes.NodeNG | None,
args: Any | None,
confidence: interfaces.Confidence | None,
col_offset: int | None,
end_lineno: int | None,
end_col_offset: int | None,
) -> None:
"""After various checks have passed a single Message is
passed to the reporter and added to stats.
"""
message_definition.check_message_definition(line, node)
# Look up "location" data of node if not yet supplied
if node:
if node.position:
if not line:
line = node.position.lineno
if not col_offset:
col_offset = node.position.col_offset
if not end_lineno:
end_lineno = node.position.end_lineno
if not end_col_offset:
end_col_offset = node.position.end_col_offset
else:
if not line:
line = node.fromlineno
if not col_offset:
col_offset = node.col_offset
if not end_lineno:
end_lineno = node.end_lineno
if not end_col_offset:
end_col_offset = node.end_col_offset
# should this message be displayed
if not self.is_message_enabled(message_definition.msgid, line, confidence):
self.file_state.handle_ignored_message(
self._get_message_state_scope(
message_definition.msgid, line, confidence
),
message_definition.msgid,
line,
)
return
# update stats
msg_cat = MSG_TYPES[message_definition.msgid[0]]
self.msg_status |= MSG_TYPES_STATUS[message_definition.msgid[0]]
self.stats.increase_single_message_count(msg_cat, 1)
self.stats.increase_single_module_message_count(
self.current_name, # type: ignore[arg-type] # Should be removable after https://github.com/PyCQA/pylint/pull/5580
msg_cat,
1,
)
try:
self.stats.by_msg[message_definition.symbol] += 1
except KeyError:
self.stats.by_msg[message_definition.symbol] = 1
# Interpolate arguments into message string
msg = message_definition.msg
if args is not None:
msg %= args
# get module and object
if node is None:
module, obj = self.current_name, ""
abspath = self.current_file
else:
module, obj = utils.get_module_and_frameid(node)
abspath = node.root().file
if abspath is not None:
path = abspath.replace(self.reporter.path_strip_prefix, "", 1)
else:
path = "configuration"
# add the message
self.reporter.handle_message(
Message(
message_definition.msgid,
message_definition.symbol,
MessageLocationTuple(
abspath or "",
path,
module or "",
obj,
line or 1,
col_offset or 0,
end_lineno,
end_col_offset,
),
msg,
confidence,
)
)
def add_message(
self,
msgid: str,
line: int | None = None,
node: nodes.NodeNG | None = None,
args: Any | None = None,
confidence: interfaces.Confidence | None = None,
col_offset: int | None = None,
end_lineno: int | None = None,
end_col_offset: int | None = None,
) -> None:
"""Adds a message given by ID or name.
If provided, the message string is expanded using args.
AST checkers must provide the node argument (but may optionally
provide line if the line number is different), raw and token checkers
must provide the line argument.
"""
if confidence is None:
confidence = interfaces.UNDEFINED
message_definitions = self.msgs_store.get_message_definitions(msgid)
for message_definition in message_definitions:
self._add_one_message(
message_definition,
line,
node,
args,
confidence,
col_offset,
end_lineno,
end_col_offset,
)
def add_ignored_message(
self,
msgid: str,
line: int,
node: nodes.NodeNG | None = None,
confidence: interfaces.Confidence | None = interfaces.UNDEFINED,
) -> None:
"""Prepares a message to be added to the ignored message storage.
Some checks return early in special cases and never reach add_message(),
even though they would normally issue a message.
This creates false positives for useless-suppression.
This function avoids this by adding those message to the ignored msgs attribute
"""
message_definitions = self.msgs_store.get_message_definitions(msgid)
for message_definition in message_definitions:
message_definition.check_message_definition(line, node)
self.file_state.handle_ignored_message(
self._get_message_state_scope(
message_definition.msgid, line, confidence
),
message_definition.msgid,
line,
)
def _emit_stashed_messages(self) -> None:
for keys, values in self._stashed_messages.items():
modname, symbol = keys
self.linter.set_current_module(modname)
for args in values:
self.add_message(
symbol,
args=args,
line=0,
confidence=HIGH,
)
self._stashed_messages = collections.defaultdict(list)
|