Extract shared snapshot loader logic

Co-authored-by: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com>

Author: badMade

src/commands.py

@@ -1,13 +1,13 @@
 from __future__ import annotations
 
-import json
 from collections.abc import Mapping
 from dataclasses import dataclass
 from functools import lru_cache
 from pathlib import Path
 from types import MappingProxyType
 
 from .models import PortingBacklog, PortingModule
+from .snapshot_loader import load_porting_modules
 
 SNAPSHOT_PATH = (
     Path(__file__).resolve().parent / "reference_data" / "commands_snapshot.json"
@@ -23,21 +23,7 @@ class CommandExecution:
     message: str
 
 
-@lru_cache(maxsize=1)
-def load_command_snapshot() -> tuple[PortingModule, ...]:
-    raw_entries = json.loads(SNAPSHOT_PATH.read_text())
-    return tuple(
-        PortingModule(
-            name=entry["name"],
-            responsibility=entry["responsibility"],
-            source_hint=entry["source_hint"],
-            status="mirrored",
-        )
-        for entry in raw_entries
-    )
-
-
-PORTED_COMMANDS = load_command_snapshot()
+PORTED_COMMANDS = load_porting_modules(SNAPSHOT_PATH)
 
 
 @lru_cache(maxsize=1)

src/snapshot_loader.py

@@ -0,0 +1,21 @@
+from __future__ import annotations
+
+import json
+from functools import lru_cache
+from pathlib import Path
+
+from .models import PortingModule
+
+
+@lru_cache(maxsize=None)
+def load_porting_modules(snapshot_path: Path) -> tuple[PortingModule, ...]:
+    raw_entries = json.loads(snapshot_path.read_text())
+    return tuple(
+        PortingModule(
+            name=entry["name"],
+            responsibility=entry["responsibility"],
+            source_hint=entry["source_hint"],
+            status="mirrored",
+        )
+        for entry in raw_entries
+    )

src/tools.py

@@ -1,14 +1,15 @@
 from __future__ import annotations
 
-import json
 from dataclasses import dataclass
-from functools import lru_cache
 from pathlib import Path
 
 from .models import PortingBacklog, PortingModule
 from .permissions import ToolPermissionContext
+from .snapshot_loader import load_porting_modules
 
-SNAPSHOT_PATH = Path(__file__).resolve().parent / 'reference_data' / 'tools_snapshot.json'
+SNAPSHOT_PATH = (
+    Path(__file__).resolve().parent / "reference_data" / "tools_snapshot.json"
+)
 
 
 @dataclass(frozen=True)
@@ -20,25 +21,11 @@ class ToolExecution:
     message: str
 
 
-@lru_cache(maxsize=1)
-def load_tool_snapshot() -> tuple[PortingModule, ...]:
-    raw_entries = json.loads(SNAPSHOT_PATH.read_text())
-    return tuple(
-        PortingModule(
-            name=entry['name'],
-            responsibility=entry['responsibility'],
-            source_hint=entry['source_hint'],
-            status='mirrored',
-        )
-        for entry in raw_entries
-    )
-
-
-PORTED_TOOLS = load_tool_snapshot()
+PORTED_TOOLS = load_porting_modules(SNAPSHOT_PATH)
 
 
 def build_tool_backlog() -> PortingBacklog:
-    return PortingBacklog(title='Tool surface', modules=list(PORTED_TOOLS))
+    return PortingBacklog(title="Tool surface", modules=list(PORTED_TOOLS))
 
 
 def tool_names() -> list[str]:
@@ -53,10 +40,15 @@ def get_tool(name: str) -> PortingModule | None:
     return None
 
 
-def filter_tools_by_permission_context(tools: tuple[PortingModule, ...], permission_context: ToolPermissionContext | None = None) -> tuple[PortingModule, ...]:
+def filter_tools_by_permission_context(
+    tools: tuple[PortingModule, ...],
+    permission_context: ToolPermissionContext | None = None,
+) -> tuple[PortingModule, ...]:
     if permission_context is None:
         return tools
-    return tuple(module for module in tools if not permission_context.blocks(module.name))
+    return tuple(
+        module for module in tools if not permission_context.blocks(module.name)
+    )
 
 
 def get_tools(
@@ -66,31 +58,56 @@ def get_tools(
 ) -> tuple[PortingModule, ...]:
     tools = list(PORTED_TOOLS)
     if simple_mode:
-        tools = [module for module in tools if module.name in {'BashTool', 'FileReadTool', 'FileEditTool'}]
+        tools = [
+            module
+            for module in tools
+            if module.name in {"BashTool", "FileReadTool", "FileEditTool"}
+        ]
     if not include_mcp:
-        tools = [module for module in tools if 'mcp' not in module.name.lower() and 'mcp' not in module.source_hint.lower()]
+        tools = [
+            module
+            for module in tools
+            if "mcp" not in module.name.lower()
+            and "mcp" not in module.source_hint.lower()
+        ]
     return filter_tools_by_permission_context(tuple(tools), permission_context)
 
 
 def find_tools(query: str, limit: int = 20) -> list[PortingModule]:
     needle = query.lower()
-    matches = [module for module in PORTED_TOOLS if needle in module.name.lower() or needle in module.source_hint.lower()]
+    matches = [
+        module
+        for module in PORTED_TOOLS
+        if needle in module.name.lower() or needle in module.source_hint.lower()
+    ]
     return matches[:limit]
 
 
-def execute_tool(name: str, payload: str = '') -> ToolExecution:
+def execute_tool(name: str, payload: str = "") -> ToolExecution:
     module = get_tool(name)
     if module is None:
-        return ToolExecution(name=name, source_hint='', payload=payload, handled=False, message=f'Unknown mirrored tool: {name}')
+        return ToolExecution(
+            name=name,
+            source_hint="",
+            payload=payload,
+            handled=False,
+            message=f"Unknown mirrored tool: {name}",
+        )
     action = f"Mirrored tool '{module.name}' from {module.source_hint} would handle payload {payload!r}."
-    return ToolExecution(name=module.name, source_hint=module.source_hint, payload=payload, handled=True, message=action)
+    return ToolExecution(
+        name=module.name,
+        source_hint=module.source_hint,
+        payload=payload,
+        handled=True,
+        message=action,
+    )
 
 
 def render_tool_index(limit: int = 20, query: str | None = None) -> str:
     modules = find_tools(query, limit) if query else list(PORTED_TOOLS[:limit])
-    lines = [f'Tool entries: {len(PORTED_TOOLS)}', '']
+    lines = [f"Tool entries: {len(PORTED_TOOLS)}", ""]
     if query:
-        lines.append(f'Filtered by: {query}')
-        lines.append('')
-    lines.extend(f'- {module.name} — {module.source_hint}' for module in modules)
-    return '\n'.join(lines)
+        lines.append(f"Filtered by: {query}")
+        lines.append("")
+    lines.extend(f"- {module.name} — {module.source_hint}" for module in modules)
+    return "\n".join(lines)

tests/test_tools.py

@@ -1,8 +1,9 @@
 from __future__ import annotations
 
 import unittest
+from src.snapshot_loader import load_porting_modules
+from src.tools import SNAPSHOT_PATH
 from src.tools import (
-    load_tool_snapshot,
     build_tool_backlog,
     tool_names,
     get_tool,
@@ -18,8 +19,8 @@
 
 
 class TestTools(unittest.TestCase):
-    def test_load_tool_snapshot(self) -> None:
-        tools = load_tool_snapshot()
+    def test_load_porting_modules(self) -> None:
+        tools = load_porting_modules(SNAPSHOT_PATH)
         self.assertIsInstance(tools, tuple)
         self.assertTrue(len(tools) > 0)
         for tool in tools: