feat: add S3 search_files method

Author: giancarloromeo

services/storage/src/simcore_service_storage/simcore_s3_dsm.py

@@ -1,9 +1,10 @@
 import contextlib
 import datetime
+import fnmatch
 import logging
 import tempfile
 import urllib.parse
-from collections.abc import Coroutine
+from collections.abc import AsyncGenerator, Coroutine
 from contextlib import suppress
 from dataclasses import dataclass
 from pathlib import Path
@@ -959,6 +960,136 @@ async def search_owned_files(
                 resolved_fmds.append(convert_db_to_model(updated_fmd))
         return resolved_fmds
 
+    def _create_file_metadata_from_s3_object(
+        self, s3_obj: S3MetaData, user_id: UserID
+    ) -> FileMetaData | None:
+        """Create FileMetaData from S3 object, return None if invalid."""
+        try:
+            return FileMetaData.from_simcore_node(
+                user_id=user_id,
+                file_id=TypeAdapter(SimcoreS3FileID).validate_python(s3_obj.object_key),
+                bucket=self.simcore_bucket_name,
+                location_id=self.get_location_id(),
+                location_name=self.get_location_name(),
+                sha256_checksum=None,
+                file_size=s3_obj.size,
+                last_modified=s3_obj.last_modified,
+                entity_tag=s3_obj.e_tag,
+            )
+        except (ValidationError, ValueError):
+            return None
+
+    async def _process_s3_page_results(
+        self,
+        current_page_results: list[FileMetaData],
+    ) -> list[FileMetaData]:
+        current_page_results.sort(
+            key=lambda x: x.last_modified
+            or datetime.datetime.min.replace(tzinfo=datetime.UTC),
+            reverse=True,
+        )
+
+        result_project_ids = list(
+            {
+                result.project_id
+                for result in current_page_results
+                if result.project_id is not None
+            }
+        )
+
+        if result_project_ids:
+            current_page_results = await _add_frontend_needed_data(
+                get_db_engine(self.app),
+                project_ids=result_project_ids,
+                data=current_page_results,
+            )
+
+        return current_page_results
+
+    async def _search_project_s3_files(
+        self,
+        proj_id: ProjectID,
+        filename_pattern: str,
+        user_id: UserID,
+        items_per_page: NonNegativeInt,
+    ) -> AsyncGenerator[list[FileMetaData], None]:
+        """Search S3 files in a specific project and yield results page by page."""
+        s3_client = get_s3_client(self.app)
+        min_parts_for_valid_s3_object = 2
+        current_page_results: list[FileMetaData] = []
+
+        try:
+            async for s3_objects in s3_client.list_objects_paginated(
+                bucket=self.simcore_bucket_name,
+                prefix=f"{proj_id}/",
+                items_per_page=items_per_page * 2,
+            ):
+                for s3_obj in s3_objects:
+                    filename = Path(s3_obj.object_key).name
+
+                    if (
+                        fnmatch.fnmatch(filename, filename_pattern)
+                        and len(s3_obj.object_key.split("/"))
+                        >= min_parts_for_valid_s3_object
+                    ):
+                        file_meta = self._create_file_metadata_from_s3_object(
+                            s3_obj, user_id
+                        )
+                        if file_meta:
+                            current_page_results.append(file_meta)
+
+                            if len(current_page_results) >= items_per_page:
+                                processed_results = await self._process_s3_page_results(
+                                    current_page_results
+                                )
+                                yield processed_results
+                                current_page_results = []
+
+            if current_page_results:
+                processed_results = await self._process_s3_page_results(
+                    current_page_results
+                )
+                yield processed_results
+
+        except S3KeyNotFoundError:
+            with log_context(
+                _logger, logging.DEBUG, f"Failed to search S3 for project {proj_id}"
+            ):
+                return
+
+    async def search_files(
+        self,
+        user_id: UserID,
+        *,
+        filename_pattern: str,
+        project_id: ProjectID | None = None,
+        items_per_page: NonNegativeInt = 100,
+    ) -> AsyncGenerator[list[FileMetaData], None]:
+        """
+        Search for files in S3 using a wildcard pattern for filenames.
+        Returns results as an async generator that yields pages of results.
+
+        Args:
+            user_id: The user requesting the search
+            filename_pattern: Wildcard pattern for filename matching (e.g., "*.txt", "test_*.json")
+            project_id: Optional project ID to limit search to specific project
+            items_per_page: Number of items to return per page
+
+        Yields:
+            List of FileMetaData objects for each page
+        """
+        # Validate access rights
+        accessible_projects_ids = await get_accessible_project_ids(
+            get_db_engine(self.app), user_id=user_id, project_id=project_id
+        )
+
+        # Search each accessible project
+        for proj_id in accessible_projects_ids:
+            async for page_results in self._search_project_s3_files(
+                proj_id, filename_pattern, user_id, items_per_page
+            ):
+                yield page_results
+
     async def create_soft_link(
         self, user_id: int, target_file_id: StorageFileID, link_file_id: StorageFileID
     ) -> FileMetaData:

services/storage/tests/unit/test_simcore_s3_dsm.py

@@ -168,6 +168,131 @@ async def test_upload_and_search(
         assert file.file_name in {"file1", "file2"}
 
 
+@pytest.mark.parametrize(
+    "location_id",
+    [SimcoreS3DataManager.get_location_id()],
+    ids=[SimcoreS3DataManager.get_location_name()],
+    indirect=True,
+)
+async def test_search_files(
+    simcore_s3_dsm: SimcoreS3DataManager,
+    upload_file: Callable[..., Awaitable[tuple[Path, SimcoreS3FileID]]],
+    file_size: ByteSize,
+    user_id: UserID,
+    project_id: ProjectID,
+    faker: Faker,
+):
+    # Upload files with different patterns
+    test_files = [
+        ("test_file1.txt", "*.txt"),
+        ("test_file2.txt", "*.txt"),
+        ("document.pdf", "*.pdf"),
+        ("data_file.csv", "data_*.csv"),
+        ("backup_file.bak", "backup_*"),
+        ("config.json", "*.json"),
+        ("temp_data.tmp", "temp_*"),
+    ]
+
+    uploaded_files = []
+    for file_name, _ in test_files:
+        checksum: SHA256Str = TypeAdapter(SHA256Str).validate_python(faker.sha256())
+        _, file_id = await upload_file(file_size, file_name, sha256_checksum=checksum)
+        uploaded_files.append((file_name, file_id, checksum))
+
+    # Test 1: Search for all .txt files
+    txt_results = []
+    async for page in simcore_s3_dsm.search_files(
+        user_id=user_id,
+        filename_pattern="*.txt",
+        project_id=project_id,
+        items_per_page=10,
+    ):
+        txt_results.extend(page)
+
+    # Should find 2 txt files
+    assert len(txt_results) == 2
+    txt_names = {file.file_name for file in txt_results}
+    assert txt_names == {"test_file1.txt", "test_file2.txt"}
+
+    # Test 2: Search with specific prefix pattern
+    data_results = []
+    async for page in simcore_s3_dsm.search_files(
+        user_id=user_id,
+        filename_pattern="data_*",
+        project_id=project_id,
+        items_per_page=10,
+    ):
+        data_results.extend(page)
+
+    # Should find 1 data file
+    assert len(data_results) == 1
+    assert data_results[0].file_name == "data_file.csv"
+
+    # Test 3: Search with pattern that matches multiple extensions
+    temp_results = []
+    async for page in simcore_s3_dsm.search_files(
+        user_id=user_id,
+        filename_pattern="temp_*",
+        project_id=project_id,
+        items_per_page=10,
+    ):
+        temp_results.extend(page)
+
+    # Should find 1 temp file
+    assert len(temp_results) == 1
+    assert temp_results[0].file_name == "temp_data.tmp"
+
+    # Test 4: Search with pattern that doesn't match anything
+    no_match_results = []
+    async for page in simcore_s3_dsm.search_files(
+        user_id=user_id,
+        filename_pattern="nonexistent_*",
+        project_id=project_id,
+        items_per_page=10,
+    ):
+        no_match_results.extend(page)
+
+    assert len(no_match_results) == 0
+
+    # Test 5: Search without project_id restriction (all accessible projects)
+    all_results = []
+    async for page in simcore_s3_dsm.search_files(
+        user_id=user_id,
+        filename_pattern="*",
+        items_per_page=10,
+    ):
+        all_results.extend(page)
+
+    # Should find at least our uploaded files
+    assert len(all_results) >= len(test_files)
+
+    # Verify that each result has expected FileMetaData structure
+    for file_meta in all_results:
+        assert isinstance(file_meta, FileMetaData)
+        assert file_meta.file_name is not None
+        assert file_meta.file_id is not None
+        assert file_meta.user_id == user_id
+        assert file_meta.project_id is not None
+
+    # Test 6: Test pagination with small page size
+    paginated_results = []
+    page_count = 0
+    async for page in simcore_s3_dsm.search_files(
+        user_id=user_id,
+        filename_pattern="*",
+        project_id=project_id,
+        items_per_page=2,  # Small page size to test pagination
+    ):
+        paginated_results.extend(page)
+        page_count += 1
+        # Each page should have at most 2 items
+        assert len(page) <= 2
+
+    # Should have multiple pages and all our files
+    assert page_count >= 4  # At least 7 files / 2 per page = 4 pages
+    assert len(paginated_results) == len(test_files)
+
+
 @pytest.fixture
 async def paths_for_export(
     random_project_with_files: Callable[