djavu-asus/elpy/rpc-venv/lib/python3.11/site-packages/flake8/main/application.py

   1 """Module containing the application logic for Flake8."""
   2 from __future__ import annotations
   3
   4 import argparse
   5 import json
   6 import logging
   7 import time
   8 from typing import Sequence
   9
  10 import flake8
  11 from flake8 import checker
  12 from flake8 import defaults
  13 from flake8 import exceptions
  14 from flake8 import style_guide
  15 from flake8.formatting.base import BaseFormatter
  16 from flake8.main import debug
  17 from flake8.options.parse_args import parse_args
  18 from flake8.plugins import finder
  19 from flake8.plugins import reporter
  20
  21
  22 LOG = logging.getLogger(__name__)
  23
  24
  25 class Application:
  26     """Abstract our application into a class."""
  27
  28     def __init__(self) -> None:
  29         """Initialize our application."""
  30         #: The timestamp when the Application instance was instantiated.
  31         self.start_time = time.time()
  32         #: The timestamp when the Application finished reported errors.
  33         self.end_time: float | None = None
  34
  35         self.plugins: finder.Plugins | None = None
  36         #: The user-selected formatter from :attr:`formatting_plugins`
  37         self.formatter: BaseFormatter | None = None
  38         #: The :class:`flake8.style_guide.StyleGuideManager` built from the
  39         #: user's options
  40         self.guide: style_guide.StyleGuideManager | None = None
  41         #: The :class:`flake8.checker.Manager` that will handle running all of
  42         #: the checks selected by the user.
  43         self.file_checker_manager: checker.Manager | None = None
  44
  45         #: The user-supplied options parsed into an instance of
  46         #: :class:`argparse.Namespace`
  47         self.options: argparse.Namespace | None = None
  48         #: The number of errors, warnings, and other messages after running
  49         #: flake8 and taking into account ignored errors and lines.
  50         self.result_count = 0
  51         #: The total number of errors before accounting for ignored errors and
  52         #: lines.
  53         self.total_result_count = 0
  54         #: Whether or not something catastrophic happened and we should exit
  55         #: with a non-zero status code
  56         self.catastrophic_failure = False
  57
  58     def exit_code(self) -> int:
  59         """Return the program exit code."""
  60         if self.catastrophic_failure:
  61             return 1
  62         assert self.options is not None
  63         if self.options.exit_zero:
  64             return 0
  65         else:
  66             return int(self.result_count > 0)
  67
  68     def make_formatter(self) -> None:
  69         """Initialize a formatter based on the parsed options."""
  70         assert self.plugins is not None
  71         assert self.options is not None
  72         self.formatter = reporter.make(self.plugins.reporters, self.options)
  73
  74     def make_guide(self) -> None:
  75         """Initialize our StyleGuide."""
  76         assert self.formatter is not None
  77         assert self.options is not None
  78         self.guide = style_guide.StyleGuideManager(
  79             self.options, self.formatter
  80         )
  81
  82     def make_file_checker_manager(self, argv: Sequence[str]) -> None:
  83         """Initialize our FileChecker Manager."""
  84         assert self.guide is not None
  85         assert self.plugins is not None
  86         self.file_checker_manager = checker.Manager(
  87             style_guide=self.guide,
  88             plugins=self.plugins.checkers,
  89             argv=argv,
  90         )
  91
  92     def run_checks(self) -> None:
  93         """Run the actual checks with the FileChecker Manager.
  94
  95         This method encapsulates the logic to make a
  96         :class:`~flake8.checker.Manger` instance run the checks it is
  97         managing.
  98         """
  99         assert self.file_checker_manager is not None
 100
 101         self.file_checker_manager.start()
 102         try:
 103             self.file_checker_manager.run()
 104         except exceptions.PluginExecutionFailed as plugin_failed:
 105             print(str(plugin_failed))
 106             print("Run flake8 with greater verbosity to see more details")
 107             self.catastrophic_failure = True
 108         LOG.info("Finished running")
 109         self.file_checker_manager.stop()
 110         self.end_time = time.time()
 111
 112     def report_benchmarks(self) -> None:
 113         """Aggregate, calculate, and report benchmarks for this run."""
 114         assert self.options is not None
 115         if not self.options.benchmark:
 116             return
 117
 118         assert self.file_checker_manager is not None
 119         assert self.end_time is not None
 120         time_elapsed = self.end_time - self.start_time
 121         statistics = [("seconds elapsed", time_elapsed)]
 122         add_statistic = statistics.append
 123         for statistic in defaults.STATISTIC_NAMES + ("files",):
 124             value = self.file_checker_manager.statistics[statistic]
 125             total_description = f"total {statistic} processed"
 126             add_statistic((total_description, value))
 127             per_second_description = f"{statistic} processed per second"
 128             add_statistic((per_second_description, int(value / time_elapsed)))
 129
 130         assert self.formatter is not None
 131         self.formatter.show_benchmarks(statistics)
 132
 133     def report_errors(self) -> None:
 134         """Report all the errors found by flake8 3.0.
 135
 136         This also updates the :attr:`result_count` attribute with the total
 137         number of errors, warnings, and other messages found.
 138         """
 139         LOG.info("Reporting errors")
 140         assert self.file_checker_manager is not None
 141         results = self.file_checker_manager.report()
 142         self.total_result_count, self.result_count = results
 143         LOG.info(
 144             "Found a total of %d violations and reported %d",
 145             self.total_result_count,
 146             self.result_count,
 147         )
 148
 149     def report_statistics(self) -> None:
 150         """Aggregate and report statistics from this run."""
 151         assert self.options is not None
 152         if not self.options.statistics:
 153             return
 154
 155         assert self.formatter is not None
 156         assert self.guide is not None
 157         self.formatter.show_statistics(self.guide.stats)
 158
 159     def initialize(self, argv: Sequence[str]) -> None:
 160         """Initialize the application to be run.
 161
 162         This finds the plugins, registers their options, and parses the
 163         command-line arguments.
 164         """
 165         self.plugins, self.options = parse_args(argv)
 166
 167         if self.options.bug_report:
 168             info = debug.information(flake8.__version__, self.plugins)
 169             print(json.dumps(info, indent=2, sort_keys=True))
 170             raise SystemExit(0)
 171
 172         self.make_formatter()
 173         self.make_guide()
 174         self.make_file_checker_manager(argv)
 175
 176     def report(self) -> None:
 177         """Report errors, statistics, and benchmarks."""
 178         assert self.formatter is not None
 179         self.formatter.start()
 180         self.report_errors()
 181         self.report_statistics()
 182         self.report_benchmarks()
 183         self.formatter.stop()
 184
 185     def _run(self, argv: Sequence[str]) -> None:
 186         self.initialize(argv)
 187         self.run_checks()
 188         self.report()
 189
 190     def run(self, argv: Sequence[str]) -> None:
 191         """Run our application.
 192
 193         This method will also handle KeyboardInterrupt exceptions for the
 194         entirety of the flake8 application. If it sees a KeyboardInterrupt it
 195         will forcibly clean up the :class:`~flake8.checker.Manager`.
 196         """
 197         try:
 198             self._run(argv)
 199         except KeyboardInterrupt as exc:
 200             print("... stopped")
 201             LOG.critical("Caught keyboard interrupt from user")
 202             LOG.exception(exc)
 203             self.catastrophic_failure = True
 204         except exceptions.ExecutionError as exc:
 205             print("There was a critical error during execution of Flake8:")
 206             print(exc)
 207             LOG.exception(exc)
 208             self.catastrophic_failure = True
 209         except exceptions.EarlyQuit:
 210             self.catastrophic_failure = True
 211             print("... stopped while processing files")
 212         else:
 213             assert self.options is not None
 214             if self.options.count:
 215                 print(self.result_count)