diff options
Diffstat (limited to 'pre_commit/repository.py')
| -rw-r--r-- | pre_commit/repository.py | 246 |
1 files changed, 246 insertions, 0 deletions
diff --git a/pre_commit/repository.py b/pre_commit/repository.py new file mode 100644 index 0000000..aa84185 --- /dev/null +++ b/pre_commit/repository.py @@ -0,0 +1,246 @@ +from __future__ import annotations + +import json +import logging +import os +import shlex +from collections.abc import Sequence +from typing import Any + +import pre_commit.constants as C +from pre_commit.all_languages import languages +from pre_commit.clientlib import load_manifest +from pre_commit.clientlib import LOCAL +from pre_commit.clientlib import META +from pre_commit.hook import Hook +from pre_commit.lang_base import environment_dir +from pre_commit.prefix import Prefix +from pre_commit.store import Store +from pre_commit.util import clean_path_on_failure +from pre_commit.util import rmtree + + +logger = logging.getLogger('pre_commit') + + +def _state_filename_v1(venv: str) -> str: + return os.path.join(venv, '.install_state_v1') + + +def _state_filename_v2(venv: str) -> str: + return os.path.join(venv, '.install_state_v2') + + +def _state(additional_deps: Sequence[str]) -> object: + return {'additional_dependencies': additional_deps} + + +def _read_state(venv: str) -> object | None: + filename = _state_filename_v1(venv) + if not os.path.exists(filename): + return None + else: + with open(filename) as f: + return json.load(f) + + +def _hook_installed(hook: Hook) -> bool: + lang = languages[hook.language] + if lang.ENVIRONMENT_DIR is None: + return True + + venv = environment_dir( + hook.prefix, + lang.ENVIRONMENT_DIR, + hook.language_version, + ) + return ( + ( + os.path.exists(_state_filename_v2(venv)) or + _read_state(venv) == _state(hook.additional_dependencies) + ) and + not lang.health_check(hook.prefix, hook.language_version) + ) + + +def _hook_install(hook: Hook) -> None: + logger.info(f'Installing environment for {hook.src}.') + logger.info('Once installed this environment will be reused.') + logger.info('This may take a few minutes...') + + if hook.language == 'python_venv': + logger.warning( + f'`repo: {hook.src}` uses deprecated `language: python_venv`. ' + f'This is an alias for `language: python`. ' + f'Often `pre-commit autoupdate --repo {shlex.quote(hook.src)}` ' + f'will fix this.', + ) + + lang = languages[hook.language] + assert lang.ENVIRONMENT_DIR is not None + + venv = environment_dir( + hook.prefix, + lang.ENVIRONMENT_DIR, + hook.language_version, + ) + + # There's potentially incomplete cleanup from previous runs + # Clean it up! + if os.path.exists(venv): + rmtree(venv) + + with clean_path_on_failure(venv): + lang.install_environment( + hook.prefix, hook.language_version, hook.additional_dependencies, + ) + health_error = lang.health_check(hook.prefix, hook.language_version) + if health_error: + raise AssertionError( + f'BUG: expected environment for {hook.language} to be healthy ' + f'immediately after install, please open an issue describing ' + f'your environment\n\n' + f'more info:\n\n{health_error}', + ) + + # TODO: remove v1 state writing, no longer needed after pre-commit 3.0 + # Write our state to indicate we're installed + state_filename = _state_filename_v1(venv) + staging = f'{state_filename}staging' + with open(staging, 'w') as state_file: + state_file.write(json.dumps(_state(hook.additional_dependencies))) + # Move the file into place atomically to indicate we've installed + os.replace(staging, state_filename) + + open(_state_filename_v2(venv), 'a+').close() + + +def _hook( + *hook_dicts: dict[str, Any], + root_config: dict[str, Any], +) -> dict[str, Any]: + ret, rest = dict(hook_dicts[0]), hook_dicts[1:] + for dct in rest: + ret.update(dct) + + lang = ret['language'] + if ret['language_version'] == C.DEFAULT: + ret['language_version'] = root_config['default_language_version'][lang] + if ret['language_version'] == C.DEFAULT: + ret['language_version'] = languages[lang].get_default_version() + + if not ret['stages']: + ret['stages'] = root_config['default_stages'] + + if languages[lang].ENVIRONMENT_DIR is None: + if ret['language_version'] != C.DEFAULT: + logger.error( + f'The hook `{ret["id"]}` specifies `language_version` but is ' + f'using language `{lang}` which does not install an ' + f'environment. ' + f'Perhaps you meant to use a specific language?', + ) + exit(1) + if ret['additional_dependencies']: + logger.error( + f'The hook `{ret["id"]}` specifies `additional_dependencies` ' + f'but is using language `{lang}` which does not install an ' + f'environment. ' + f'Perhaps you meant to use a specific language?', + ) + exit(1) + + return ret + + +def _non_cloned_repository_hooks( + repo_config: dict[str, Any], + store: Store, + root_config: dict[str, Any], +) -> tuple[Hook, ...]: + def _prefix(language_name: str, deps: Sequence[str]) -> Prefix: + language = languages[language_name] + # pygrep / script / system / docker_image do not have + # environments so they work out of the current directory + if language.ENVIRONMENT_DIR is None: + return Prefix(os.getcwd()) + else: + return Prefix(store.make_local(deps)) + + return tuple( + Hook.create( + repo_config['repo'], + _prefix(hook['language'], hook['additional_dependencies']), + _hook(hook, root_config=root_config), + ) + for hook in repo_config['hooks'] + ) + + +def _cloned_repository_hooks( + repo_config: dict[str, Any], + store: Store, + root_config: dict[str, Any], +) -> tuple[Hook, ...]: + repo, rev = repo_config['repo'], repo_config['rev'] + manifest_path = os.path.join(store.clone(repo, rev), C.MANIFEST_FILE) + by_id = {hook['id']: hook for hook in load_manifest(manifest_path)} + + for hook in repo_config['hooks']: + if hook['id'] not in by_id: + logger.error( + f'`{hook["id"]}` is not present in repository {repo}. ' + f'Typo? Perhaps it is introduced in a newer version? ' + f'Often `pre-commit autoupdate` fixes this.', + ) + exit(1) + + hook_dcts = [ + _hook(by_id[hook['id']], hook, root_config=root_config) + for hook in repo_config['hooks'] + ] + return tuple( + Hook.create( + repo_config['repo'], + Prefix(store.clone(repo, rev, hook['additional_dependencies'])), + hook, + ) + for hook in hook_dcts + ) + + +def _repository_hooks( + repo_config: dict[str, Any], + store: Store, + root_config: dict[str, Any], +) -> tuple[Hook, ...]: + if repo_config['repo'] in {LOCAL, META}: + return _non_cloned_repository_hooks(repo_config, store, root_config) + else: + return _cloned_repository_hooks(repo_config, store, root_config) + + +def install_hook_envs(hooks: Sequence[Hook], store: Store) -> None: + def _need_installed() -> list[Hook]: + seen: set[tuple[Prefix, str, str, tuple[str, ...]]] = set() + ret = [] + for hook in hooks: + if hook.install_key not in seen and not _hook_installed(hook): + ret.append(hook) + seen.add(hook.install_key) + return ret + + if not _need_installed(): + return + with store.exclusive_lock(): + # Another process may have already completed this work + for hook in _need_installed(): + _hook_install(hook) + + +def all_hooks(root_config: dict[str, Any], store: Store) -> tuple[Hook, ...]: + return tuple( + hook + for repo in root_config['repos'] + for hook in _repository_hooks(repo, store, root_config) + ) |