refactor: Convert legacy filter to be filter_query

cloud_pipelines_backend/filter_query_sql.py

@@ -164,11 +164,12 @@ def build_list_filters(
         page_token.filter_query if page_token_value else filter_query_value
     )
 
-    where_clauses, resolved_filter = _build_filter_where_clauses(
-        filter_value=filter_value,
-        current_user=current_user,
-    )
+    if filter_value:
+        filter_query_value = _convert_legacy_filter_to_filter_query(
+            filter_value=filter_value,
+        )
 
+    where_clauses: list[sql.ColumnElement] = []
     if filter_query_value:
         parsed = filter_query_models.FilterQuery.model_validate_json(filter_query_value)
         where_clauses.append(
@@ -180,7 +181,7 @@ def build_list_filters(
 
     next_page_token = PageToken(
         offset=offset + page_size,
-        filter=resolved_filter,
+        filter=None,
         filter_query=filter_query_value,
     )
 
@@ -219,39 +220,31 @@ def _parse_filter(filter: str) -> dict[str, str]:
     return parsed_filter
 
 
-def _build_filter_where_clauses(
+def _convert_legacy_filter_to_filter_query(
     *,
-    filter_value: str | None,
-    current_user: str | None,
-) -> tuple[list[sql.ColumnElement], str | None]:
-    """Parse a filter string into SQLAlchemy WHERE clauses.
+    filter_value: str,
+) -> str:
+    """Convert a legacy ``filter`` string to an equivalent ``filter_query`` JSON string.
 
-    Returns (where_clauses, next_page_filter_value). The second value is the
-    filter string with shorthand values resolved (e.g. "created_by:me" becomes
-    "created_by:[email protected]") so it can be embedded in the next page token.
+    Only ``created_by`` is supported. ``"me"`` is NOT resolved here — the
+    downstream ``_maybe_resolve_system_values`` handles that.
     """
-    where_clauses: list[sql.ColumnElement] = []
-    parsed_filter = _parse_filter(filter_value) if filter_value else {}
-    for key, value in parsed_filter.items():
+    parsed = _parse_filter(filter_value)
+    predicates: list[dict] = []
+    for key, value in parsed.items():
         if key == "_text":
             raise NotImplementedError("Text search is not implemented yet.")
         elif key == "created_by":
-            if value == "me":
-                if current_user is None:
-                    current_user = ""
-                value = current_user
-                # TODO: Maybe make this a bit more robust.
-                # We need to change the filter since it goes into the next_page_token.
-                filter_value = filter_value.replace(
-                    "created_by:me", f"created_by:{current_user}"
+            if not value:
+                raise errors.ApiValidationError(
+                    "Legacy filter 'created_by' requires a non-empty value."
                 )
-            if value:
-                where_clauses.append(bts.PipelineRun.created_by == value)
-            else:
-                where_clauses.append(bts.PipelineRun.created_by == None)
+            predicates.append(
+                {"value_equals": {"key": SystemKey.CREATED_BY, "value": value}}
+            )
         else:
             raise NotImplementedError(f"Unsupported filter {filter_value}.")
-    return where_clauses, filter_value
+    return json.dumps({"and": predicates})
 
 
 # ---------------------------------------------------------------------------

tests/test_api_server_sql.py

@@ -159,27 +159,13 @@ def test_list_filter_created_by(self, session_factory, service):
         assert len(result.pipeline_runs) == 1
         assert result.pipeline_runs[0].created_by == "user1"
 
-    def test_list_filter_created_by_empty(self, session_factory, service):
-        _create_run(
-            session_factory,
-            service,
-            root_task=_make_task_spec(),
-            created_by=None,
-        )
-        _create_run(
-            session_factory,
-            service,
-            root_task=_make_task_spec(),
-            created_by="user1",
-        )
-
+    def test_list_filter_created_by_empty_raises(self, session_factory, service):
         with session_factory() as session:
-            result = service.list(
-                session=session,
-                filter="created_by:",
-            )
-        assert len(result.pipeline_runs) == 1
-        assert result.pipeline_runs[0].created_by is None
+            with pytest.raises(errors.ApiValidationError, match="non-empty value"):
+                service.list(
+                    session=session,
+                    filter="created_by:",
+                )
 
     def test_list_pagination(self, session_factory, service):
         for i in range(12):

tests/test_filter_query_sql.py

@@ -1,3 +1,5 @@
+import json
+
 import pytest
 import sqlalchemy as sql
 from sqlalchemy.dialects import sqlite as sqlite_dialect
@@ -161,59 +163,46 @@ def test_decode_empty_string(self):
         assert token.filter_query is None
 
 
-class TestBuildFilterWhereClauses:
-    def test_no_filter(self):
-        clauses, next_filter = filter_query_sql._build_filter_where_clauses(
-            filter_value=None,
-            current_user=None,
-        )
-        assert clauses == []
-        assert next_filter is None
-
+class TestConvertLegacyFilterToFilterQuery:
     def test_created_by_literal(self):
-        clauses, next_filter = filter_query_sql._build_filter_where_clauses(
+        result = filter_query_sql._convert_legacy_filter_to_filter_query(
             filter_value="created_by:alice",
-            current_user=None,
         )
-        assert len(clauses) == 1
-        assert next_filter == "created_by:alice"
-
-    def test_created_by_me_resolves(self):
-        clauses, next_filter = filter_query_sql._build_filter_where_clauses(
+        parsed = json.loads(result)
+        assert parsed == {
+            "and": [
+                {
+                    "value_equals": {
+                        "key": "system/pipeline_run.created_by",
+                        "value": "alice",
+                    }
+                }
+            ]
+        }
+
+    def test_created_by_me_not_resolved(self):
+        result = filter_query_sql._convert_legacy_filter_to_filter_query(
             filter_value="created_by:me",
-            current_user="[email protected]",
-        )
-        assert len(clauses) == 1
-        assert next_filter == "created_by:[email protected]"
-
-    def test_created_by_me_no_current_user(self):
-        clauses, next_filter = filter_query_sql._build_filter_where_clauses(
-            filter_value="created_by:me",
-            current_user=None,
         )
-        assert len(clauses) == 1
-        assert next_filter == "created_by:"
+        parsed = json.loads(result)
+        assert parsed["and"][0]["value_equals"]["value"] == "me"
 
-    def test_created_by_empty_value(self):
-        clauses, next_filter = filter_query_sql._build_filter_where_clauses(
-            filter_value="created_by:",
-            current_user=None,
-        )
-        assert len(clauses) == 1
-        assert next_filter == "created_by:"
+    def test_created_by_empty_raises(self):
+        with pytest.raises(errors.ApiValidationError, match="non-empty value"):
+            filter_query_sql._convert_legacy_filter_to_filter_query(
+                filter_value="created_by:",
+            )
 
     def test_unsupported_key_raises(self):
         with pytest.raises(NotImplementedError, match="Unsupported filter"):
-            filter_query_sql._build_filter_where_clauses(
+            filter_query_sql._convert_legacy_filter_to_filter_query(
                 filter_value="unknown_key:value",
-                current_user=None,
             )
 
     def test_text_search_raises(self):
         with pytest.raises(NotImplementedError, match="Text search"):
-            filter_query_sql._build_filter_where_clauses(
+            filter_query_sql._convert_legacy_filter_to_filter_query(
                 filter_value="some_text_without_colon",
-                current_user=None,
             )
 
 
@@ -244,7 +233,7 @@ def test_mutual_exclusivity_raises(self):
                 page_size=10,
             )
 
-    def test_legacy_filter_produces_clauses(self):
+    def test_legacy_filter_produces_annotation_clause(self):
         clauses, offset, next_token = filter_query_sql.build_list_filters(
             filter_value="created_by:alice",
             filter_query_value=None,
@@ -253,8 +242,12 @@ def test_legacy_filter_produces_clauses(self):
             page_size=10,
         )
         assert len(clauses) == 1
+        compiled = _compile(clauses[0])
+        assert "EXISTS" in compiled.upper()
+        assert "pipeline_run_annotation" in compiled
         assert offset == 0
-        assert next_token.filter == "created_by:alice"
+        assert next_token.filter is None
+        assert next_token.filter_query is not None
 
     def test_filter_query_produces_clauses(self):
         fq = '{"and": [{"key_exists": {"key": "team"}}]}'
@@ -270,7 +263,7 @@ def test_filter_query_produces_clauses(self):
         assert "EXISTS" in compiled.upper()
         assert next_token.filter_query == fq
 
-    def test_page_token_restores_offset_and_filters(self):
+    def test_page_token_with_legacy_filter_converts(self):
         token = filter_query_sql.PageToken(
             offset=20,
             filter="created_by:alice",
@@ -284,8 +277,11 @@ def test_page_token_restores_offset_and_filters(self):
         )
         assert offset == 20
         assert len(clauses) == 1
+        compiled = _compile(clauses[0])
+        assert "EXISTS" in compiled.upper()
         assert next_token.offset == 30
-        assert next_token.filter == "created_by:alice"
+        assert next_token.filter is None
+        assert next_token.filter_query is not None
 
     def test_page_token_restores_filter_query(self):
         fq = '{"and": [{"key_exists": {"key": "env"}}]}'
@@ -321,7 +317,10 @@ def test_created_by_me_resolved_in_next_token(self):
             page_size=10,
         )
         assert len(clauses) == 1
-        assert next_token.filter == "created_by:[email protected]"
+        assert next_token.filter is None
+        assert next_token.filter_query is not None
+        parsed_fq = json.loads(next_token.filter_query)
+        assert parsed_fq["and"][0]["value_equals"]["value"] == "me"
 
 
 class TestSystemKeyValidation: