{{ message }}
Repository navigation
Expand file tree
/
Copy pathregistry.py
More file actions
326 lines (268 loc) · 10.2 KB
/
Copy pathregistry.py
File metadata and controls
326 lines (268 loc) · 10.2 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
"""Multi-repo registry and connection pool.
Manages a registry of multiple repositories at ``~/.code-review-graph/registry.json``
and provides a connection pool for concurrent access to multiple graph databases.
"""
from __future__ import annotations
import json
import logging
import sqlite3
import threading
from collections import OrderedDict
from pathlib import Path
from .constants import crg_home
logger = logging.getLogger(__name__)
def default_registry_path() -> Path:
"""Return the full path to ``registry.json``.
Lives under :func:`~code_review_graph.constants.crg_home`, so ``$CRG_HOME``
redirects it along with the rest of the per-user state.
"""
return crg_home() / "registry.json"
class Registry:
"""Manages a JSON-based registry of code-review-graph repositories.
Each entry stores the repo path and an optional alias.
The registry lives at ``~/.code-review-graph/registry.json``, or under
``$CRG_HOME`` when that is set.
"""
def __init__(self, path: Path | None = None) -> None:
self._path = path or default_registry_path()
self._path.parent.mkdir(parents=True, exist_ok=True)
self._lock = threading.Lock()
self._repos: list[dict[str, str]] = []
self._load()
def _load(self) -> None:
"""Load registry from disk."""
if self._path.exists():
try:
data = json.loads(self._path.read_text(encoding="utf-8", errors="replace"))
self._repos = data.get("repos", [])
except (json.JSONDecodeError, KeyError, TypeError):
logger.warning("Invalid registry file, starting fresh: %s", self._path)
self._repos = []
else:
self._repos = []
def _save(self) -> None:
"""Write registry to disk."""
self._path.parent.mkdir(parents=True, exist_ok=True)
data = {"repos": self._repos}
self._path.write_text(
json.dumps(data, indent=2, ensure_ascii=False) + "\n", encoding="utf-8"
)
def register(
self, path: str, alias: str | None = None, data_dir: str | None = None,
) -> dict[str, str]:
"""Register a repository path.
Validates that the path contains a ``.git`` or ``.code-review-graph``
directory.
Args:
path: Absolute or relative path to the repository root.
alias: Optional short alias for the repository.
data_dir: Optional external directory for graph database.
Returns:
The registered entry dict.
Raises:
ValueError: If the path is not a valid repository.
"""
resolved = Path(path).resolve()
if not resolved.is_dir():
raise ValueError(f"Path is not a directory: {resolved}")
has_repo_marker = (
(resolved / ".git").exists()
or (resolved / ".svn").exists()
or (resolved / ".code-review-graph").exists()
)
if not has_repo_marker:
raise ValueError(
f"Path does not look like a repository "
f"(no .git, .svn, or .code-review-graph): {resolved}"
)
with self._lock:
# Check for duplicate path
str_path = str(resolved)
for entry in self._repos:
if entry["path"] == str_path:
# Update alias and/or data_dir if provided
if alias:
entry["alias"] = alias
if data_dir:
entry["data_dir"] = str(Path(data_dir).resolve())
self._save()
return entry
new_entry: dict[str, str] = {"path": str_path}
if alias:
new_entry["alias"] = alias
if data_dir:
new_entry["data_dir"] = str(Path(data_dir).resolve())
self._repos.append(new_entry)
self._save()
return new_entry
def unregister(self, path_or_alias: str) -> bool:
"""Remove a repository by path or alias.
Args:
path_or_alias: Either the absolute path or the alias.
Returns:
True if an entry was removed, False otherwise.
"""
with self._lock:
resolved = str(Path(path_or_alias).resolve())
original_len = len(self._repos)
self._repos = [
entry for entry in self._repos
if entry["path"] != resolved
and entry.get("alias") != path_or_alias
]
if len(self._repos) < original_len:
self._save()
return True
return False
def list_repos(self) -> list[dict[str, str]]:
"""Return list of all registered repositories.
Returns:
List of dicts with 'path' and optional 'alias' keys.
"""
with self._lock:
return list(self._repos)
def find_by_alias(self, alias: str) -> dict[str, str] | None:
"""Look up a repository by its alias.
Args:
alias: The alias to search for.
Returns:
The matching entry, or None.
"""
with self._lock:
for entry in self._repos:
if entry.get("alias") == alias:
return dict(entry)
return None
def find_by_path(self, path: str) -> dict[str, str] | None:
"""Look up a repository by its path.
Args:
path: The path to search for.
Returns:
The matching entry, or None.
"""
resolved = str(Path(path).resolve())
with self._lock:
for entry in self._repos:
if entry["path"] == resolved:
return dict(entry)
return None
def set_data_dir(self, path: str, data_dir: str) -> dict[str, str]:
"""Set the external data directory for a repository.
Args:
path: Repository path (absolute or relative).
data_dir: External directory path to store graph database.
Returns:
The updated or created registry entry.
"""
resolved = str(Path(path).resolve())
data_resolved = str(Path(data_dir).resolve())
with self._lock:
# Check for existing entry
for entry in self._repos:
if entry["path"] == resolved:
entry["data_dir"] = data_resolved
self._save()
return dict(entry)
# Create new entry if not found
new_entry = {
"path": resolved,
"data_dir": data_resolved
}
self._repos.append(new_entry)
self._save()
return new_entry
def get_data_dir_for_repo(self, path: str) -> str | None:
"""Get the stored data directory for a repository.
Args:
path: Repository path (absolute or relative).
Returns:
The stored data_dir path, or None if not set.
"""
resolved = str(Path(path).resolve())
with self._lock:
for entry in self._repos:
if entry["path"] == resolved:
return entry.get("data_dir")
return None
class ConnectionPool:
"""LRU connection pool for SQLite graph databases.
Caches open connections keyed by database path, evicting the least
recently used connection when the pool is full.
"""
def __init__(self, max_size: int = 10) -> None:
self._max_size = max_size
self._pool: OrderedDict[str, sqlite3.Connection] = OrderedDict()
self._lock = threading.Lock()
def get(self, db_path: str) -> sqlite3.Connection:
"""Get or create a connection for the given database path.
Args:
db_path: Path to the SQLite database file.
Returns:
An open SQLite connection.
"""
key = str(Path(db_path).resolve())
with self._lock:
if key in self._pool:
self._pool.move_to_end(key)
return self._pool[key]
# Evict LRU if full
while len(self._pool) >= self._max_size:
evict_key, evict_conn = self._pool.popitem(last=False)
try:
evict_conn.close()
except sqlite3.Error:
logger.debug("Failed to close evicted connection: %s", evict_key)
logger.debug("Evicted connection: %s", evict_key)
conn = sqlite3.connect(
key, timeout=30, check_same_thread=False,
isolation_level=None,
)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA busy_timeout=5000")
self._pool[key] = conn
return conn
def close_all(self) -> None:
"""Close all connections in the pool."""
with self._lock:
for key, conn in self._pool.items():
try:
conn.close()
except sqlite3.Error:
logger.debug("Failed to close connection: %s", key)
self._pool.clear()
@property
def size(self) -> int:
"""Current number of open connections."""
with self._lock:
return len(self._pool)
def resolve_repo(
registry: Registry,
repo: str | None,
cwd: str | None = None,
) -> str | None:
"""Resolve a repo parameter to an absolute path.
Resolution order:
1. If repo is given, try as alias first.
2. If repo is given and not an alias, try as a direct path.
3. If repo is None, use cwd.
Args:
registry: The Registry instance.
repo: Alias or path string, or None.
cwd: Current working directory fallback.
Returns:
Resolved absolute path string, or None if unresolvable.
"""
if repo:
# Try alias first
entry = registry.find_by_alias(repo)
if entry:
return entry["path"]
# Try as direct path
path = Path(repo).resolve()
if path.is_dir():
return str(path)
# Fall back to CWD
if cwd:
return str(Path(cwd).resolve())
return None
You can’t perform that action at this time.