♻️(core) simplify data flow

I remove DB access

Signed-off-by: charles <[email protected]>

Author: mascarpon3

src/backend/core/api/viewsets.py

@@ -1204,41 +1204,30 @@ def _search_simple(self, request, text):
             },
         )
 
-    def _search_with_indexer(self, indexer, request, params):
+    @staticmethod
+    def _search_with_indexer(indexer, request, params):
         """
-        Returns a list of documents matching the query according to the configured indexer.
+        Returns a list of documents matching the query (q) according to the configured indexer.
         """
-        text = params.validated_data["q"]
-        path = (
-            params.validated_data["path"] if "path" in params.validated_data else None
-        )
         queryset = models.Document.objects.all()
 
-        # Retrieve the documents ids according to indexer.
         results = indexer.search(
-            text=text,
+            q=params.validated_data["q"],
             token=request.session.get("oidc_access_token"),
-            path=path,
+            path=(
+                params.validated_data["path"]
+                if "path" in params.validated_data
+                else None
+            ),
             visited=get_visited_document_ids_of(queryset, request.user),
         )
 
-        docs_by_uuid = {str(d.pk): d for d in queryset.filter(pk__in=results)}
-        ordered_docs = [docs_by_uuid[id] for id in results if id in docs_by_uuid]
-
-        serializer = self.get_serializer(
-            ordered_docs,
-            many=True,
-            context={
-                "request": request,
-            },
-        )
-
         return drf_response.Response(
             {
-                "count": len(serializer.data),
+                "count": len(results),
                 "next": None,
                 "previous": None,
-                "results": serializer.data,
+                "results": results,
             }
         )
 

src/backend/core/services/search_indexers.py

@@ -185,15 +185,15 @@ def push(self, data):
         """
 
     # pylint: disable-next=too-many-arguments,too-many-positional-arguments
-    def search(self, text, token, visited=(), nb_results=None, path=None):
+    def search(self, q, token, visited=(), nb_results=None, path=None):
         """
         Search for documents in Find app.
         Ensure the same default ordering as "Docs" list : -updated_at
 
         Returns ids of the documents
 
         Args:
-            text (str): Text search content.
+            q (str): Text search content.
             token (str): OIDC Authentication token.
             visited (list, optional):
                 List of ids of active public documents with LinkTrace
@@ -205,9 +205,9 @@ def search(self, text, token, visited=(), nb_results=None, path=None):
                 The path to filter documents.
         """
         nb_results = nb_results or self.search_limit
-        response = self.search_query(
+        results = self.search_query(
             data={
-                "q": text,
+                "q": q,
                 "visited": visited,
                 "services": ["docs"],
                 "nb_results": nb_results,
@@ -218,7 +218,7 @@ def search(self, text, token, visited=(), nb_results=None, path=None):
             token=token,
         )
 
-        return [d["_id"] for d in response]
+        return results
 
     @abstractmethod
     def search_query(self, data, token) -> dict:
@@ -234,6 +234,51 @@ class SearchIndexer(BaseDocumentIndexer):
     Document indexer that pushes documents to La Suite Find app.
     """
 
+    def search(self, q, token, visited=(), nb_results=None, path=None):
+        """format Find search results"""
+        search_results = super().search(q, token, visited, nb_results, path)
+        return [
+            {
+                "id": hit["_id"],
+                "title": self.get_title(hit["_source"]),
+                **hit["_source"],
+            }
+            for hit in search_results
+        ]
+
+    @staticmethod
+    def get_title(source):
+        """
+        Extract the title from a search result source dictionary.
+
+        Handles multiple cases:
+        - Localized title fields like "title.<some_extension>"
+        - Fallback to plain "title" field if localized version not found
+        - Returns empty string if no title field exists
+
+        Args:
+            source (dict): The _source dictionary from a search hit
+
+        Returns:
+            str: The extracted title or empty string if not found
+
+        Example:
+            >>> get_title({"title.fr": "Bonjour", "id": 1})
+            "Bonjour"
+            >>> get_title({"title": "Hello", "id": 1})
+            "Hello"
+            >>> get_title({"id": 1})
+            ""
+        """
+        titles = utils.get_value_by_pattern(source, r"^title\.")
+        if titles:
+            title = titles[0]
+        elif "title" in source:
+            title = source["title"]
+        else:
+            title = ""
+        return title
+
     def serialize_document(self, document, accesses):
         """
         Convert a Document to the JSON format expected by La Suite Find.

src/backend/core/tests/test_services_search_indexers.py

@@ -633,3 +633,51 @@ def test_services_search_indexers_search_nb_results(mock_post, indexer_settings)
 
     assert args[0] == indexer_settings.SEARCH_INDEXER_QUERY_URL
     assert kwargs.get("json")["nb_results"] == 109
+
+
+def test_search_indexer_get_title_with_localized_field():
+    """Test extracting title from localized title field."""
+    source = {"title.extension": "Bonjour", "id": 1, "content": "test"}
+    result = SearchIndexer.get_title(source)
+
+    assert result == "Bonjour"
+
+
+def test_search_indexer_get_title_with_multiple_localized_fields():
+    """Test that first matching localized title is returned."""
+    source = {"title.extension": "Bonjour", "title.en": "Hello", "id": 1}
+    result = SearchIndexer.get_title(source)
+
+    assert result in ["Bonjour", "Hello"]
+
+
+def test_search_indexer_get_title_fallback_to_plain_title():
+    """Test fallback to plain 'title' field when no localized field exists."""
+    source = {"title": "Hello World", "id": 1}
+    result = SearchIndexer.get_title(source)
+
+    assert result == "Hello World"
+
+
+def test_search_indexer_get_title_no_title_field():
+    """Test that empty string is returned when no title field exists."""
+    source = {"id": 1, "content": "test"}
+    result = SearchIndexer.get_title(source)
+
+    assert result == ""
+
+
+def test_search_indexer_get_title_with_empty_localized_title():
+    """Test that fallback works when localized title is empty."""
+    source = {"title.extension": "", "title": "Fallback Title", "id": 1}
+    result = SearchIndexer.get_title(source)
+
+    assert result == "Fallback Title"
+
+
+def test_search_indexer_get_title_with_multiple_extension():
+    """Test extracting title from title field with multiple extensions."""
+    source = {"title.extension_1.extension_2": "Bonjour", "id": 1, "content": "test"}
+    result = SearchIndexer.get_title(source)
+
+    assert result == "Bonjour"

src/backend/core/tests/test_utils.py

@@ -100,3 +100,27 @@ def test_utils_get_ancestor_to_descendants_map_multiple_paths():
         "000100020005": {"000100020005"},
         "00010003": {"00010003"},
     }
+
+
+def test_utils_get_value_by_pattern_matching_key():
+    """Test extracting value from a dictionary with a matching key pattern."""
+    data = {"title.extension": "Bonjour", "id": 1, "content": "test"}
+    result = utils.get_value_by_pattern(data, r"^title\.")
+
+    assert result == ["Bonjour"]
+
+
+def test_utils_get_value_by_pattern_multiple_matches():
+    """Test that all matching keys are returned."""
+    data = {"title.extension_1": "Bonjour", "title.extension_2": "Hello", "id": 1}
+    result = utils.get_value_by_pattern(data, r"^title\.")
+
+    assert set(result) == {"Bonjour", "Hello"}
+
+
+def test_utils_get_value_by_pattern_no_match():
+    """Test that empty list is returned when no key matches the pattern."""
+    data = {"name": "Test", "id": 1}
+    result = utils.get_value_by_pattern(data, r"^title\.")
+
+    assert result == []

src/backend/core/utils.py

@@ -10,6 +10,26 @@
 from core import enums
 
 
+def get_value_by_pattern(data, pattern):
+    """
+    Get all values from keys matching a regex pattern in a dictionary.
+
+    Args:
+        data (dict): Source dictionary to search
+        pattern (str): Regex pattern to match against keys
+
+    Returns:
+        list: List of values for all matching keys, empty list if no matches
+
+    Example:
+        >>> get_value_by_pattern({"title.fr": "Bonjour", "id": 1}, r"^title\.")
+        ["Bonjour"]
+        >>> get_value_by_pattern({"title.fr": "Bonjour", "title.en": "Hello"}, r"^title\.")
+        ["Bonjour", "Hello"]
+    """
+    return [value for key, value in data.items() if re.compile(pattern).match(key)]
+
+
 def get_ancestor_to_descendants_map(paths, steplen):
     """
     Given a list of document paths, return a mapping of ancestor_path -> set of descendant_paths.