Add search endpoints to the API. Closes #7579.

components/api_server/.vulture_ignore_list.py

@@ -1,8 +1,10 @@
-text  # unused variable (src/model/issue_tracker.py:26)
-api  # unused variable (src/routes/plugins/auth_plugin.py:26)
-_.apply  # unused method (src/routes/plugins/auth_plugin.py:31)
+text  # unused variable (src/model/issue_tracker.py:28)
+_.content_type  # unused attribute (src/routes/pdf.py:22)
+api  # unused variable (src/routes/plugins/auth_plugin.py:25)
+_.apply  # unused method (src/routes/plugins/auth_plugin.py:30)
 api  # unused variable (src/routes/plugins/injection_plugin.py:19)
 _.setup  # unused method (src/routes/plugins/injection_plugin.py:26)
 _.apply  # unused method (src/routes/plugins/injection_plugin.py:33)
-_.content_type  # unused attribute (src/routes/report.py:122)
-_.content_type  # unused attribute (src/routes/report.py:140)
+search_query  # unused variable (src/routes/search.py:35)
+ok  # unused variable (src/routes/search.py:42)
+ok  # unused variable (src/routes/search.py:51)

components/api_server/src/routes/__init__.py

@@ -41,6 +41,7 @@
     post_report_new,
 )
 from .reports_overview import export_reports_overview_as_pdf, get_reports_overview, post_reports_overview_attribute
+from .search import search
 from .server import get_server, QUALITY_TIME_VERSION
 from .settings import get_settings, update_settings
 from .source import (

components/api_server/src/routes/search.py

@@ -0,0 +1,101 @@
+"""Search endpoint."""
+
+import logging
+from collections.abc import Sequence
+from typing import Literal, TypedDict, TYPE_CHECKING
+
+import bottle
+from pymongo.database import Database
+
+from shared.utils.type import ItemId
+
+from database.reports import latest_reports
+from model.iterators import metrics, sources, subjects
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+    from model.report import Report
+
+
+DomainObjectType = Literal["metric", "report", "source", "subject"]
+SearchQuery = dict[str, str]
+
+# Base types for the search endpoint responses:
+
+
+class BaseSearchResponse(TypedDict):
+    """Base class for search responses."""
+
+    domain_object_type: DomainObjectType
+
+
+class BaseQueryParsedSearchResponse(BaseSearchResponse):
+    """Base class for responses after successfully parsing the search query."""
+
+    search_query: SearchQuery
+
+
+class BaseFailedSearchResponse(BaseSearchResponse):
+    """Base class for responses after failed searches."""
+
+    error: str
+    ok: Literal[False]  # Would be nice to set False as default value as well, but that's not possible with TypedDict.
+
+
+# Concrete types for the search endpoint responses:
+
+
+class SearchResults(BaseQueryParsedSearchResponse):
+    """Response for successful searches."""
+
+    ok: Literal[True]  # Would be nice to set True as default value as well, but that's not possible with TypedDict.
+    uuids: Sequence[ItemId]
+
+
+class ParseError(BaseFailedSearchResponse):
+    """Response for failed search query parsing."""
+
+
+class SearchError(BaseQueryParsedSearchResponse, BaseFailedSearchResponse):  # type: ignore[misc]
+    """Response for searches that failed after successfully parsing the search query."""
+
+    # Ignore the false positive mypy error: Overwriting TypedDict field "domain_object_type" while merging.
+    # See https://github.com/python/mypy/issues/8714.
+
+
+def match_attribute(
+    domain_object_type: DomainObjectType, domain_object, attribute_name: str, attribute_value: str
+) -> bool:
+    """Return whether the domain object has an attribute with the specified name and value."""
+    # Note that we don't use domain_object.get(attribute_name) because that would make it impossible to search for None.
+    if domain_object_type == "source":
+        parameters = domain_object["parameters"]
+        if attribute_name in parameters and parameters[attribute_name] == attribute_value:
+            return True
+    return attribute_name in domain_object and domain_object[attribute_name] == attribute_value
+
+
+@bottle.post("/api/v3/<domain_object_type>/search", authentication_required=False)
+def search(domain_object_type: DomainObjectType, database: Database) -> SearchResults | ParseError | SearchError:
+    """Search for domain objects of the specified type by attribute value."""
+    try:
+        query = dict(bottle.request.json)
+        attribute_name, attribute_value = set(query.items()).pop()
+    except Exception as reason:  # noqa: BLE001
+        logging.warning("Parsing search query for %s failed: %s", domain_object_type, reason)
+        return ParseError(domain_object_type=domain_object_type, error=str(reason), ok=False)
+    try:
+        reports = latest_reports(database)
+        if domain_object_type == "report":
+            domain_objects: list[Report] | Iterator[Report] = reports
+        else:
+            domain_objects = {"metric": metrics, "source": sources, "subject": subjects}[domain_object_type](*reports)
+        uuids = [
+            domain_object.uuid
+            for domain_object in domain_objects
+            if match_attribute(domain_object_type, domain_object, attribute_name, attribute_value)
+        ]
+    except Exception as reason:  # pragma: no feature-test-cover # noqa: BLE001
+        logging.warning("Searching for %s failed: %s", domain_object_type, reason)
+        return SearchError(domain_object_type=domain_object_type, error=str(reason), ok=False, search_query=query)
+    return SearchResults(domain_object_type=domain_object_type, ok=True, search_query=query, uuids=uuids)

components/api_server/tests/routes/test_search.py

@@ -0,0 +1,161 @@
+"""Unit tests for the search endpoint."""
+
+from unittest.mock import Mock, patch
+
+from routes import search
+
+from tests.base import DatabaseTestCase, disable_logging
+from tests.fixtures import METRIC_ID, METRIC_ID2, REPORT_ID, REPORT_ID2, SOURCE_ID, SOURCE_ID2, SUBJECT_ID, SUBJECT_ID2
+
+
+class SearchTest(DatabaseTestCase):
+    """Unit tests for searching domain objects."""
+
+    @patch("bottle.request", Mock(json={"attribute_name": "attribute_value"}))
+    def test_no_reports(self):
+        """Test that no domain objects can be found if there are no reports."""
+        self.database.reports.find.return_value = []
+        for domain_object_type in ("metric", "report", "source", "subject"):
+            expected_results = {
+                "domain_object_type": domain_object_type,
+                "ok": True,
+                "search_query": {"attribute_name": "attribute_value"},
+                "uuids": [],
+            }
+            self.assertEqual(expected_results, search(domain_object_type, self.database))
+
+    @patch("bottle.request", Mock(json={"title": "Report 1"}))
+    def test_search_report(self):
+        """Test that a report can be found."""
+        self.database.reports.find.return_value = [
+            {"title": "Report 1", "report_uuid": REPORT_ID},
+            {"title": "Report 2", "report_uuid": REPORT_ID2},
+        ]
+        expected_results = {
+            "domain_object_type": "report",
+            "ok": True,
+            "search_query": {"title": "Report 1"},
+            "uuids": [REPORT_ID],
+        }
+        self.assertEqual(expected_results, search("report", self.database))
+
+    @patch("bottle.request", Mock(json={"name": "Subject 1"}))
+    def test_search_subject(self):
+        """Test that a subject can be found."""
+        self.database.reports.find.return_value = [
+            {"subjects": {SUBJECT_ID: {"name": "Subject 1"}}},
+            {"subjects": {SUBJECT_ID2: {"name": "Subject 2"}}},
+        ]
+        expected_results = {
+            "domain_object_type": "subject",
+            "ok": True,
+            "search_query": {"name": "Subject 1"},
+            "uuids": [SUBJECT_ID],
+        }
+        self.assertEqual(expected_results, search("subject", self.database))
+
+    @patch("bottle.request", Mock(json={"name": "Metric 1"}))
+    def test_search_metric(self):
+        """Test that a metric can be found."""
+        self.database.reports.find.return_value = [
+            {"subjects": {SUBJECT_ID: {"metrics": {METRIC_ID: {"name": "Metric 1"}}}}},
+            {"subjects": {SUBJECT_ID2: {"metrics": {METRIC_ID2: {"name": "Metric 2"}}}}},
+        ]
+        expected_results = {
+            "domain_object_type": "metric",
+            "ok": True,
+            "search_query": {"name": "Metric 1"},
+            "uuids": [METRIC_ID],
+        }
+        self.assertEqual(expected_results, search("metric", self.database))
+
+    @patch("bottle.request", Mock(json={"name": "Metric"}))
+    def test_search_metric_multiple_matches(self):
+        """Test that multiple metrics can be found."""
+        self.database.reports.find.return_value = [
+            {"subjects": {SUBJECT_ID: {"metrics": {METRIC_ID: {"name": "Metric"}}}}},
+            {"subjects": {SUBJECT_ID2: {"metrics": {METRIC_ID2: {"name": "Metric"}}}}},
+        ]
+        expected_results = {
+            "domain_object_type": "metric",
+            "ok": True,
+            "search_query": {"name": "Metric"},
+            "uuids": [METRIC_ID, METRIC_ID2],
+        }
+        self.assertEqual(expected_results, search("metric", self.database))
+
+    @patch("bottle.request", Mock(json={"name": "Source 1"}))
+    def test_search_source(self):
+        """Test that a source can be found."""
+        self.database.reports.find.return_value = [
+            {
+                "subjects": {
+                    SUBJECT_ID: {
+                        "metrics": {METRIC_ID: {"sources": {SOURCE_ID: {"name": "Source 1", "parameters": {}}}}}
+                    }
+                }
+            },
+            {
+                "subjects": {
+                    SUBJECT_ID2: {
+                        "metrics": {METRIC_ID2: {"sources": {SOURCE_ID2: {"name": "Source 2", "parameters": {}}}}}
+                    }
+                }
+            },
+        ]
+        expected_results = {
+            "domain_object_type": "source",
+            "ok": True,
+            "search_query": {"name": "Source 1"},
+            "uuids": [SOURCE_ID],
+        }
+        self.assertEqual(expected_results, search("source", self.database))
+
+    @patch("bottle.request", Mock(json={"url": "https://example.org"}))
+    def test_search_source_by_parameter(self):
+        """Test that a source can be found by parameter."""
+        self.database.reports.find.return_value = [
+            {
+                "subjects": {
+                    SUBJECT_ID: {
+                        "metrics": {METRIC_ID: {"sources": {SOURCE_ID: {"parameters": {"url": "https://example.org"}}}}}
+                    }
+                }
+            },
+            {
+                "subjects": {
+                    SUBJECT_ID2: {
+                        "metrics": {
+                            METRIC_ID2: {"sources": {SOURCE_ID2: {"parameters": {"url": "https://example2.org"}}}}
+                        }
+                    }
+                }
+            },
+        ]
+        expected_results = {
+            "domain_object_type": "source",
+            "ok": True,
+            "search_query": {"url": "https://example.org"},
+            "uuids": [SOURCE_ID],
+        }
+        self.assertEqual(expected_results, search("source", self.database))
+
+    @disable_logging
+    @patch("bottle.request", Mock(json={"name": "Source"}))
+    def test_failed_search(self):
+        """Test that an error response is returned when an exception occurs during the search."""
+        self.database.reports.find.side_effect = [RuntimeError("error message")]
+        expected_results = {
+            "domain_object_type": "source",
+            "error": "error message",
+            "ok": False,
+            "search_query": {"name": "Source"},
+        }
+        self.assertEqual(expected_results, search("source", self.database))
+
+    @disable_logging
+    @patch("bottle.request", Mock(json={}))
+    def test_failed_parsing_of_search_query(self):
+        """Test that an error response is returned when the search query could not be parsed."""
+        expected_results = {"domain_object_type": "source", "error": "'pop from an empty set'", "ok": False}
+        self.assertEqual(expected_results, search("source", self.database))