Source code for solitude.linter.linter

# Copyright (c) 2019, Solitude Developers
# This source code is licensed under the BSD-3-Clause license found in the
# COPYING file in the root directory of this source tree

from typing import List, Union, Iterator, Tuple  # noqa
from collections import OrderedDict

from solitude.common import (
    ContractSourceList, FileMessage, path_to_unitname)

from multiprocessing.pool import ThreadPool
from solitude.linter.solium_wrapper import SoliumWrapper

[docs]class Linter: """The linter object allows linting groups of contracts. """
[docs] def __init__( self, executable: str, plugins: List[str], rules: dict, parallelism: int=4): """Create a Linter :param executable: path to the solium executable file :param plugins: list of solium plugins :param rules: dictionary containing solium rules :param parallelism: maximum number of parallel instances to be run """ self._solium = SoliumWrapper(executable, plugins, rules) self._parallelism = parallelism
[docs] def lint(self, sourcelist: ContractSourceList) -> Iterator[Tuple[str, FileMessage]]: """Lint a group of contracts :param sourcelist: source files to lint, as ContractSourceList :return: an iterator of FileMessage objects containing the linter output information """ pool = ThreadPool(processes=self._parallelism) futures = [] for path in sourcelist.file_sources: futures.append( pool.apply_async(self._solium.lint_file, args=(path, ))) for unitname, source in sourcelist.text_sources.items(): futures.append( pool.apply_async(self._solium.lint_source, args=(source, unitname))) for future in futures: errors = future.get() for error in errors: yield error