LS, refactoring: separate out search_files so it can be used outside of the LS

src/multilspy/language_server.py

@@ -26,7 +26,7 @@ from typing import AsyncIterator, Dict, Iterator, List, Optional, Tuple, Union,
 
 import pathspec
 
-from serena.text_utils import LineType, MatchedConsecutiveLines, TextLine, search_text
+from serena.text_utils import LineType, MatchedConsecutiveLines, TextLine, search_files, search_text
 from . import multilspy_types
 from .lsp_protocol_handler import lsp_types as LSPTypes
 from .lsp_protocol_handler.lsp_constants import LSPConstants
@@ -1156,34 +1156,17 @@ class LanguageServer:
         if isinstance(pattern, str):
             pattern = re.compile(pattern)
         
-        matches = []
-        all_files = await self.request_parsed_files()
-        for path in all_files:
-            # Apply glob filters if provided
-            # TODO: fnmatch is not exactly the same as glob
-            if paths_include_glob and not fnmatch(path, paths_include_glob):
-                self.logger.log(f"Skipping {path}: does not match include pattern {paths_include_glob}", logging.DEBUG)
-                continue
-            
-            if paths_exclude_glob and fnmatch(path, paths_exclude_glob):
-                self.logger.log(f"Skipping {path}: matches exclude pattern {paths_exclude_glob}", logging.DEBUG)
-                continue
-                
-            file_content = self.retrieve_full_file_content(path)
-            search_results = search_text(
-                pattern, 
-                file_content, 
-                source_file_path=path, 
-                allow_multiline_match=True, 
-                context_lines_before=context_lines_before, 
-                context_lines_after=context_lines_after
-            )
-            if len(search_results) > 0:
-                self.logger.log(f"Found {len(search_results)} matches in {path}", logging.DEBUG)
-                matches.extend(search_results)
+        relative_file_paths = await self.request_parsed_files()
+        return search_files(
+            relative_file_paths,
+            pattern,
+            content_reader=self.retrieve_full_file_content,
+            context_lines_before=context_lines_before,
+            context_lines_after=context_lines_after,
+            paths_include_glob=paths_include_glob,
+            paths_exclude_glob=paths_exclude_glob
+        )
         
-        return matches
-                
     async def request_referencing_symbols(
         self,
         relative_file_path: str,

src/serena/text_utils.py

@@ -1,6 +1,11 @@
+import logging
 import re
+from collections.abc import Callable
 from dataclasses import dataclass, field
 from enum import StrEnum
+from fnmatch import fnmatch
+
+log = logging.getLogger(__name__)
 
 
 class LineType(StrEnum):
@@ -205,3 +210,44 @@ def search_text(
                 matches.append(MatchedConsecutiveLines(lines=context_lines, source_file_path=source_file_path))
 
     return matches
+
+
+def default_content_reader(file_path: str) -> str:
+    with open(file_path) as f:
+        return f.read()
+
+
+def search_files(
+    file_paths: list[str],
+    pattern: re.Pattern | str,
+    content_reader: Callable[[str], str] = default_content_reader,
+    context_lines_before: int = 0,
+    context_lines_after: int = 0,
+    paths_include_glob: str | None = None,
+    paths_exclude_glob: str | None = None,
+) -> list[MatchedConsecutiveLines]:
+    matches = []
+    for path in file_paths:
+        # TODO: fnmatch is not exactly the same as glob
+        if paths_include_glob and not fnmatch(path, paths_include_glob):
+            log.debug(f"Skipping {path}: does not match include pattern {paths_include_glob}")
+            continue
+
+        if paths_exclude_glob and fnmatch(path, paths_exclude_glob):
+            log.debug(f"Skipping {path}: matches exclude pattern {paths_exclude_glob}")
+            continue
+
+        file_content = content_reader(path)
+        search_results = search_text(
+            pattern,
+            file_content,
+            source_file_path=path,
+            allow_multiline_match=True,
+            context_lines_before=context_lines_before,
+            context_lines_after=context_lines_after,
+        )
+        if len(search_results) > 0:
+            log.debug(f"Found {len(search_results)} matches in {path}")
+            matches.extend(search_results)
+
+    return matches