Search for files by name or content with performance optimizations.
(self, request: FileSearchRequest)
| 594 | return all_files |
| 595 | |
| 596 | async def search(self, request: FileSearchRequest) -> ActionResult: |
| 597 | """Search for files by name or content with performance optimizations.""" |
| 598 | try: |
| 599 | absolute = self._policy.resolve_relative(request.path) |
| 600 | relative = self._policy.to_relative(absolute) |
| 601 | results: list[SearchResult] = [] |
| 602 | |
| 603 | # Performance optimizations |
| 604 | file_types_set = set(request.file_types) if request.file_types else None |
| 605 | search_query = request.query if request.case_sensitive else request.query.lower() |
| 606 | max_matches_per_file = 50 |
| 607 | |
| 608 | def _match_name(name: str) -> bool: |
| 609 | """Check if filename matches query.""" |
| 610 | if not request.case_sensitive: |
| 611 | return search_query in name.lower() |
| 612 | return request.query in name |
| 613 | |
| 614 | async def _search_file(file_path: Path) -> Optional[SearchResult]: |
| 615 | """Search within a single file.""" |
| 616 | if request.by == "name": |
| 617 | if _match_name(file_path.name): |
| 618 | return SearchResult(path=self._policy.to_relative(file_path), matches=[]) |
| 619 | return None |
| 620 | |
| 621 | # Content search with optimizations |
| 622 | try: |
| 623 | data = await self._storage.read_bytes(file_path) |
| 624 | text = data.decode('utf-8', errors='ignore') |
| 625 | lines = text.splitlines() |
| 626 | matches: list[SearchMatch] = [] |
| 627 | |
| 628 | for idx, line in enumerate(lines, 1): |
| 629 | if len(matches) >= max_matches_per_file: |
| 630 | break |
| 631 | |
| 632 | hay = line if request.case_sensitive else line.lower() |
| 633 | if search_query in hay: |
| 634 | matches.append(SearchMatch(line=idx, text=line)) |
| 635 | |
| 636 | if matches: |
| 637 | return SearchResult(path=self._policy.to_relative(file_path), matches=matches) |
| 638 | except Exception: |
| 639 | return None |
| 640 | return None |
| 641 | |
| 642 | async def _walk(dir_path: Path) -> None: |
| 643 | """Walk directory tree and search files.""" |
| 644 | nonlocal results |
| 645 | try: |
| 646 | entries = list(dir_path.iterdir()) |
| 647 | # Process directories first, then files for better performance |
| 648 | dirs = [e for e in entries if e.is_dir()] |
| 649 | files = [e for e in entries if e.is_file()] |
| 650 | |
| 651 | # Process directories concurrently |
| 652 | if dirs: |
| 653 | tasks = [_walk(d) for d in dirs] |
no test coverage detected