We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
special option to `pre-commit run` which works against a readonly filesystem
(mostly as a hardening measure for [pre-commit.ci](https://pre-commit.ci))

I believe the only current r/w operation on the store is to mark the configuration as used.  this could be skipped and installations could raise an error

(I might be missing something)

easiest reproduction is via docker:

```console
$ docker run --rm -ti -e PRE_COMMIT_HOME=/pc --workdir /src -v ~/workspace/pre-commit:/src:ro -v $PWD/pc:/pc:rw $IMG pre-commit install-hooks
[INFO] Initializing environment for https://github.com/pre-commit/pre-commit-hooks.
[INFO] Initializing environment for https://gitlab.com/pycqa/flake8.
[INFO] Initializing environment for https://gitlab.com/pycqa/flake8:flake8-typing-imports==1.6.0.
[INFO] Initializing environment for https://github.com/pre-commit/mirrors-autopep8.
[INFO] Initializing environment for https://github.com/pre-commit/pre-commit.
[INFO] Initializing environment for https://github.com/asottile/pyupgrade.
[INFO] Initializing environment for https://github.com/asottile/reorder_python_imports.
[INFO] Initializing environment for https://github.com/asottile/add-trailing-comma.
[INFO] Initializing environment for https://github.com/asottile/setup-cfg-fmt.
[INFO] Initializing environment for https://github.com/pre-commit/mirrors-mypy.
[INFO] Installing environment for https://github.com/pre-commit/pre-commit-hooks.
[INFO] Once installed this environment will be reused.
[INFO] This may take a few minutes...
[INFO] Installing environment for https://gitlab.com/pycqa/flake8.
[INFO] Once installed this environment will be reused.
[INFO] This may take a few minutes...
[INFO] Installing environment for https://github.com/pre-commit/mirrors-autopep8.
[INFO] Once installed this environment will be reused.
[INFO] This may take a few minutes...
[INFO] Installing environment for https://github.com/pre-commit/pre-commit.
[INFO] Once installed this environment will be reused.
[INFO] This may take a few minutes...
[INFO] Installing environment for https://github.com/asottile/pyupgrade.
[INFO] Once installed this environment will be reused.
[INFO] This may take a few minutes...
[INFO] Installing environment for https://github.com/asottile/reorder_python_imports.
[INFO] Once installed this environment will be reused.
[INFO] This may take a few minutes...
[INFO] Installing environment for https://github.com/asottile/add-trailing-comma.
[INFO] Once installed this environment will be reused.
[INFO] This may take a few minutes...
[INFO] Installing environment for https://github.com/asottile/setup-cfg-fmt.
[INFO] Once installed this environment will be reused.
[INFO] This may take a few minutes...
[INFO] Installing environment for https://github.com/pre-commit/mirrors-mypy.
[INFO] Once installed this environment will be reused.
[INFO] This may take a few minutes...
$ docker run --rm -ti -e PRE_COMMIT_HOME=/pc --workdir /src -v ~/workspace/pre-commit:/src:ro -v $PWD/pc:/pc:ro $IMG pre-commit run --all-files
An unexpected error has occurred: OperationalError: attempt to write a readonly database
Check the log at /pc/pre-commit.log
Traceback (most recent call last):
  File "/venv/lib/python3.8/site-packages/pre_commit/error_handler.py", line 56, in error_handler
    yield
  File "/venv/lib/python3.8/site-packages/pre_commit/main.py", line 351, in main
    store.mark_config_used(args.config)
  File "/venv/lib/python3.8/site-packages/pre_commit/store.py", line 228, in mark_config_used
    db.execute('INSERT OR IGNORE INTO configs VALUES (?)', (path,))
sqlite3.OperationalError: attempt to write a readonly database

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/venv/bin/pre-commit", line 8, in <module>
    sys.exit(main())
  File "/venv/lib/python3.8/site-packages/pre_commit/main.py", line 404, in main
    raise AssertionError(
  File "/usr/lib/python3.8/contextlib.py", line 131, in __exit__
    self.gen.throw(type, value, traceback)
  File "/venv/lib/python3.8/site-packages/pre_commit/error_handler.py", line 64, in error_handler
    _log_and_exit(msg, e, traceback.format_exc())
  File "/venv/lib/python3.8/site-packages/pre_commit/error_handler.py", line 24, in _log_and_exit
    with open(log_path, 'wb') as log:
OSError: [Errno 30] Read-only file system: '/pc/pre-commit.log'
```

may also need another place to put `pre-commit.log` as well 🤔 

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `pre_commit/store.py`
Content:
```
1 import contextlib
2 import logging
3 import os.path
4 import sqlite3
5 import tempfile
6 from typing import Callable
7 from typing import Generator
8 from typing import List
9 from typing import Optional
10 from typing import Sequence
11 from typing import Tuple
12 
13 import pre_commit.constants as C
14 from pre_commit import file_lock
15 from pre_commit import git
16 from pre_commit.util import CalledProcessError
17 from pre_commit.util import clean_path_on_failure
18 from pre_commit.util import cmd_output_b
19 from pre_commit.util import resource_text
20 from pre_commit.util import rmtree
21 
22 
23 logger = logging.getLogger('pre_commit')
24 
25 
26 def _get_default_directory() -> str:
27     """Returns the default directory for the Store.  This is intentionally
28     underscored to indicate that `Store.get_default_directory` is the intended
29     way to get this information.  This is also done so
30     `Store.get_default_directory` can be mocked in tests and
31     `_get_default_directory` can be tested.
32     """
33     ret = os.environ.get('PRE_COMMIT_HOME') or os.path.join(
34         os.environ.get('XDG_CACHE_HOME') or os.path.expanduser('~/.cache'),
35         'pre-commit',
36     )
37     return os.path.realpath(ret)
38 
39 
40 class Store:
41     get_default_directory = staticmethod(_get_default_directory)
42 
43     def __init__(self, directory: Optional[str] = None) -> None:
44         self.directory = directory or Store.get_default_directory()
45         self.db_path = os.path.join(self.directory, 'db.db')
46 
47         if not os.path.exists(self.directory):
48             os.makedirs(self.directory, exist_ok=True)
49             with open(os.path.join(self.directory, 'README'), 'w') as f:
50                 f.write(
51                     'This directory is maintained by the pre-commit project.\n'
52                     'Learn more: https://github.com/pre-commit/pre-commit\n',
53                 )
54 
55         if os.path.exists(self.db_path):
56             return
57         with self.exclusive_lock():
58             # Another process may have already completed this work
59             if os.path.exists(self.db_path):  # pragma: no cover (race)
60                 return
61             # To avoid a race where someone ^Cs between db creation and
62             # execution of the CREATE TABLE statement
63             fd, tmpfile = tempfile.mkstemp(dir=self.directory)
64             # We'll be managing this file ourselves
65             os.close(fd)
66             with self.connect(db_path=tmpfile) as db:
67                 db.executescript(
68                     'CREATE TABLE repos ('
69                     '    repo TEXT NOT NULL,'
70                     '    ref TEXT NOT NULL,'
71                     '    path TEXT NOT NULL,'
72                     '    PRIMARY KEY (repo, ref)'
73                     ');',
74                 )
75                 self._create_config_table(db)
76 
77             # Atomic file move
78             os.rename(tmpfile, self.db_path)
79 
80     @contextlib.contextmanager
81     def exclusive_lock(self) -> Generator[None, None, None]:
82         def blocked_cb() -> None:  # pragma: no cover (tests are in-process)
83             logger.info('Locking pre-commit directory')
84 
85         with file_lock.lock(os.path.join(self.directory, '.lock'), blocked_cb):
86             yield
87 
88     @contextlib.contextmanager
89     def connect(
90             self,
91             db_path: Optional[str] = None,
92     ) -> Generator[sqlite3.Connection, None, None]:
93         db_path = db_path or self.db_path
94         # sqlite doesn't close its fd with its contextmanager >.<
95         # contextlib.closing fixes this.
96         # See: https://stackoverflow.com/a/28032829/812183
97         with contextlib.closing(sqlite3.connect(db_path)) as db:
98             # this creates a transaction
99             with db:
100                 yield db
101 
102     @classmethod
103     def db_repo_name(cls, repo: str, deps: Sequence[str]) -> str:
104         if deps:
105             return f'{repo}:{",".join(sorted(deps))}'
106         else:
107             return repo
108 
109     def _new_repo(
110             self,
111             repo: str,
112             ref: str,
113             deps: Sequence[str],
114             make_strategy: Callable[[str], None],
115     ) -> str:
116         repo = self.db_repo_name(repo, deps)
117 
118         def _get_result() -> Optional[str]:
119             # Check if we already exist
120             with self.connect() as db:
121                 result = db.execute(
122                     'SELECT path FROM repos WHERE repo = ? AND ref = ?',
123                     (repo, ref),
124                 ).fetchone()
125                 return result[0] if result else None
126 
127         result = _get_result()
128         if result:
129             return result
130         with self.exclusive_lock():
131             # Another process may have already completed this work
132             result = _get_result()
133             if result:  # pragma: no cover (race)
134                 return result
135 
136             logger.info(f'Initializing environment for {repo}.')
137 
138             directory = tempfile.mkdtemp(prefix='repo', dir=self.directory)
139             with clean_path_on_failure(directory):
140                 make_strategy(directory)
141 
142             # Update our db with the created repo
143             with self.connect() as db:
144                 db.execute(
145                     'INSERT INTO repos (repo, ref, path) VALUES (?, ?, ?)',
146                     [repo, ref, directory],
147                 )
148         return directory
149 
150     def _complete_clone(self, ref: str, git_cmd: Callable[..., None]) -> None:
151         """Perform a complete clone of a repository and its submodules """
152 
153         git_cmd('fetch', 'origin', '--tags')
154         git_cmd('checkout', ref)
155         git_cmd('submodule', 'update', '--init', '--recursive')
156 
157     def _shallow_clone(self, ref: str, git_cmd: Callable[..., None]) -> None:
158         """Perform a shallow clone of a repository and its submodules """
159 
160         git_config = 'protocol.version=2'
161         git_cmd('-c', git_config, 'fetch', 'origin', ref, '--depth=1')
162         git_cmd('checkout', 'FETCH_HEAD')
163         git_cmd(
164             '-c', git_config, 'submodule', 'update', '--init', '--recursive',
165             '--depth=1',
166         )
167 
168     def clone(self, repo: str, ref: str, deps: Sequence[str] = ()) -> str:
169         """Clone the given url and checkout the specific ref."""
170 
171         def clone_strategy(directory: str) -> None:
172             git.init_repo(directory, repo)
173             env = git.no_git_env()
174 
175             def _git_cmd(*args: str) -> None:
176                 cmd_output_b('git', *args, cwd=directory, env=env)
177 
178             try:
179                 self._shallow_clone(ref, _git_cmd)
180             except CalledProcessError:
181                 self._complete_clone(ref, _git_cmd)
182 
183         return self._new_repo(repo, ref, deps, clone_strategy)
184 
185     LOCAL_RESOURCES = (
186         'Cargo.toml', 'main.go', 'go.mod', 'main.rs', '.npmignore',
187         'package.json', 'pre_commit_dummy_package.gemspec', 'setup.py',
188         'environment.yml', 'Makefile.PL',
189     )
190 
191     def make_local(self, deps: Sequence[str]) -> str:
192         def make_local_strategy(directory: str) -> None:
193             for resource in self.LOCAL_RESOURCES:
194                 contents = resource_text(f'empty_template_{resource}')
195                 with open(os.path.join(directory, resource), 'w') as f:
196                     f.write(contents)
197 
198             env = git.no_git_env()
199 
200             # initialize the git repository so it looks more like cloned repos
201             def _git_cmd(*args: str) -> None:
202                 cmd_output_b('git', *args, cwd=directory, env=env)
203 
204             git.init_repo(directory, '<<unknown>>')
205             _git_cmd('add', '.')
206             git.commit(repo=directory)
207 
208         return self._new_repo(
209             'local', C.LOCAL_REPO_VERSION, deps, make_local_strategy,
210         )
211 
212     def _create_config_table(self, db: sqlite3.Connection) -> None:
213         db.executescript(
214             'CREATE TABLE IF NOT EXISTS configs ('
215             '   path TEXT NOT NULL,'
216             '   PRIMARY KEY (path)'
217             ');',
218         )
219 
220     def mark_config_used(self, path: str) -> None:
221         path = os.path.realpath(path)
222         # don't insert config files that do not exist
223         if not os.path.exists(path):
224             return
225         with self.connect() as db:
226             # TODO: eventually remove this and only create in _create
227             self._create_config_table(db)
228             db.execute('INSERT OR IGNORE INTO configs VALUES (?)', (path,))
229 
230     def select_all_configs(self) -> List[str]:
231         with self.connect() as db:
232             self._create_config_table(db)
233             rows = db.execute('SELECT path FROM configs').fetchall()
234             return [path for path, in rows]
235 
236     def delete_configs(self, configs: List[str]) -> None:
237         with self.connect() as db:
238             rows = [(path,) for path in configs]
239             db.executemany('DELETE FROM configs WHERE path = ?', rows)
240 
241     def select_all_repos(self) -> List[Tuple[str, str, str]]:
242         with self.connect() as db:
243             return db.execute('SELECT repo, ref, path from repos').fetchall()
244 
245     def delete_repo(self, db_repo_name: str, ref: str, path: str) -> None:
246         with self.connect() as db:
247             db.execute(
248                 'DELETE FROM repos WHERE repo = ? and ref = ?',
249                 (db_repo_name, ref),
250             )
251         rmtree(path)
252
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/pre_commit/store.py b/pre_commit/store.py
--- a/pre_commit/store.py
+++ b/pre_commit/store.py
@@ -43,6 +43,10 @@
     def __init__(self, directory: Optional[str] = None) -> None:
         self.directory = directory or Store.get_default_directory()
         self.db_path = os.path.join(self.directory, 'db.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)
@@ -218,6 +222,8 @@
         )
 
     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):

{"golden_diff": "diff --git a/pre_commit/store.py b/pre_commit/store.py\n--- a/pre_commit/store.py\n+++ b/pre_commit/store.py\n@@ -43,6 +43,10 @@\n     def __init__(self, directory: Optional[str] = None) -> None:\n         self.directory = directory or Store.get_default_directory()\n         self.db_path = os.path.join(self.directory, 'db.db')\n+        self.readonly = (\n+            os.path.exists(self.directory) and\n+            not os.access(self.directory, os.W_OK)\n+        )\n \n         if not os.path.exists(self.directory):\n             os.makedirs(self.directory, exist_ok=True)\n@@ -218,6 +222,8 @@\n         )\n \n     def mark_config_used(self, path: str) -> None:\n+        if self.readonly:  # pragma: win32 no cover\n+            return\n         path = os.path.realpath(path)\n         # don't insert config files that do not exist\n         if not os.path.exists(path):\n", "issue": "special option to `pre-commit run` which works against a readonly filesystem\n(mostly as a hardening measure for [pre-commit.ci](https://pre-commit.ci))\r\n\r\nI believe the only current r/w operation on the store is to mark the configuration as used.  this could be skipped and installations could raise an error\r\n\r\n(I might be missing something)\r\n\r\neasiest reproduction is via docker:\r\n\r\n```console\r\n$ docker run --rm -ti -e PRE_COMMIT_HOME=/pc --workdir /src -v ~/workspace/pre-commit:/src:ro -v $PWD/pc:/pc:rw $IMG pre-commit install-hooks\r\n[INFO] Initializing environment for https://github.com/pre-commit/pre-commit-hooks.\r\n[INFO] Initializing environment for https://gitlab.com/pycqa/flake8.\r\n[INFO] Initializing environment for https://gitlab.com/pycqa/flake8:flake8-typing-imports==1.6.0.\r\n[INFO] Initializing environment for https://github.com/pre-commit/mirrors-autopep8.\r\n[INFO] Initializing environment for https://github.com/pre-commit/pre-commit.\r\n[INFO] Initializing environment for https://github.com/asottile/pyupgrade.\r\n[INFO] Initializing environment for https://github.com/asottile/reorder_python_imports.\r\n[INFO] Initializing environment for https://github.com/asottile/add-trailing-comma.\r\n[INFO] Initializing environment for https://github.com/asottile/setup-cfg-fmt.\r\n[INFO] Initializing environment for https://github.com/pre-commit/mirrors-mypy.\r\n[INFO] Installing environment for https://github.com/pre-commit/pre-commit-hooks.\r\n[INFO] Once installed this environment will be reused.\r\n[INFO] This may take a few minutes...\r\n[INFO] Installing environment for https://gitlab.com/pycqa/flake8.\r\n[INFO] Once installed this environment will be reused.\r\n[INFO] This may take a few minutes...\r\n[INFO] Installing environment for https://github.com/pre-commit/mirrors-autopep8.\r\n[INFO] Once installed this environment will be reused.\r\n[INFO] This may take a few minutes...\r\n[INFO] Installing environment for https://github.com/pre-commit/pre-commit.\r\n[INFO] Once installed this environment will be reused.\r\n[INFO] This may take a few minutes...\r\n[INFO] Installing environment for https://github.com/asottile/pyupgrade.\r\n[INFO] Once installed this environment will be reused.\r\n[INFO] This may take a few minutes...\r\n[INFO] Installing environment for https://github.com/asottile/reorder_python_imports.\r\n[INFO] Once installed this environment will be reused.\r\n[INFO] This may take a few minutes...\r\n[INFO] Installing environment for https://github.com/asottile/add-trailing-comma.\r\n[INFO] Once installed this environment will be reused.\r\n[INFO] This may take a few minutes...\r\n[INFO] Installing environment for https://github.com/asottile/setup-cfg-fmt.\r\n[INFO] Once installed this environment will be reused.\r\n[INFO] This may take a few minutes...\r\n[INFO] Installing environment for https://github.com/pre-commit/mirrors-mypy.\r\n[INFO] Once installed this environment will be reused.\r\n[INFO] This may take a few minutes...\r\n$ docker run --rm -ti -e PRE_COMMIT_HOME=/pc --workdir /src -v ~/workspace/pre-commit:/src:ro -v $PWD/pc:/pc:ro $IMG pre-commit run --all-files\r\nAn unexpected error has occurred: OperationalError: attempt to write a readonly database\r\nCheck the log at /pc/pre-commit.log\r\nTraceback (most recent call last):\r\n  File \"/venv/lib/python3.8/site-packages/pre_commit/error_handler.py\", line 56, in error_handler\r\n    yield\r\n  File \"/venv/lib/python3.8/site-packages/pre_commit/main.py\", line 351, in main\r\n    store.mark_config_used(args.config)\r\n  File \"/venv/lib/python3.8/site-packages/pre_commit/store.py\", line 228, in mark_config_used\r\n    db.execute('INSERT OR IGNORE INTO configs VALUES (?)', (path,))\r\nsqlite3.OperationalError: attempt to write a readonly database\r\n\r\nDuring handling of the above exception, another exception occurred:\r\n\r\nTraceback (most recent call last):\r\n  File \"/venv/bin/pre-commit\", line 8, in <module>\r\n    sys.exit(main())\r\n  File \"/venv/lib/python3.8/site-packages/pre_commit/main.py\", line 404, in main\r\n    raise AssertionError(\r\n  File \"/usr/lib/python3.8/contextlib.py\", line 131, in __exit__\r\n    self.gen.throw(type, value, traceback)\r\n  File \"/venv/lib/python3.8/site-packages/pre_commit/error_handler.py\", line 64, in error_handler\r\n    _log_and_exit(msg, e, traceback.format_exc())\r\n  File \"/venv/lib/python3.8/site-packages/pre_commit/error_handler.py\", line 24, in _log_and_exit\r\n    with open(log_path, 'wb') as log:\r\nOSError: [Errno 30] Read-only file system: '/pc/pre-commit.log'\r\n```\r\n\r\nmay also need another place to put `pre-commit.log` as well \ud83e\udd14 \n", "before_files": [{"content": "import contextlib\nimport logging\nimport os.path\nimport sqlite3\nimport tempfile\nfrom typing import Callable\nfrom typing import Generator\nfrom typing import List\nfrom typing import Optional\nfrom typing import Sequence\nfrom typing import Tuple\n\nimport pre_commit.constants as C\nfrom pre_commit import file_lock\nfrom pre_commit import git\nfrom pre_commit.util import CalledProcessError\nfrom pre_commit.util import clean_path_on_failure\nfrom pre_commit.util import cmd_output_b\nfrom pre_commit.util import resource_text\nfrom pre_commit.util import rmtree\n\n\nlogger = logging.getLogger('pre_commit')\n\n\ndef _get_default_directory() -> str:\n    \"\"\"Returns the default directory for the Store.  This is intentionally\n    underscored to indicate that `Store.get_default_directory` is the intended\n    way to get this information.  This is also done so\n    `Store.get_default_directory` can be mocked in tests and\n    `_get_default_directory` can be tested.\n    \"\"\"\n    ret = os.environ.get('PRE_COMMIT_HOME') or os.path.join(\n        os.environ.get('XDG_CACHE_HOME') or os.path.expanduser('~/.cache'),\n        'pre-commit',\n    )\n    return os.path.realpath(ret)\n\n\nclass Store:\n    get_default_directory = staticmethod(_get_default_directory)\n\n    def __init__(self, directory: Optional[str] = None) -> None:\n        self.directory = directory or Store.get_default_directory()\n        self.db_path = os.path.join(self.directory, 'db.db')\n\n        if not os.path.exists(self.directory):\n            os.makedirs(self.directory, exist_ok=True)\n            with open(os.path.join(self.directory, 'README'), 'w') as f:\n                f.write(\n                    'This directory is maintained by the pre-commit project.\\n'\n                    'Learn more: https://github.com/pre-commit/pre-commit\\n',\n                )\n\n        if os.path.exists(self.db_path):\n            return\n        with self.exclusive_lock():\n            # Another process may have already completed this work\n            if os.path.exists(self.db_path):  # pragma: no cover (race)\n                return\n            # To avoid a race where someone ^Cs between db creation and\n            # execution of the CREATE TABLE statement\n            fd, tmpfile = tempfile.mkstemp(dir=self.directory)\n            # We'll be managing this file ourselves\n            os.close(fd)\n            with self.connect(db_path=tmpfile) as db:\n                db.executescript(\n                    'CREATE TABLE repos ('\n                    '    repo TEXT NOT NULL,'\n                    '    ref TEXT NOT NULL,'\n                    '    path TEXT NOT NULL,'\n                    '    PRIMARY KEY (repo, ref)'\n                    ');',\n                )\n                self._create_config_table(db)\n\n            # Atomic file move\n            os.rename(tmpfile, self.db_path)\n\n    @contextlib.contextmanager\n    def exclusive_lock(self) -> Generator[None, None, None]:\n        def blocked_cb() -> None:  # pragma: no cover (tests are in-process)\n            logger.info('Locking pre-commit directory')\n\n        with file_lock.lock(os.path.join(self.directory, '.lock'), blocked_cb):\n            yield\n\n    @contextlib.contextmanager\n    def connect(\n            self,\n            db_path: Optional[str] = None,\n    ) -> Generator[sqlite3.Connection, None, None]:\n        db_path = db_path or self.db_path\n        # sqlite doesn't close its fd with its contextmanager >.<\n        # contextlib.closing fixes this.\n        # See: https://stackoverflow.com/a/28032829/812183\n        with contextlib.closing(sqlite3.connect(db_path)) as db:\n            # this creates a transaction\n            with db:\n                yield db\n\n    @classmethod\n    def db_repo_name(cls, repo: str, deps: Sequence[str]) -> str:\n        if deps:\n            return f'{repo}:{\",\".join(sorted(deps))}'\n        else:\n            return repo\n\n    def _new_repo(\n            self,\n            repo: str,\n            ref: str,\n            deps: Sequence[str],\n            make_strategy: Callable[[str], None],\n    ) -> str:\n        repo = self.db_repo_name(repo, deps)\n\n        def _get_result() -> Optional[str]:\n            # Check if we already exist\n            with self.connect() as db:\n                result = db.execute(\n                    'SELECT path FROM repos WHERE repo = ? AND ref = ?',\n                    (repo, ref),\n                ).fetchone()\n                return result[0] if result else None\n\n        result = _get_result()\n        if result:\n            return result\n        with self.exclusive_lock():\n            # Another process may have already completed this work\n            result = _get_result()\n            if result:  # pragma: no cover (race)\n                return result\n\n            logger.info(f'Initializing environment for {repo}.')\n\n            directory = tempfile.mkdtemp(prefix='repo', dir=self.directory)\n            with clean_path_on_failure(directory):\n                make_strategy(directory)\n\n            # Update our db with the created repo\n            with self.connect() as db:\n                db.execute(\n                    'INSERT INTO repos (repo, ref, path) VALUES (?, ?, ?)',\n                    [repo, ref, directory],\n                )\n        return directory\n\n    def _complete_clone(self, ref: str, git_cmd: Callable[..., None]) -> None:\n        \"\"\"Perform a complete clone of a repository and its submodules \"\"\"\n\n        git_cmd('fetch', 'origin', '--tags')\n        git_cmd('checkout', ref)\n        git_cmd('submodule', 'update', '--init', '--recursive')\n\n    def _shallow_clone(self, ref: str, git_cmd: Callable[..., None]) -> None:\n        \"\"\"Perform a shallow clone of a repository and its submodules \"\"\"\n\n        git_config = 'protocol.version=2'\n        git_cmd('-c', git_config, 'fetch', 'origin', ref, '--depth=1')\n        git_cmd('checkout', 'FETCH_HEAD')\n        git_cmd(\n            '-c', git_config, 'submodule', 'update', '--init', '--recursive',\n            '--depth=1',\n        )\n\n    def clone(self, repo: str, ref: str, deps: Sequence[str] = ()) -> str:\n        \"\"\"Clone the given url and checkout the specific ref.\"\"\"\n\n        def clone_strategy(directory: str) -> None:\n            git.init_repo(directory, repo)\n            env = git.no_git_env()\n\n            def _git_cmd(*args: str) -> None:\n                cmd_output_b('git', *args, cwd=directory, env=env)\n\n            try:\n                self._shallow_clone(ref, _git_cmd)\n            except CalledProcessError:\n                self._complete_clone(ref, _git_cmd)\n\n        return self._new_repo(repo, ref, deps, clone_strategy)\n\n    LOCAL_RESOURCES = (\n        'Cargo.toml', 'main.go', 'go.mod', 'main.rs', '.npmignore',\n        'package.json', 'pre_commit_dummy_package.gemspec', 'setup.py',\n        'environment.yml', 'Makefile.PL',\n    )\n\n    def make_local(self, deps: Sequence[str]) -> str:\n        def make_local_strategy(directory: str) -> None:\n            for resource in self.LOCAL_RESOURCES:\n                contents = resource_text(f'empty_template_{resource}')\n                with open(os.path.join(directory, resource), 'w') as f:\n                    f.write(contents)\n\n            env = git.no_git_env()\n\n            # initialize the git repository so it looks more like cloned repos\n            def _git_cmd(*args: str) -> None:\n                cmd_output_b('git', *args, cwd=directory, env=env)\n\n            git.init_repo(directory, '<<unknown>>')\n            _git_cmd('add', '.')\n            git.commit(repo=directory)\n\n        return self._new_repo(\n            'local', C.LOCAL_REPO_VERSION, deps, make_local_strategy,\n        )\n\n    def _create_config_table(self, db: sqlite3.Connection) -> None:\n        db.executescript(\n            'CREATE TABLE IF NOT EXISTS configs ('\n            '   path TEXT NOT NULL,'\n            '   PRIMARY KEY (path)'\n            ');',\n        )\n\n    def mark_config_used(self, path: str) -> None:\n        path = os.path.realpath(path)\n        # don't insert config files that do not exist\n        if not os.path.exists(path):\n            return\n        with self.connect() as db:\n            # TODO: eventually remove this and only create in _create\n            self._create_config_table(db)\n            db.execute('INSERT OR IGNORE INTO configs VALUES (?)', (path,))\n\n    def select_all_configs(self) -> List[str]:\n        with self.connect() as db:\n            self._create_config_table(db)\n            rows = db.execute('SELECT path FROM configs').fetchall()\n            return [path for path, in rows]\n\n    def delete_configs(self, configs: List[str]) -> None:\n        with self.connect() as db:\n            rows = [(path,) for path in configs]\n            db.executemany('DELETE FROM configs WHERE path = ?', rows)\n\n    def select_all_repos(self) -> List[Tuple[str, str, str]]:\n        with self.connect() as db:\n            return db.execute('SELECT repo, ref, path from repos').fetchall()\n\n    def delete_repo(self, db_repo_name: str, ref: str, path: str) -> None:\n        with self.connect() as db:\n            db.execute(\n                'DELETE FROM repos WHERE repo = ? and ref = ?',\n                (db_repo_name, ref),\n            )\n        rmtree(path)\n", "path": "pre_commit/store.py"}], "after_files": [{"content": "import contextlib\nimport logging\nimport os.path\nimport sqlite3\nimport tempfile\nfrom typing import Callable\nfrom typing import Generator\nfrom typing import List\nfrom typing import Optional\nfrom typing import Sequence\nfrom typing import Tuple\n\nimport pre_commit.constants as C\nfrom pre_commit import file_lock\nfrom pre_commit import git\nfrom pre_commit.util import CalledProcessError\nfrom pre_commit.util import clean_path_on_failure\nfrom pre_commit.util import cmd_output_b\nfrom pre_commit.util import resource_text\nfrom pre_commit.util import rmtree\n\n\nlogger = logging.getLogger('pre_commit')\n\n\ndef _get_default_directory() -> str:\n    \"\"\"Returns the default directory for the Store.  This is intentionally\n    underscored to indicate that `Store.get_default_directory` is the intended\n    way to get this information.  This is also done so\n    `Store.get_default_directory` can be mocked in tests and\n    `_get_default_directory` can be tested.\n    \"\"\"\n    ret = os.environ.get('PRE_COMMIT_HOME') or os.path.join(\n        os.environ.get('XDG_CACHE_HOME') or os.path.expanduser('~/.cache'),\n        'pre-commit',\n    )\n    return os.path.realpath(ret)\n\n\nclass Store:\n    get_default_directory = staticmethod(_get_default_directory)\n\n    def __init__(self, directory: Optional[str] = None) -> None:\n        self.directory = directory or Store.get_default_directory()\n        self.db_path = os.path.join(self.directory, 'db.db')\n        self.readonly = (\n            os.path.exists(self.directory) and\n            not os.access(self.directory, os.W_OK)\n        )\n\n        if not os.path.exists(self.directory):\n            os.makedirs(self.directory, exist_ok=True)\n            with open(os.path.join(self.directory, 'README'), 'w') as f:\n                f.write(\n                    'This directory is maintained by the pre-commit project.\\n'\n                    'Learn more: https://github.com/pre-commit/pre-commit\\n',\n                )\n\n        if os.path.exists(self.db_path):\n            return\n        with self.exclusive_lock():\n            # Another process may have already completed this work\n            if os.path.exists(self.db_path):  # pragma: no cover (race)\n                return\n            # To avoid a race where someone ^Cs between db creation and\n            # execution of the CREATE TABLE statement\n            fd, tmpfile = tempfile.mkstemp(dir=self.directory)\n            # We'll be managing this file ourselves\n            os.close(fd)\n            with self.connect(db_path=tmpfile) as db:\n                db.executescript(\n                    'CREATE TABLE repos ('\n                    '    repo TEXT NOT NULL,'\n                    '    ref TEXT NOT NULL,'\n                    '    path TEXT NOT NULL,'\n                    '    PRIMARY KEY (repo, ref)'\n                    ');',\n                )\n                self._create_config_table(db)\n\n            # Atomic file move\n            os.rename(tmpfile, self.db_path)\n\n    @contextlib.contextmanager\n    def exclusive_lock(self) -> Generator[None, None, None]:\n        def blocked_cb() -> None:  # pragma: no cover (tests are in-process)\n            logger.info('Locking pre-commit directory')\n\n        with file_lock.lock(os.path.join(self.directory, '.lock'), blocked_cb):\n            yield\n\n    @contextlib.contextmanager\n    def connect(\n            self,\n            db_path: Optional[str] = None,\n    ) -> Generator[sqlite3.Connection, None, None]:\n        db_path = db_path or self.db_path\n        # sqlite doesn't close its fd with its contextmanager >.<\n        # contextlib.closing fixes this.\n        # See: https://stackoverflow.com/a/28032829/812183\n        with contextlib.closing(sqlite3.connect(db_path)) as db:\n            # this creates a transaction\n            with db:\n                yield db\n\n    @classmethod\n    def db_repo_name(cls, repo: str, deps: Sequence[str]) -> str:\n        if deps:\n            return f'{repo}:{\",\".join(sorted(deps))}'\n        else:\n            return repo\n\n    def _new_repo(\n            self,\n            repo: str,\n            ref: str,\n            deps: Sequence[str],\n            make_strategy: Callable[[str], None],\n    ) -> str:\n        repo = self.db_repo_name(repo, deps)\n\n        def _get_result() -> Optional[str]:\n            # Check if we already exist\n            with self.connect() as db:\n                result = db.execute(\n                    'SELECT path FROM repos WHERE repo = ? AND ref = ?',\n                    (repo, ref),\n                ).fetchone()\n                return result[0] if result else None\n\n        result = _get_result()\n        if result:\n            return result\n        with self.exclusive_lock():\n            # Another process may have already completed this work\n            result = _get_result()\n            if result:  # pragma: no cover (race)\n                return result\n\n            logger.info(f'Initializing environment for {repo}.')\n\n            directory = tempfile.mkdtemp(prefix='repo', dir=self.directory)\n            with clean_path_on_failure(directory):\n                make_strategy(directory)\n\n            # Update our db with the created repo\n            with self.connect() as db:\n                db.execute(\n                    'INSERT INTO repos (repo, ref, path) VALUES (?, ?, ?)',\n                    [repo, ref, directory],\n                )\n        return directory\n\n    def _complete_clone(self, ref: str, git_cmd: Callable[..., None]) -> None:\n        \"\"\"Perform a complete clone of a repository and its submodules \"\"\"\n\n        git_cmd('fetch', 'origin', '--tags')\n        git_cmd('checkout', ref)\n        git_cmd('submodule', 'update', '--init', '--recursive')\n\n    def _shallow_clone(self, ref: str, git_cmd: Callable[..., None]) -> None:\n        \"\"\"Perform a shallow clone of a repository and its submodules \"\"\"\n\n        git_config = 'protocol.version=2'\n        git_cmd('-c', git_config, 'fetch', 'origin', ref, '--depth=1')\n        git_cmd('checkout', 'FETCH_HEAD')\n        git_cmd(\n            '-c', git_config, 'submodule', 'update', '--init', '--recursive',\n            '--depth=1',\n        )\n\n    def clone(self, repo: str, ref: str, deps: Sequence[str] = ()) -> str:\n        \"\"\"Clone the given url and checkout the specific ref.\"\"\"\n\n        def clone_strategy(directory: str) -> None:\n            git.init_repo(directory, repo)\n            env = git.no_git_env()\n\n            def _git_cmd(*args: str) -> None:\n                cmd_output_b('git', *args, cwd=directory, env=env)\n\n            try:\n                self._shallow_clone(ref, _git_cmd)\n            except CalledProcessError:\n                self._complete_clone(ref, _git_cmd)\n\n        return self._new_repo(repo, ref, deps, clone_strategy)\n\n    LOCAL_RESOURCES = (\n        'Cargo.toml', 'main.go', 'go.mod', 'main.rs', '.npmignore',\n        'package.json', 'pre_commit_dummy_package.gemspec', 'setup.py',\n        'environment.yml', 'Makefile.PL',\n    )\n\n    def make_local(self, deps: Sequence[str]) -> str:\n        def make_local_strategy(directory: str) -> None:\n            for resource in self.LOCAL_RESOURCES:\n                contents = resource_text(f'empty_template_{resource}')\n                with open(os.path.join(directory, resource), 'w') as f:\n                    f.write(contents)\n\n            env = git.no_git_env()\n\n            # initialize the git repository so it looks more like cloned repos\n            def _git_cmd(*args: str) -> None:\n                cmd_output_b('git', *args, cwd=directory, env=env)\n\n            git.init_repo(directory, '<<unknown>>')\n            _git_cmd('add', '.')\n            git.commit(repo=directory)\n\n        return self._new_repo(\n            'local', C.LOCAL_REPO_VERSION, deps, make_local_strategy,\n        )\n\n    def _create_config_table(self, db: sqlite3.Connection) -> None:\n        db.executescript(\n            'CREATE TABLE IF NOT EXISTS configs ('\n            '   path TEXT NOT NULL,'\n            '   PRIMARY KEY (path)'\n            ');',\n        )\n\n    def mark_config_used(self, path: str) -> None:\n        if self.readonly:  # pragma: win32 no cover\n            return\n        path = os.path.realpath(path)\n        # don't insert config files that do not exist\n        if not os.path.exists(path):\n            return\n        with self.connect() as db:\n            # TODO: eventually remove this and only create in _create\n            self._create_config_table(db)\n            db.execute('INSERT OR IGNORE INTO configs VALUES (?)', (path,))\n\n    def select_all_configs(self) -> List[str]:\n        with self.connect() as db:\n            self._create_config_table(db)\n            rows = db.execute('SELECT path FROM configs').fetchall()\n            return [path for path, in rows]\n\n    def delete_configs(self, configs: List[str]) -> None:\n        with self.connect() as db:\n            rows = [(path,) for path in configs]\n            db.executemany('DELETE FROM configs WHERE path = ?', rows)\n\n    def select_all_repos(self) -> List[Tuple[str, str, str]]:\n        with self.connect() as db:\n            return db.execute('SELECT repo, ref, path from repos').fetchall()\n\n    def delete_repo(self, db_repo_name: str, ref: str, path: str) -> None:\n        with self.connect() as db:\n            db.execute(\n                'DELETE FROM repos WHERE repo = ? and ref = ?',\n                (db_repo_name, ref),\n            )\n        rmtree(path)\n", "path": "pre_commit/store.py"}]}