pre-commit/flake8
1
0
Fork 0
mirror of https://github.com/PyCQA/flake8.git synced 2026-04-05 20:46:54 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

remove vcs integration

Browse source
This commit is contained in:
Anthony Sottile 2021-03-29 19:01:59 -07:00
parent 434c108f74
commit 1a1d850e99
13 changed files with 2 additions and 659 deletions
Download patch file Download diff file Expand all files Collapse all files

7
src/flake8/checker.py
View file

@ -200,7 +200,6 @@ class Manager(object):
paths = ["."]
filename_patterns = self.options.filename
running_from_vcs = self.options._running_from_vcs
running_from_diff = self.options.diff
# NOTE(sigmavirus24): Yes this is a little unsightly, but it's our
@ -218,10 +217,8 @@ class Manager(object):
# the event that the argument and the filename are identical.
# If it was specified explicitly, the user intended for it to be
# checked.
explicitly_provided = (
not running_from_vcs
and not running_from_diff
and (argument == filename)
explicitly_provided = not running_from_diff and (
argument == filename
)
return (
explicitly_provided or matches_filename_patterns

48
src/flake8/exceptions.py
View file

@ -93,51 +93,3 @@ class PluginExecutionFailed(Flake8Exception):
"name": self.plugin["plugin_name"],
"exc": self.original_exception,
}
class HookInstallationError(Flake8Exception):
"""Parent exception for all hooks errors."""
class GitHookAlreadyExists(HookInstallationError):
"""Exception raised when the git pre-commit hook file already exists."""
def __init__(self, path): # type: (str) -> None
"""Initialize the exception message from the `path`."""
self.path = path
tmpl = (
"The Git pre-commit hook ({0}) already exists. To convince "
"Flake8 to install the hook, please remove the existing "
"hook."
)
super(GitHookAlreadyExists, self).__init__(tmpl.format(self.path))
class MercurialHookAlreadyExists(HookInstallationError):
"""Exception raised when a mercurial hook is already configured."""
hook_name = None # type: str
def __init__(self, path, value): # type: (str, str) -> None
"""Initialize the relevant attributes."""
self.path = path
self.value = value
tmpl = (
'The Mercurial {0} hook already exists with "{1}" in {2}. '
"To convince Flake8 to install the hook, please remove the "
"{0} configuration from the [hooks] section of your hgrc."
)
msg = tmpl.format(self.hook_name, self.value, self.path)
super(MercurialHookAlreadyExists, self).__init__(msg)
class MercurialCommitHookAlreadyExists(MercurialHookAlreadyExists):
"""Exception raised when the hg commit hook is already configured."""
hook_name = "commit"
class MercurialQRefreshHookAlreadyExists(MercurialHookAlreadyExists):
"""Exception raised when the hg commit hook is already configured."""
hook_name = "qrefresh"

2
src/flake8/main/application.py
View file

@ -191,8 +191,6 @@ class Application(object):
if not self.parsed_diff:
self.exit()
self.options._running_from_vcs = False
self.check_plugins.provide_options(
self.option_manager, self.options, self.args
)

262
src/flake8/main/git.py
View file

@ -1,262 +0,0 @@
"""Module containing the main git hook interface and helpers.
.. autofunction:: hook
.. autofunction:: install
"""
import contextlib
import os
import os.path
import shutil
import stat
import subprocess
import sys
import tempfile
from flake8 import defaults
from flake8 import exceptions
__all__ = ("hook", "install")
def hook(lazy=False, strict=False):
"""Execute Flake8 on the files in git's index.
Determine which files are about to be committed and run Flake8 over them
to check for violations.
:param bool lazy:
Find files not added to the index prior to committing. This is useful
if you frequently use ``git commit -a`` for example. This defaults to
False since it will otherwise include files not in the index.
:param bool strict:
If True, return the total number of errors/violations found by Flake8.
This will cause the hook to fail.
:returns:
Total number of errors found during the run.
:rtype:
int
"""
# NOTE(sigmavirus24): Delay import of application until we need it.
from flake8.main import application
app = application.Application()
with make_temporary_directory() as tempdir:
filepaths = list(copy_indexed_files_to(tempdir, lazy))
app.initialize(["."])
app.options.exclude = update_excludes(app.options.exclude, tempdir)
app.options._running_from_vcs = True
# Apparently there are times when there are no files to check (e.g.,
# when amending a commit). In those cases, let's not try to run checks
# against nothing.
if filepaths:
app.run_checks(filepaths)
# If there were files to check, update their paths and report the errors
if filepaths:
update_paths(app.file_checker_manager, tempdir)
app.report_errors()
if strict:
return app.result_count
return 0
def install():
"""Install the git hook script.
This searches for the ``.git`` directory and will install an executable
pre-commit python script in the hooks sub-directory if one does not
already exist.
It will also print a message to stdout about how to configure the hook.
:returns:
True if successful, False if the git directory doesn't exist.
:rtype:
bool
:raises:
flake8.exceptions.GitHookAlreadyExists
"""
git_directory = find_git_directory()
if git_directory is None or not os.path.exists(git_directory):
return False
hooks_directory = os.path.join(git_directory, "hooks")
if not os.path.exists(hooks_directory):
os.mkdir(hooks_directory)
pre_commit_file = os.path.abspath(
os.path.join(hooks_directory, "pre-commit")
)
if os.path.exists(pre_commit_file):
raise exceptions.GitHookAlreadyExists(path=pre_commit_file)
executable = get_executable()
with open(pre_commit_file, "w") as fd:
fd.write(_HOOK_TEMPLATE.format(executable=executable))
# NOTE(sigmavirus24): The following sets:
# - read, write, and execute permissions for the owner
# - read permissions for people in the group
# - read permissions for other people
# The owner needs the file to be readable, writable, and executable
# so that git can actually execute it as a hook.
pre_commit_permissions = stat.S_IRWXU | stat.S_IRGRP | stat.S_IROTH
os.chmod(pre_commit_file, pre_commit_permissions)
print("git pre-commit hook installed, for configuration options see")
print("http://flake8.pycqa.org/en/latest/user/using-hooks.html")
return True
def get_executable():
if sys.executable is not None:
return sys.executable
return "/usr/bin/env python"
def find_git_directory():
rev_parse = piped_process(["git", "rev-parse", "--git-dir"])
(stdout, _) = rev_parse.communicate()
stdout = to_text(stdout)
if rev_parse.returncode == 0:
return stdout.strip()
return None
def copy_indexed_files_to(temporary_directory, lazy):
# some plugins (e.g. flake8-isort) need these files to run their checks
setup_cfgs = find_setup_cfgs(lazy)
for filename in setup_cfgs:
contents = get_staged_contents_from(filename)
copy_file_to(temporary_directory, filename, contents)
modified_files = find_modified_files(lazy)
for filename in modified_files:
contents = get_staged_contents_from(filename)
yield copy_file_to(temporary_directory, filename, contents)
def copy_file_to(destination_directory, filepath, contents):
directory, filename = os.path.split(os.path.abspath(filepath))
temporary_directory = make_temporary_directory_from(
destination_directory, directory
)
if not os.path.exists(temporary_directory):
os.makedirs(temporary_directory)
temporary_filepath = os.path.join(temporary_directory, filename)
with open(temporary_filepath, "wb") as fd:
fd.write(contents)
return temporary_filepath
def make_temporary_directory_from(destination, directory):
prefix = os.path.commonprefix([directory, destination])
common_directory_path = os.path.relpath(directory, start=prefix)
return os.path.join(destination, common_directory_path)
def find_modified_files(lazy):
diff_index_cmd = [
"git",
"diff-index",
"--cached",
"--name-only",
"--diff-filter=ACMRTUXB",
"HEAD",
]
if lazy:
diff_index_cmd.remove("--cached")
diff_index = piped_process(diff_index_cmd)
(stdout, _) = diff_index.communicate()
stdout = to_text(stdout)
return stdout.splitlines()
def find_setup_cfgs(lazy):
setup_cfg_cmd = ["git", "ls-files", "--cached", "*setup.cfg"]
if lazy:
setup_cfg_cmd.remove("--cached")
extra_files = piped_process(setup_cfg_cmd)
(stdout, _) = extra_files.communicate()
stdout = to_text(stdout)
return stdout.splitlines()
def get_staged_contents_from(filename):
git_show = piped_process(["git", "show", ":{0}".format(filename)])
(stdout, _) = git_show.communicate()
return stdout
@contextlib.contextmanager
def make_temporary_directory():
temporary_directory = tempfile.mkdtemp()
yield temporary_directory
shutil.rmtree(temporary_directory, ignore_errors=True)
def to_text(string):
"""Ensure that the string is text."""
if callable(getattr(string, "decode", None)):
return string.decode("utf-8")
return string
def piped_process(command):
return subprocess.Popen(
command, stdout=subprocess.PIPE, stderr=subprocess.PIPE
)
def git_config_for(parameter):
config = piped_process(["git", "config", "--get", "--bool", parameter])
(stdout, _) = config.communicate()
return to_text(stdout).strip()
def config_for(parameter):
environment_variable = "flake8_{0}".format(parameter).upper()
git_variable = "flake8.{0}".format(parameter)
value = os.environ.get(environment_variable, git_config_for(git_variable))
return value.lower() in defaults.TRUTHY_VALUES
def update_excludes(exclude_list, temporary_directory_path):
return [
(temporary_directory_path + pattern)
if os.path.isabs(pattern)
else pattern
for pattern in exclude_list
]
def update_paths(checker_manager, temp_prefix):
temp_prefix_length = len(temp_prefix)
for checker in checker_manager.checkers:
filename = checker.display_name
if filename.startswith(temp_prefix):
checker.display_name = os.path.relpath(
filename[temp_prefix_length:]
)
_HOOK_TEMPLATE = """#!{executable}
import sys
from flake8.main import git
if __name__ == '__main__':
sys.exit(
git.hook(
strict=git.config_for('strict'),
lazy=git.config_for('lazy'),
)
)
"""

145
src/flake8/main/mercurial.py
View file

@ -1,145 +0,0 @@
"""Module containing the main mecurial hook interface and helpers.
.. autofunction:: hook
.. autofunction:: install
"""
import configparser
import os
import subprocess
from typing import Set
from flake8 import exceptions as exc
__all__ = ("hook", "install")
def hook(ui, repo, **kwargs):
"""Execute Flake8 on the repository provided by Mercurial.
To understand the parameters read more of the Mercurial documentation
around Hooks: https://www.mercurial-scm.org/wiki/Hook.
We avoid using the ``ui`` attribute because it can cause issues with
the GPL license that Mercurial is under. We don't import it, but we
avoid using it all the same.
"""
from flake8.main import application
hgrc = find_hgrc(create_if_missing=False)
if hgrc is None:
print("Cannot locate your root mercurial repository.")
raise SystemExit(True)
hgconfig = configparser_for(hgrc)
strict = hgconfig.get("flake8", "strict", fallback=True)
filenames = list(get_filenames_from(repo, kwargs))
app = application.Application()
app.initialize(filenames)
app.options._running_from_vcs = True
app.run_checks()
app.report()
if strict:
return app.result_count
return 0
def install():
"""Ensure that the mercurial hooks are installed.
This searches for the ``.hg/hgrc`` configuration file and will add commit
and qrefresh hooks to it, if they do not already exist.
It will also print a message to stdout about how to configure the hook.
:returns:
True if successful, False if the ``.hg/hgrc`` file doesn't exist.
:rtype:
bool
:raises:
flake8.exceptions.MercurialCommitHookAlreadyExists
:raises:
flake8.exceptions.MercurialQRefreshHookAlreadyExists
"""
hgrc = find_hgrc(create_if_missing=True)
if hgrc is None:
return False
hgconfig = configparser_for(hgrc)
if not hgconfig.has_section("hooks"):
hgconfig.add_section("hooks")
if hgconfig.has_option("hooks", "commit"):
raise exc.MercurialCommitHookAlreadyExists(
path=hgrc, value=hgconfig.get("hooks", "commit")
)
if hgconfig.has_option("hooks", "qrefresh"):
raise exc.MercurialQRefreshHookAlreadyExists(
path=hgrc, value=hgconfig.get("hooks", "qrefresh")
)
hgconfig.set("hooks", "commit", "python:flake8.main.mercurial.hook")
hgconfig.set("hooks", "qrefresh", "python:flake8.main.mercurial.hook")
if not hgconfig.has_section("flake8"):
hgconfig.add_section("flake8")
if not hgconfig.has_option("flake8", "strict"):
hgconfig.set("flake8", "strict", False)
with open(hgrc, "w") as fd:
hgconfig.write(fd)
print("mercurial hooks installed, for configuration options see")
print("http://flake8.pycqa.org/en/latest/user/using-hooks.html")
return True
def get_filenames_from(repository, kwargs):
seen_filenames = set() # type: Set[str]
node = kwargs["node"]
for revision in range(repository[node], len(repository)):
for filename in repository[revision].files():
full_filename = os.path.join(repository.root, filename)
have_seen_filename = full_filename in seen_filenames
filename_does_not_exist = not os.path.exists(full_filename)
if have_seen_filename or filename_does_not_exist:
continue
seen_filenames.add(full_filename)
if full_filename.endswith(".py"):
yield full_filename
def find_hgrc(create_if_missing=False):
root = subprocess.Popen(
["hg", "root"], stdout=subprocess.PIPE, stderr=subprocess.PIPE
)
(hg_directory, _) = root.communicate()
if callable(getattr(hg_directory, "decode", None)):
hg_directory = hg_directory.decode("utf-8")
if not os.path.isdir(hg_directory):
return None
hgrc = os.path.abspath(os.path.join(hg_directory, ".hg", "hgrc"))
if not os.path.exists(hgrc):
if create_if_missing:
open(hgrc, "w").close()
else:
return None
return hgrc
def configparser_for(path):
parser = configparser.ConfigParser(interpolation=None)
parser.read(path)
return parser

9
src/flake8/main/options.py
View file

@ -4,7 +4,6 @@ import functools
from flake8 import defaults
from flake8.main import debug
from flake8.main import vcs
def register_preliminary_options(parser):
@ -322,14 +321,6 @@ def register_default_options(option_manager):
help='Exit with status code "0" even if there are errors.',
)
add_option(
"--install-hook",
action=vcs.InstallAction,
choices=vcs.choices(),
help="Install a hook that is run prior to a commit for the supported "
"version control system.",
)
add_option(
"-j",
"--jobs",

48
src/flake8/main/vcs.py
View file

@ -1,48 +0,0 @@
"""Module containing some of the logic for our VCS installation logic."""
from __future__ import print_function
import argparse
import sys
from flake8 import exceptions as exc
from flake8.main import git
from flake8.main import mercurial
# NOTE(sigmavirus24): In the future, we may allow for VCS hooks to be defined
# as plugins, e.g., adding a flake8.vcs entry-point. In that case, this
# dictionary should disappear, and this module might contain more code for
# managing those bits (in conjunction with flake8.plugins.manager).
_INSTALLERS = {"git": git.install, "mercurial": mercurial.install}
class InstallAction(argparse.Action):
"""argparse action to run the hook installation."""
def __call__(self, parser, namespace, value, option_string=None):
"""Perform the argparse action for installing vcs hooks."""
installer = _INSTALLERS[value]
errored = False
successful = False
try:
successful = installer()
except exc.HookInstallationError as hook_error:
print(str(hook_error))
errored = True
if not successful:
print("Could not find the {0} directory".format(value))
print(
"\nWARNING: flake8 vcs hooks integration is deprecated and "
"scheduled for removal in 4.x. For more information, see "
"https://gitlab.com/pycqa/flake8/issues/568",
file=sys.stderr,
)
raise SystemExit(not successful and errored)
def choices():
"""Return the list of VCS choices."""
return list(_INSTALLERS)

7
src/flake8/plugins/pyflakes.py
View file

@ -2,13 +2,6 @@
# -*- coding: utf-8 -*-
from __future__ import absolute_import
try:
# The 'demandimport' breaks pyflakes and flake8.plugins.pyflakes
from mercurial import demandimport
except ImportError:
pass
else:
demandimport.disable()
import os
from typing import List