use correct util

src/prefect/server/schemas/filters.py

@@ -266,13 +266,16 @@ class FlowRunFilterTags(PrefectOperatorFilterBaseModel):
     )
 
     def _get_filter_list(self) -> List:
-        from prefect.server.utilities.database import json_contains, json_has_all_keys
+        from prefect.server.utilities.database import (
+            json_has_all_keys,
+            json_has_any_key,
+        )
 
         filters = []
         if self.all_ is not None:
             filters.append(json_has_all_keys(orm_models.FlowRun.tags, self.all_))
         if self.any_ is not None:
-            filters.append(json_contains(orm_models.FlowRun.tags, self.any_))
+            filters.append(json_has_any_key(orm_models.FlowRun.tags, self.any_))
         if self.is_null_ is not None:
             filters.append(
                 orm_models.FlowRun.tags == []

tests/client/test_prefect_client.py

@@ -52,6 +52,7 @@
     LogFilter,
     LogFilterFlowRunId,
     TaskRunFilter,
+    TaskRunFilterFlowRunId,
 )
 from prefect.client.schemas.objects import (
     Flow,
@@ -983,24 +984,53 @@ def bar():
     assert {flow_run.id for flow_run in flow_runs} == {fr_id_4, fr_id_5}
 
 
-async def test_read_flow_runs_with_tags(prefect_client):
+@pytest.mark.parametrize(
+    "run_tags,filter_tags,expected_match",
+    [
+        # Basic single tag matching
+        (["tag-1"], ["tag-1"], True),
+        (["tag-2"], ["tag-1"], False),
+        # Any matching - should match if ANY tag in filter matches
+        (["tag-1", "tag-2"], ["tag-1", "tag-3"], True),
+        (["tag-1"], ["tag-1", "tag-2"], True),
+        (["tag-2"], ["tag-1", "tag-2"], True),
+        # No matches
+        (["tag-1"], ["tag-2", "tag-3"], False),
+        (["tag-1"], ["get-real"], False),
+        # Empty cases
+        ([], ["tag-1"], False),
+        (["tag-1"], [], False),
+    ],
+    ids=[
+        "single_tag_match",
+        "single_tag_no_match",
+        "multiple_tags_partial_match",
+        "subset_match_1",
+        "subset_match_2",
+        "no_matching_tags",
+        "nonexistent_tag",
+        "empty_run_tags",
+        "empty_filter_tags",
+    ],
+)
+async def test_read_flow_runs_with_tags(
+    prefect_client, run_tags, filter_tags, expected_match
+):
     @flow
     def foo():
         pass
 
-    fr_id = (await prefect_client.create_flow_run(foo, tags=["tag-1"])).id
-    (await prefect_client.create_flow_run(foo, tags=["tag-2"])).id
+    flow_run = await prefect_client.create_flow_run(foo, tags=run_tags)
 
     flow_runs = await prefect_client.read_flow_runs(
-        flow_run_filter=FlowRunFilter(tags=FlowRunFilterTags(any_=["tag-1"]))
+        flow_run_filter=FlowRunFilter(tags=FlowRunFilterTags(any_=filter_tags))
     )
-    assert len(flow_runs) == 1
-    assert {flow_run.id for flow_run in flow_runs} == {fr_id}
 
-    no_flow_runs = await prefect_client.read_flow_runs(
-        flow_run_filter=FlowRunFilter(tags=FlowRunFilterTags(any_=["get-real"]))
-    )
-    assert len(no_flow_runs) == 0
+    if expected_match:
+        assert len(flow_runs) == 1
+        assert flow_runs[0].id == flow_run.id
+    else:
+        assert len(flow_runs) == 0
 
 
 async def test_read_flows_without_filter(prefect_client):
@@ -1272,7 +1302,7 @@ def foo():
     )
 
     autonotask_runs = await prefect_client.read_task_runs(
-        task_run_filter=TaskRunFilter(flow_run_id=dict(is_null_=True))
+        task_run_filter=TaskRunFilter(flow_run_id=TaskRunFilterFlowRunId(is_null_=True))
     )
 
     assert len(autonotask_runs) == 2