multilint

  1#!/usr/bin/env python3
  2"""
  3Multilint is a runner for multiple code quality tools.
  4
  5Multilint enables running various Python linters (and other CQ tools) under the
  6same interface, making it easier to configure and run all code quality tools
  7from a centralized location.
  8"""
  9from __future__ import annotations
 10
 11import logging
 12import re
 13import sys
 14from argparse import Namespace
 15from collections.abc import Iterable, Mapping
 16from collections.abc import Sequence as Seq
 17from dataclasses import dataclass, field
 18from enum import Enum, auto
 19from glob import glob
 20from io import TextIOBase
 21from logging import Formatter, Logger, StreamHandler
 22from pathlib import Path
 23from typing import Any, TextIO, TypeVar, cast
 24from unittest.mock import patch
 25
 26import pydocstyle  # type: ignore
 27import toml
 28from autoflake import _main as autoflake_main  # type: ignore
 29from black import main as black_main
 30from isort import files as isort_files
 31from isort.api import sort_file as isort_file
 32from isort.settings import DEFAULT_CONFIG
 33from mypy.main import main as mypy_main  # pylint: disable=no-name-in-module
 34from pylint.lint import Run as PylintRun  # type: ignore
 35from pyupgrade._main import _fix_file as pyupgrade_fix_file  # type: ignore
 36
 37FILE_DIR: Path = Path(__file__).resolve().parent
 38ROOT_DIR: Path = FILE_DIR
 39PYPROJECT_TOML_FILENAME: str = "pyproject.toml"
 40
 41LogLevel = TypeVar("LogLevel", bound=int)
 42
 43LOG_FMT: str = "%(asctime)s [%(levelname)s] [%(name)s] %(msg)s"
 44logging.basicConfig(level=logging.INFO, format=LOG_FMT)
 45LOGGER: Logger = logging.getLogger("multilint")
 46
 47
 48class Tool(Enum):
 49    """Encapsulates all supported linters, including Multilint itself."""
 50
 51    AUTOFLAKE = "autoflake"
 52    BLACK = "black"
 53    ISORT = "isort"
 54    MULTILINT = "multilint"
 55    MYPY = "mypy"
 56    PYDOCSTYLE = "pydocstyle"
 57    PYLINT = "pylint"
 58    PYUPGRADE = "pyupgrade"
 59
 60
 61class ToolResult(Enum):
 62    """ToolResult describes a generic run result from a code quality tool."""
 63
 64    SUCCESS = auto()
 65    SUCCESS_PARTIAL = auto()
 66    FAILURE = auto()
 67
 68
 69class ToolLogger(Logger):
 70    """ToolLogger allows setting format on itself during instantiation."""
 71
 72    def __init__(
 73        self: ToolLogger, name: str, level: LogLevel, logfmt: str = LOG_FMT
 74    ) -> None:
 75        """Create a ToolLogger with the specified name, level, and format.
 76
 77        Log format gets applied immediately.
 78        """
 79        super().__init__(name, level=level)
 80
 81        self.set_format(logfmt)
 82
 83    def set_format(self: ToolLogger, fmtstr: str) -> None:
 84        """Set the specified log message format.
 85
 86        Uses a StreamHandler by setting a formatter on it with the specified
 87        format string, and adds the StreamHanlder to the logger instance.
 88        """
 89        handler: StreamHandler = StreamHandler()
 90        handler.setFormatter(Formatter(fmtstr))
 91        self.addHandler(handler)
 92
 93
 94class TextIOLogger(TextIOBase, ToolLogger):
 95    """Logger object that can be written to like a stream-like object.
 96
 97    A logger that masquerades as a TextIO-compatible object, allowing it to be
 98    passed into code quality tools that write to TextIO interfaces. This way,
 99    it is possible to wrap the stdout / stderr / other streams with our
100    common logging.
101    """
102
103    def __init__(
104        self: TextIOLogger, name: str, level: LogLevel, logfmt: str = LOG_FMT
105    ) -> None:
106        """Construct a TextIOLogger."""
107        self._name: str = name
108        self._level: int = level
109
110        super().__init__(self._name, self._level, logfmt)
111
112    def write(self: TextIOLogger, msg: str) -> int:
113        """Write data to the logger as if it were a stream-like object.
114
115        The write() method is implemented to forward data to the logging
116        part of this object, allowing capturing logs that would normally be
117        written to stdout / stderr.
118        """
119        if msg in ("", "\n"):
120            return 0
121
122        for line in filter(None, msg.split("\n")):
123            self.log(self._level, line)
124
125        return len(msg)
126
127
128class ToolRunner:
129    """Base class for integrating code quality tools.
130
131    ToolRunner is a base class for any plugin that integrates a code quality
132    tool. Subclasses only have to implement the run() method. There is a
133    convenience method available,
134    """
135
136    def __init__(
137        self: ToolRunner,
138        tool: Tool,
139        src_paths: Seq[Path] = [Path(".")],
140        config: Mapping[str, Any] = {},
141    ) -> None:
142        """Initialize a ToolRunner object."""
143        self._tool: Tool = tool
144        self.src_paths: Seq[Path] = src_paths
145        self.config: Mapping[str, Any] = config
146
147    def make_logger(
148        self: ToolRunner,
149        cls: type[ToolLogger] | type[TextIOLogger],
150        level: LogLevel,
151    ) -> ToolLogger | TextIOLogger:
152        """Create a logger for the ToolRunner object.
153
154        Creates a logger object from the specified logger class (can be
155        either a ToolLogger or a TextIOLogger) with the specified default log
156        level.
157        """
158        return cls(f"tool.{self._tool.value}", level)
159
160    def run(self: ToolRunner) -> ToolResult:
161        """Is implemented by subclasses to run the CQ (code quality) tool."""
162        raise NotImplementedError("run() needs to be implemented by subclass!")
163
164
165class AutoflakeRunner(ToolRunner):
166    """Runs autoflake.
167
168    Autoflake removes unused imports and variables among other
169    things. Reads autoflake arguments from pyproject.toml. Arguments are
170    specified by their full name, with underscores or dashes - either style is
171    accepted.
172    """
173
174    def _make_autoflake_args(self: AutoflakeRunner) -> list[str]:
175        args: list[str] = []
176
177        for key, val in self.config.items():
178            if key == "src_paths":
179                for src in val:
180                    args.append(src)
181
182                continue
183
184            opt: str = f"--{key.replace('_', '-')}"
185
186            if isinstance(val, bool):
187                args.append(opt)
188
189                continue
190
191            args.append(f"{opt}={val}")
192
193        return args
194
195    def run(self: AutoflakeRunner) -> ToolResult:
196        """Run autoflake."""
197        logger: Logger = self.make_logger(TextIOLogger, logging.INFO)
198        autoflake_args: list[str] = self._make_autoflake_args()
199        if all(cfgval.startswith("--") for cfgval in autoflake_args):
200            autoflake_args.extend([str(p) for p in self.src_paths])
201
202        if "--in-place" not in autoflake_args:
203            autoflake_args.append("--in-place")
204
205        retcode: int = autoflake_main(
206            [self._tool.value, *autoflake_args], logger, logger
207        )
208
209        return ToolResult.SUCCESS if retcode == 0 else ToolResult.FAILURE
210
211
212@dataclass
213class ISortResult:
214    """Isort run result.
215
216    Encapsulates a result from an isort run: the Python file, the ToolResult,
217    and an error message (if and when applicable).
218    """
219
220    pyfile: Path
221    result: ToolResult
222    errmsg: str = field(default="")
223
224    def __hash__(self) -> int:
225        """Hashes all attributes of the dataclass."""
226        return hash(f"{self.pyfile}{self.result.value}{self.errmsg}")
227
228
229class ISortRunner(ToolRunner):
230    """Runs isort.
231
232    Isort is able to sort imports in a Python source file according to a defined
233    rule set (with sensible defaults).
234    """
235
236    # pylint: disable=too-many-branches
237    def run(self: ISortRunner) -> ToolResult:
238        """Run isort."""
239        logger: Logger = self.make_logger(ToolLogger, logging.INFO)
240        results: set[ISortResult] = set()
241
242        isort_logger: TextIOLogger = cast(
243            TextIOLogger, self.make_logger(TextIOLogger, logging.INFO)
244        )
245
246        # fmt: off
247        for file in isort_files.find(
248            [str(p) for p in self.src_paths]
249            or cast(Iterable[str], DEFAULT_CONFIG.src_paths),
250            DEFAULT_CONFIG, [], [],
251        ):
252            try:
253                with patch("sys.stdout", isort_logger):
254                    isort_file(file)
255
256                results.add(ISortResult(Path(file), ToolResult.SUCCESS))
257            # pylint: disable=broad-except
258            except Exception as ex:
259                results.add(ISortResult(
260                    Path(file), ToolResult.FAILURE,
261                    getattr(ex, "message") if hasattr(ex, "message") else "",
262                ))
263        # fmt: on
264
265        failed: set[ISortResult] = set()
266        for isort_result in results:
267            if isort_result.result == ToolResult.FAILURE:
268                failed.add(isort_result)
269
270        if len(failed) > 0:
271            logger.error("isort failed on some files:")
272
273            for failed_result in failed:
274                logger.error(f"{failed_result.pyfile}: {failed_result.errmsg}")
275
276        if len(failed) == len(results):
277            return ToolResult.FAILURE
278
279        if 0 < len(failed) < len(results):
280            return ToolResult.SUCCESS_PARTIAL
281
282        return ToolResult.SUCCESS
283
284
285class BlackRunner(ToolRunner):
286    """Runs black.
287
288    Black is an opinionated code formatter.
289    """
290
291    def run(self: BlackRunner) -> ToolResult:
292        """Run black."""
293        iologger: Logger = self.make_logger(TextIOLogger, logging.INFO)
294        sys_stdout_orig = sys.stdout
295        sys_stderr_orig = sys.stderr
296
297        try:
298            sys.stdout = cast(TextIO, iologger)
299            sys.stderr = cast(TextIO, iologger)
300            # pylint: disable=no-value-for-parameter
301            black_main([str(p) for p in self.src_paths])
302
303            return ToolResult.SUCCESS
304
305        except SystemExit as sysexit:
306            return ToolResult.SUCCESS if sysexit.code == 0 else ToolResult.FAILURE
307
308        finally:
309            sys.stdout = sys_stdout_orig
310            sys.stderr = sys_stderr_orig
311
312
313class MypyRunner(ToolRunner):
314    """Runs Mypy.
315
316    Mypy is a static type checker for Python.
317    """
318
319    def run(self: MypyRunner) -> ToolResult:
320        """Run mypy."""
321        logger: TextIOLogger = cast(
322            TextIOLogger, self.make_logger(TextIOLogger, logging.INFO)
323        )
324
325        logger_as_textio: TextIO = cast(TextIO, logger)
326
327        try:
328            mypy_main(stdout=logger_as_textio, stderr=logger_as_textio, clean_exit=True)
329
330            return ToolResult.SUCCESS
331
332        except SystemExit as sysexit:
333            return ToolResult.SUCCESS if sysexit.code == 0 else ToolResult.FAILURE
334
335
336class PylintRunner(ToolRunner):
337    """Runs pylint."""
338
339    def run(self: PylintRunner) -> ToolResult:
340        """Run pylint."""
341        with patch("sys.stdout", self.make_logger(TextIOLogger, logging.INFO)):
342            try:
343                PylintRun([str(p) for p in self.src_paths])
344
345                return ToolResult.SUCCESS
346
347            except SystemExit as sysexit:
348                return ToolResult.SUCCESS if sysexit.code == 0 else ToolResult.FAILURE
349
350
351class PydocstyleRunner(ToolRunner):
352    """Runs pydocstyle.
353
354    Pydocstyle checks for best practices for writing Python documentation within
355    source code.
356    """
357
358    def run(self: PydocstyleRunner) -> ToolResult:
359        """Run pydocstyle."""
360
361        class InfoToolLogger(ToolLogger):
362            """Disobedient logger that stays at the specified level.
363
364            Shim logger to default to a specified level regardless of level
365            passed.
366            """
367
368            def setLevel(self: InfoToolLogger, _: int | str) -> None:
369                self.level = logging.INFO
370
371        info_logger: InfoToolLogger = InfoToolLogger(self._tool.value, logging.INFO)
372        pydocstyle_log_orig: Logger = pydocstyle.utils.log  # type: ignore
373        try:
374            pydocstyle.utils.log = info_logger  # type: ignore
375            pydocstyle.checker.log = info_logger
376            # pylint: disable=import-outside-toplevel
377            from pydocstyle.cli import run_pydocstyle as pydocstyle_main  # type: ignore
378
379            with patch("sys.stdout", self.make_logger(TextIOLogger, logging.INFO)):
380                return [
381                    ToolResult.SUCCESS,
382                    ToolResult.SUCCESS_PARTIAL,
383                    ToolResult.FAILURE,
384                ][pydocstyle_main()]
385
386        finally:
387            pydocstyle.utils.log = pydocstyle_log_orig  # type: ignore
388            pydocstyle.checker.log = pydocstyle_log_orig
389
390
391class PyupgradeRunner(ToolRunner):
392    """Runs Pyupgrade.
393
394    Pyupgrade automatically upgrades Python syntax to the latest for the
395    specified Python version.
396    """
397
398    def _validate_config(self: PyupgradeRunner) -> None:
399        if "min_version" in self.config and not re.match(
400            r"^[0-9].[0-9]", self.config["min_version"]
401        ):
402            raise ValueError("min_version must be a valid Python version!")
403
404    def run(self: PyupgradeRunner) -> ToolResult:
405        """Run Pyupgrade."""
406        self._validate_config()
407
408        logger: ToolLogger = self.make_logger(TextIOLogger, logging.INFO)
409
410        with patch("sys.stdout", logger), patch("sys.stderr", logger):
411            retcode: int = 0
412            for src_path in [p for p in self.src_paths if p.is_file()]:
413                retcode |= pyupgrade_fix_file(
414                    src_path.resolve().name,
415                    Namespace(
416                        exit_zero_even_if_changed=self.config.get(
417                            "exit_zero_even_if_changed", None
418                        ),
419                        keep_mock=self.config.get("keep_mock", None),
420                        keep_percent_format=self.config.get(
421                            "keep_percent_format", None
422                        ),
423                        keep_runtime_typing=self.config.get(
424                            "keep_runtime_typing", None
425                        ),
426                        min_version=tuple(
427                            int(v)
428                            for v in cast(
429                                str, self.config.get("min_version", "2.7")
430                            ).split(".")
431                        ),
432                    ),
433                )
434
435            return ToolResult.SUCCESS if retcode == 0 else ToolResult.FAILURE
436
437
438def find_pyproject_toml(path: Path = Path(".")) -> Path | None:
439    """Discover closest pyproject.toml.
440
441    Finds the first pyproject.toml by searching in the current directory,
442    traversing upward to the filesystem root if not found.
443    """
444    if path == Path("/"):
445        return None
446
447    filepath: Path = path / PYPROJECT_TOML_FILENAME
448
449    if filepath.exists() and filepath.is_file():
450        return path / PYPROJECT_TOML_FILENAME
451
452    return find_pyproject_toml(path.resolve().parent)
453
454
455def parse_pyproject_toml(pyproject_toml_path: Path = Path(".")) -> Mapping[str, Any]:
456    """Parse pyproject.toml into a config map.
457
458    Reads in the pyproject.toml file and returns a parsed version of it as a
459    Mapping.
460    """
461    pyproject_toml: Path | None = find_pyproject_toml(pyproject_toml_path)
462    if pyproject_toml is None:
463        return {}
464
465    with pyproject_toml.open("r") as file:
466        return toml.load(file)
467
468
469def expand_src_paths(src_paths: Seq[Path]) -> list[Path]:
470    """Expand source paths in case they are globs."""
471    return sum(
472        (
473            [Path(ge) for ge in glob(p.name)] if "*" in p.name else [p]
474            for p in src_paths
475        ),
476        [],
477    )
478
479
480TOOL_RUNNERS: Mapping[Tool, type[ToolRunner]] = {
481    Tool.AUTOFLAKE: AutoflakeRunner,
482    Tool.BLACK: BlackRunner,
483    Tool.ISORT: ISortRunner,
484    Tool.MYPY: MypyRunner,
485    Tool.PYDOCSTYLE: PydocstyleRunner,
486    Tool.PYLINT: PylintRunner,
487    Tool.PYUPGRADE: PyupgradeRunner,
488}
489
490
491class Multilint:
492    """The core logic of this project.
493
494    Multilint ties together multiple linting and other code quality tools
495    under a single interface, the ToolRunner base class. By subclassing
496    ToolRunner and implementing its .run() method, adding support for linters /
497    other tool plugins is easy.
498    """
499
500    DEFAULT_TOOL_ORDER: Seq[Tool] = [
501        Tool.PYUPGRADE,
502        Tool.AUTOFLAKE,
503        Tool.ISORT,
504        Tool.BLACK,
505        Tool.MYPY,
506        Tool.PYLINT,
507        Tool.PYDOCSTYLE,
508    ]
509
510    def __init__(
511        self: Multilint,
512        src_paths: Seq[Path] = [Path(".")],
513        pyproject_toml_path: Path = Path(".") / PYPROJECT_TOML_FILENAME,
514    ) -> None:
515        """Construct a new Multilint instance."""
516        self._config: Mapping[str, Any] = {}
517        self._config = parse_pyproject_toml(pyproject_toml_path)
518
519        self._multilint_config = self._get_tool_config(Tool.MULTILINT)
520        self._src_paths: Seq[Path] = (
521            src_paths
522            if len(src_paths) > 0
523            else self._multilint_config.get("src_paths", ["."])
524        )
525
526        self._tool_order: Seq[Tool] = [
527            Tool(t) for t in self._multilint_config.get("tool_order", [])
528        ]
529        if len(self._tool_order) == 0:
530            self._tool_order = self.DEFAULT_TOOL_ORDER
531
532    def _get_tool_config(self: Multilint, tool: Tool) -> Mapping[str, Any]:
533        return self._config.get("tool", {}).get(tool.value, {})
534
535    def run_tool(self: Multilint, tool: Tool) -> ToolResult:
536        """Run a single CQ tool.
537
538        Runs a single specified linter or other code quality tool. Returns
539        a ToolResult from the run.
540        """
541        LOGGER.info(f"Running {tool.value}...")
542        tool_config: Mapping[str, Any] = self._get_tool_config(tool)
543
544        # fmt: off
545        result: ToolResult = cast(ToolRunner, TOOL_RUNNERS[tool](
546            tool,
547            expand_src_paths(
548                [Path(sp) for sp in tool_config.get("src_paths", self._src_paths)]
549            ),
550            tool_config,
551        )).run()
552        # fmt: on
553
554        LOGGER.info(f"{tool.value} exited with {result}")
555
556        return result
557
558    def run_all_tools(
559        self: Multilint, order: Seq[Tool] = []
560    ) -> Mapping[Tool, ToolResult]:
561        """Run tools in specified order."""
562        results: dict[Tool, ToolResult] = {}
563
564        if not order:
565            order = self._tool_order
566
567        for tool in order:
568            results[tool] = self.run_tool(tool)
569
570        return results
571
572
573def main(
574    src_paths: Seq[Path] = [Path(p) for p in sys.argv[1:]], do_exit: bool = True  # type: ignore
575) -> int | None:
576    """Acts as the default entry point for Multilint.
577
578    The main / default entry point to multilint. Runs all tools and logs
579    their results.
580    """
581    results: Mapping[Tool, ToolResult] = Multilint(src_paths).run_all_tools()
582
583    LOGGER.info("Results:")
584    for tool, result in results.items():
585        LOGGER.info(f"{tool}: {result}")
586
587    retcode: int = 0 if all(r == ToolResult.SUCCESS for r in results.values()) else 1
588
589    if do_exit:
590        sys.exit(retcode)
591
592    return retcode
593
594
595if __name__ == "__main__":
596    sys.exit(main([Path(arg) for arg in sys.argv[1:]]))