Range min/max query on list

Author: hoijnet

terminusdb_client/tests/integration_tests/test_woql_range_min_max.py

@@ -0,0 +1,113 @@
+"""
+Integration tests for WOQL RangeMin and RangeMax predicates.
+
+Tests finding minimum and maximum values in lists:
+- Integer lists
+- Date lists
+- Single element
+- Empty list
+- Equal elements
+"""
+
+import pytest
+
+from terminusdb_client import Client
+from terminusdb_client.woqlquery.woql_query import WOQLQuery
+
+test_user_agent = "terminusdb-client-python-tests"
+
+
+def intv(v):
+    """Build an xsd:integer typed literal."""
+    return {"@type": "xsd:integer", "@value": v}
+
+
+def datv(v):
+    """Build an xsd:date typed literal."""
+    return {"@type": "xsd:date", "@value": v}
+
+
+class TestRangeMin:
+    """Integration tests for RangeMin."""
+
+    @pytest.fixture(autouse=True)
+    def setup_teardown(self, docker_url):
+        self.client = Client(docker_url, user_agent=test_user_agent)
+        self.client.connect()
+        self.db_name = "test_range_min"
+        if self.db_name in self.client.list_databases():
+            self.client.delete_database(self.db_name)
+        self.client.create_database(self.db_name)
+        yield
+        self.client.delete_database(self.db_name)
+
+    def test_min_integers(self):
+        """Minimum of [7, 2, 9, 1, 5] is 1."""
+        query = WOQLQuery().range_min(
+            [intv(7), intv(2), intv(9), intv(1), intv(5)], "v:m"
+        )
+        result = self.client.query(query)
+        assert len(result["bindings"]) == 1
+        assert result["bindings"][0]["m"]["@value"] == 1
+
+    def test_min_single_element(self):
+        """Single element list returns that element."""
+        query = WOQLQuery().range_min([intv(42)], "v:m")
+        result = self.client.query(query)
+        assert len(result["bindings"]) == 1
+        assert result["bindings"][0]["m"]["@value"] == 42
+
+    def test_min_empty_list(self):
+        """Empty list yields no bindings."""
+        query = WOQLQuery().range_min([], "v:m")
+        result = self.client.query(query)
+        assert len(result["bindings"]) == 0
+
+    def test_min_dates(self):
+        """Minimum of dates."""
+        query = WOQLQuery().range_min(
+            [datv("2024-06-15"), datv("2024-01-01"), datv("2024-03-01")], "v:m"
+        )
+        result = self.client.query(query)
+        assert len(result["bindings"]) == 1
+        assert result["bindings"][0]["m"]["@value"] == "2024-01-01"
+
+    def test_min_equal_elements(self):
+        """All equal elements returns that element."""
+        query = WOQLQuery().range_min([intv(3), intv(3), intv(3)], "v:m")
+        result = self.client.query(query)
+        assert len(result["bindings"]) == 1
+        assert result["bindings"][0]["m"]["@value"] == 3
+
+
+class TestRangeMax:
+    """Integration tests for RangeMax."""
+
+    @pytest.fixture(autouse=True)
+    def setup_teardown(self, docker_url):
+        self.client = Client(docker_url, user_agent=test_user_agent)
+        self.client.connect()
+        self.db_name = "test_range_max"
+        if self.db_name in self.client.list_databases():
+            self.client.delete_database(self.db_name)
+        self.client.create_database(self.db_name)
+        yield
+        self.client.delete_database(self.db_name)
+
+    def test_max_integers(self):
+        """Maximum of [7, 2, 9, 1, 5] is 9."""
+        query = WOQLQuery().range_max(
+            [intv(7), intv(2), intv(9), intv(1), intv(5)], "v:m"
+        )
+        result = self.client.query(query)
+        assert len(result["bindings"]) == 1
+        assert result["bindings"][0]["m"]["@value"] == 9
+
+    def test_max_dates(self):
+        """Maximum of dates."""
+        query = WOQLQuery().range_max(
+            [datv("2024-06-15"), datv("2024-01-01"), datv("2024-03-01")], "v:m"
+        )
+        result = self.client.query(query)
+        assert len(result["bindings"]) == 1
+        assert result["bindings"][0]["m"]["@value"] == "2024-06-15"

terminusdb_client/tests/test_woql_range_min_max.py

@@ -0,0 +1,53 @@
+"""Unit tests for WOQLQuery.range_min and range_max JSON serialization."""
+
+import pytest
+
+from terminusdb_client.woqlquery.woql_query import WOQLQuery
+
+
+class TestRangeMinSerialization:
+    """Tests that range_min produces the correct WOQL JSON AST."""
+
+    def test_variable_result(self):
+        """List and variable result."""
+        query = WOQLQuery().range_min("v:list", "v:m")
+        expected = {
+            "@type": "RangeMin",
+            "list": {"@type": "Value", "variable": "list"},
+            "result": {"@type": "Value", "variable": "m"},
+        }
+        assert query.to_dict() == expected
+
+    def test_raises_on_none_list(self):
+        """Raises ValueError when list is None."""
+        with pytest.raises(ValueError, match="RangeMin takes two parameters"):
+            WOQLQuery().range_min(None, "v:m")
+
+    def test_raises_on_none_result(self):
+        """Raises ValueError when result is None."""
+        with pytest.raises(ValueError, match="RangeMin takes two parameters"):
+            WOQLQuery().range_min("v:list", None)
+
+
+class TestRangeMaxSerialization:
+    """Tests that range_max produces the correct WOQL JSON AST."""
+
+    def test_variable_result(self):
+        """List and variable result."""
+        query = WOQLQuery().range_max("v:list", "v:m")
+        expected = {
+            "@type": "RangeMax",
+            "list": {"@type": "Value", "variable": "list"},
+            "result": {"@type": "Value", "variable": "m"},
+        }
+        assert query.to_dict() == expected
+
+    def test_raises_on_none_list(self):
+        """Raises ValueError when list is None."""
+        with pytest.raises(ValueError, match="RangeMax takes two parameters"):
+            WOQLQuery().range_max(None, "v:m")
+
+    def test_raises_on_none_result(self):
+        """Raises ValueError when result is None."""
+        with pytest.raises(ValueError, match="RangeMax takes two parameters"):
+            WOQLQuery().range_max("v:list", None)

terminusdb_client/woqlquery/woql_query.py

@@ -2470,6 +2470,60 @@ def interval_relation_typed(self, relation, x, y):
         self._cursor["y"] = self._clean_object(y)
         return self
 
+    def range_min(self, input_list, result):
+        """Find the minimum value in a list using the standard ordering.
+
+        Works with any comparable types: numbers, dates, strings.
+        Empty list produces no bindings.
+
+        Parameters
+        ----------
+        input_list : list or str or dict
+            the list of values to search
+        result : str or dict
+            variable or value for the minimum
+
+        Returns
+        -------
+        WOQLQuery object
+            query object that can be chained and/or execute
+        """
+        if input_list is None or result is None:
+            raise ValueError("RangeMin takes two parameters")
+        if self._cursor.get("@type"):
+            self._wrap_cursor_with_and()
+        self._cursor["@type"] = "RangeMin"
+        self._cursor["list"] = self._clean_object(input_list)
+        self._cursor["result"] = self._clean_object(result)
+        return self
+
+    def range_max(self, input_list, result):
+        """Find the maximum value in a list using the standard ordering.
+
+        Works with any comparable types: numbers, dates, strings.
+        Empty list produces no bindings.
+
+        Parameters
+        ----------
+        input_list : list or str or dict
+            the list of values to search
+        result : str or dict
+            variable or value for the maximum
+
+        Returns
+        -------
+        WOQLQuery object
+            query object that can be chained and/or execute
+        """
+        if input_list is None or result is None:
+            raise ValueError("RangeMax takes two parameters")
+        if self._cursor.get("@type"):
+            self._wrap_cursor_with_and()
+        self._cursor["@type"] = "RangeMax"
+        self._cursor["list"] = self._clean_object(input_list)
+        self._cursor["result"] = self._clean_object(result)
+        return self
+
     def interval(self, start, end, interval_val):
         """Constructs or deconstructs a half-open xdd:dateTimeInterval [start, end).