✨ Add metrics aggregation for numeric types

Author: simonwoerpel

openaleph_search/parse/parser.py

@@ -198,6 +198,9 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
         # expand query with name synonyms (name_symbols and name_keys)
         self.synonyms = self.getbool("synonyms", False)
 
+        # metric aggregations (sum, avg, min, max) on numeric fields
+        self.metrics = self.prefixed_items("metric:")
+
     @cached_property
     def collection_ids(self) -> set[str]:
         collections = self.filters.get("collection_id", set())
@@ -334,4 +337,5 @@ def to_dict(self) -> dict[str, Any]:
         parser["synonyms"] = self.synonyms
         parser["include_fields"] = list(self.include_fields)
         parser["dehydrate"] = self.dehydrate
+        parser["metrics"] = {key: list(val) for key, val in self.metrics.items()}
         return parser

openaleph_search/query/base.py

@@ -40,6 +40,7 @@ class Query:
         "score": "_score",
     }
     SORT_DEFAULT: ClassVar[list[str | dict[str, Any]]] = ["_score"]
+    METRIC_TYPES: ClassVar[tuple[str, ...]] = ("sum", "avg", "min", "max")
     SOURCE: ClassVar[dict[str, Any]] = {}
 
     def __init__(self, parser: SearchQueryParser) -> None:
@@ -279,6 +280,15 @@ def get_aggregations(self) -> dict[str, Any]:
                 },
             }
 
+        # Metric aggregations (sum, avg, min, max) on numeric fields
+        for metric_type, fields in self.parser.metrics.items():
+            if metric_type not in self.METRIC_TYPES:
+                continue
+            for field in fields:
+                es_field = f"{Field.NUMERIC}.{field}"
+                agg_name = f"{field}.{metric_type}"
+                aggregations[agg_name] = {metric_type: {"field": es_field}}
+
         return aggregations
 
     def get_significant_background(self) -> BoolQuery | None:

tests/test_search.py

@@ -617,6 +617,64 @@ def test_search_synonyms_name_keys(cleanup_after):
     assert result["hits"]["hits"][0]["_id"] == "darc-limited-company"
 
 
+def test_search_metric_aggregations(cleanup_after):
+    """Test metric aggregations (sum, avg, min, max) on numeric fields"""
+    entities = [
+        make_entity(
+            {
+                "id": "payment1",
+                "schema": "Payment",
+                "properties": {"amount": ["100"], "date": ["2024-01-01"]},
+            }
+        ),
+        make_entity(
+            {
+                "id": "payment2",
+                "schema": "Payment",
+                "properties": {"amount": ["250"], "date": ["2024-02-01"]},
+            }
+        ),
+        make_entity(
+            {
+                "id": "payment3",
+                "schema": "Payment",
+                "properties": {"amount": ["150"], "date": ["2024-03-01"]},
+            }
+        ),
+    ]
+    index_bulk("test_metrics", entities, sync=True)
+
+    # Test sum (filter:schemata=Interval to include the intervals index)
+    query = _create_query(
+        "/search?filter:dataset=test_metrics&filter:schemata=Interval"
+        "&metric:sum=amount"
+    )
+    result = query.search()
+    assert result["hits"]["total"]["value"] == 3
+    assert result["aggregations"]["amount.sum"]["value"] == 500.0
+
+    # Test multiple metrics at once
+    query = _create_query(
+        "/search?filter:dataset=test_metrics&filter:schemata=Interval"
+        "&metric:sum=amount&metric:avg=amount&metric:min=amount&metric:max=amount"
+    )
+    result = query.search()
+    aggs = result["aggregations"]
+    assert aggs["amount.sum"]["value"] == 500.0
+    assert aggs["amount.avg"]["value"] == pytest.approx(500.0 / 3)
+    assert aggs["amount.min"]["value"] == 100.0
+    assert aggs["amount.max"]["value"] == 250.0
+
+    # Test with a filter narrowing results
+    query = _create_query(
+        "/search?filter:dataset=test_metrics&filter:schemata=Interval"
+        "&filter:gte:properties.date=2024-02-01&metric:sum=amount"
+    )
+    result = query.search()
+    assert result["hits"]["total"]["value"] == 2
+    assert result["aggregations"]["amount.sum"]["value"] == 400.0
+
+
 def test_search_translation_plaintext(cleanup_after):
     """Test that PlainText translatedText is searchable via ES copy_to into the
     translation field."""

tests/test_search_parser.py

@@ -1,6 +1,6 @@
 from unittest import TestCase
 
-from openaleph_search.parse.parser import QueryParser
+from openaleph_search.parse.parser import QueryParser, SearchQueryParser
 
 args = QueryParser(
     [
@@ -75,6 +75,15 @@ def test_to_dict(self):
         self.assertEqual(set(parser_dict["filters"]["key2"]), set(["foo3", "foo5"]))
         self.assertEqual(set(parser_dict["filters"]["key3"]), set(["foo4"]))
 
+    def test_metric_parsing(self):
+        from werkzeug.datastructures import OrderedMultiDict
+
+        parser = SearchQueryParser(
+            OrderedMultiDict([("metric:sum", "amount"), ("metric:sum", "salary")]),
+            None,
+        )
+        assert parser.metrics == {"sum": {"amount", "salary"}}
+
     def test_limit_zero(self):
         """Test that limit=0 is preserved and not converted to default."""
         # Test with limit=0 in query args

tests/test_search_query.py

@@ -176,6 +176,27 @@ def test_highlight_text(self):
             "bar",
         )
 
+    def test_metric_aggregations(self):
+        q = query(
+            [
+                ("metric:sum", "amount"),
+                ("metric:avg", "amount"),
+                ("metric:min", "registrationArea"),
+            ]
+        )
+        aggs = q.get_aggregations()
+        self.assertEqual(aggs["amount.sum"], {"sum": {"field": "numeric.amount"}})
+        self.assertEqual(aggs["amount.avg"], {"avg": {"field": "numeric.amount"}})
+        self.assertEqual(
+            aggs["registrationArea.min"],
+            {"min": {"field": "numeric.registrationArea"}},
+        )
+
+    def test_metric_invalid_type(self):
+        q = query([("metric:percentile", "amount")])
+        aggs = q.get_aggregations()
+        self.assertNotIn("amount.percentile", aggs)
+
     def test_schema_filter(self):
         q = query([("filter:schema", "Person")])
         assert q.get_filters() == [{"term": {"schema": "Person"}}]