from __future__ import annotations
import contextlib
import json
import logging
import os.path
import sqlite3
import tempfile
from collections.abc import Callable
from collections.abc import Generator
from typing import Any
import pre_commit.constants as C
from pre_commit import clientlib
from pre_commit import file_lock
from pre_commit import git
from pre_commit.util import CalledProcessError
from pre_commit.util import clean_path_on_failure
from pre_commit.util import cmd_output_b
from pre_commit.util import resource_text
logger = logging.getLogger('pre_commit')
def _get_default_directory() -> str:
"""Returns the default directory for the Store. This is intentionally
underscored to indicate that `Store.get_default_directory` is the intended
way to get this information. This is also done so
`Store.get_default_directory` can be mocked in tests and
`_get_default_directory` can be tested.
"""
ret = os.environ.get('PRE_COMMIT_HOME') or os.path.join(
os.environ.get('XDG_CACHE_HOME') or os.path.expanduser('~/.cache'),
'pre-commit',
)
return os.path.realpath(ret)
_LOCAL_RESOURCES = (
'Cargo.toml', 'main.go', 'go.mod', 'main.rs', '.npmignore',
'package.json', 'pre-commit-package-dev-1.rockspec',
'pre_commit_placeholder_package.gemspec', 'setup.py',
'environment.yml', 'Makefile.PL', 'pubspec.yaml',
'renv.lock', 'renv/activate.R', 'renv/LICENSE.renv',
)
def _make_local_repo(directory: str) -> None:
for resource in _LOCAL_RESOURCES:
resource_dirname, resource_basename = os.path.split(resource)
contents = resource_text(f'empty_template_{resource_basename}')
target_dir = os.path.join(directory, resource_dirname)
target_file = os.path.join(target_dir, resource_basename)
os.makedirs(target_dir, exist_ok=True)
with open(target_file, 'w') as f:
f.write(contents)
class Store:
get_default_directory = staticmethod(_get_default_directory)
def __init__(self, directory: str | None = None) -> None:
self.directory = directory or Store.get_default_directory()
self.db_path = os.path.join(self.directory, 'db5.db')
self.readonly = (
os.path.exists(self.directory) and
not os.access(self.directory, os.W_OK)
)
if not os.path.exists(self.directory):
os.makedirs(self.directory, exist_ok=True)
with open(os.path.join(self.directory, 'README'), 'w') as f:
f.write(
'This directory is maintained by the pre-commit project.\n'
'Learn more: https://github.com/pre-commit/pre-commit\n',
)
if os.path.exists(self.db_path):
return
with self.exclusive_lock():
# Another process may have already completed this work
if os.path.exists(self.db_path): # pragma: no cover (race)
return
# To avoid a race where someone ^Cs between db creation and
# execution of the CREATE TABLE statements
fd, tmpfile = tempfile.mkstemp(dir=self.directory)
# We'll be managing this file ourselves
os.close(fd)
with self.connect(db_path=tmpfile) as db:
db.executescript(
'CREATE TABLE configs ('
' path TEXT NOT NULL,'
' PRIMARY KEY (path)'
');',
)
db.executescript(
'CREATE TABLE manifests ('
' repo TEXT NOT NULL,'
' rev TEXT NOT NULL,'
' manifest TEXT NOT NULL,'
' PRIMARY KEY (repo, rev)'
');',
)
db.executescript(
'CREATE TABLE clones ('
' repo TEXT NOT NULL,'
' rev TEXT NOT NULL,'
' path TEXT NOT NULL,'
' PRIMARY KEY (repo, rev)'
');',
)
db.executescript(
'CREATE TABLE installs ('
' key TEXT NOT NULL,'
' path TEXT NOT NULL,'
' PRIMARY KEY (key)'
');',
)
# Atomic file move
os.replace(tmpfile, self.db_path)
@contextlib.contextmanager
def exclusive_lock(self) -> Generator[None]:
def blocked_cb() -> None: # pragma: no cover (tests are in-process)
logger.info('Locking pre-commit directory')
with file_lock.lock(os.path.join(self.directory, '.lock'), blocked_cb):
yield
@contextlib.contextmanager
def connect(
self,
db_path: str | None = None,
) -> Generator[sqlite3.Connection]:
db_path = db_path or self.db_path
# sqlite doesn't close its fd with its contextmanager >.<
# contextlib.closing fixes this.
# See: https://stackoverflow.com/a/28032829/812183
with contextlib.closing(sqlite3.connect(db_path)) as db:
# this creates a transaction
with db:
yield db
def _complete_clone(self, rev: str, git_cmd: Callable[..., None]) -> None:
"""Perform a complete clone of a repository and its submodules """
git_cmd('fetch', 'origin', '--tags')
git_cmd('checkout', rev)
git_cmd('submodule', 'update', '--init', '--recursive')
def _shallow_clone(self, rev: str, git_cmd: Callable[..., None]) -> None:
"""Perform a shallow clone of a repository and its submodules """
v2 = ('-c', 'protocol.version=2')
git_cmd(*v2, 'fetch', 'origin', rev, '--depth=1')
git_cmd('checkout', 'FETCH_HEAD')
git_cmd(
*v2, 'submodule', 'update', '--init', '--recursive',
'--depth=1',
)
def clone(self, repo: str, rev: str) -> str:
def _get_result() -> str | None:
with self.connect() as db:
result = db.execute(
'SELECT path FROM clones WHERE repo = ? AND rev = ?',
(repo, rev),
).fetchone()
return result[0] if result else None
result = _get_result()
if result:
return result
with self.exclusive_lock():
# Another process may have already completed this work
result = _get_result()
if result: # pragma: no cover (race)
return result
logger.info(f'Cloning {repo}...')
directory = tempfile.mkdtemp(prefix='clone', dir=self.directory)
with clean_path_on_failure(directory):
git.init_repo(directory, repo)
env = git.no_git_env()
def _git_cmd(*args: str) -> None:
cmd_output_b('git', *args, cwd=directory, env=env)
try:
self._shallow_clone(rev, _git_cmd)
except CalledProcessError:
self._complete_clone(rev, _git_cmd)
manifest = clientlib.load_raw(
os.path.join(directory, C.MANIFEST_FILE),
display_filename=f'({repo})/{C.MANIFEST_FILE}',
exc_tp=clientlib.InvalidManifestError,
)
# Update our db with the created repo
with self.connect() as db:
db.execute(
'INSERT INTO clones VALUES (?, ?, ?)',
(repo, rev, directory),
)
db.execute(
'INSERT INTO manifests VALUES (?, ?, ?)',
(repo, rev, json.dumps(manifest)),
)
clientlib.warn_for_stages_on_repo_init(repo, directory)
return directory
def maybe_manifest(
self,
repo: str,
rev: str,
) -> dict[str, dict[str, Any]] | None:
with self.connect() as db:
result = db.execute(
'SELECT manifest FROM manifests WHERE repo = ? AND rev = ?',
(repo, rev),
).fetchone()
if result is not None:
return clientlib.load_manifest_contents(repo, result[0])
else:
return None
def manifest(self, repo: str, rev: str) -> dict[str, dict[str, Any]]:
ret = self.maybe_manifest(repo, rev)
if ret is None:
self.clone(repo, rev)
ret = self.maybe_manifest(repo, rev)
assert ret is not None
return ret
def mark_config_used(self, path: str) -> None:
if self.readonly: # pragma: win32 no cover
return
path = os.path.realpath(path)
# don't insert config files that do not exist
if not os.path.exists(path):
return
with self.connect() as db:
db.execute('INSERT OR IGNORE INTO configs VALUES (?)', (path,))