Filter deployments by any tag

docs/v3/api-ref/rest-api/server/schema.json

@@ -15710,6 +15710,27 @@
                             ]
                         ]
                     },
+                    "any_": {
+                        "anyOf": [
+                            {
+                                "items": {
+                                    "type": "string"
+                                },
+                                "type": "array"
+                            },
+                            {
+                                "type": "null"
+                            }
+                        ],
+                        "title": "Any",
+                        "description": "A list of tags to include",
+                        "examples": [
+                            [
+                                "tag-1",
+                                "tag-2"
+                            ]
+                        ]
+                    },
                     "is_null_": {
                         "anyOf": [
                             {

src/prefect/client/schemas/filters.py

@@ -505,6 +505,11 @@ class DeploymentFilterTags(PrefectBaseModel, OperatorMixin):
             " superset of the list"
         ),
     )
+    any_: Optional[list[str]] = Field(
+        default=None,
+        examples=[["tag-1", "tag-2"]],
+        description="A list of tags to include",
+    )
     is_null_: Optional[bool] = Field(
         default=None, description="If true, only include deployments without tags"
     )

src/prefect/server/schemas/filters.py

@@ -11,7 +11,7 @@
 from pydantic import ConfigDict, Field
 
 import prefect.server.schemas as schemas
-from prefect.server.database import PrefectDBInterface, db_injector
+from prefect.server.database import PrefectDBInterface, db_injector, orm_models
 from prefect.server.utilities.schemas.bases import PrefectBaseModel
 from prefect.types import DateTime
 from prefect.utilities.collections import AutoEnum
@@ -1158,6 +1158,12 @@ class DeploymentFilterTags(PrefectOperatorFilterBaseModel):
             " superset of the list"
         ),
     )
+    any_: Optional[list[str]] = Field(
+        default=None,
+        examples=[["tag-1", "tag-2"]],
+        description="A list of tags to include",
+    )
+
     is_null_: Optional[bool] = Field(
         default=None, description="If true, only include deployments without tags"
     )
@@ -1167,7 +1173,9 @@ def _get_filter_list(
     ) -> Iterable[sa.ColumnExpressionArgument[bool]]:
         filters: list[sa.ColumnElement[bool]] = []
         if self.all_ is not None:
-            filters.append(db.Deployment.tags.has_all(_as_array(self.all_)))
+            filters.append(orm_models.Deployment.tags.has_all(_as_array(self.all_)))
+        if self.any_ is not None:
+            filters.append(orm_models.Deployment.tags.has_any(_as_array(self.any_)))
         if self.is_null_ is not None:
             filters.append(
                 db.Deployment.tags == [] if self.is_null_ else db.Deployment.tags != []

tests/client/test_prefect_client.py

@@ -45,6 +45,8 @@
 from prefect.client.schemas.filters import (
     ArtifactFilter,
     ArtifactFilterKey,
+    DeploymentFilter,
+    DeploymentFilterTags,
     FlowFilter,
     FlowRunFilter,
     FlowRunFilterTags,
@@ -786,6 +788,59 @@ def foo():
     assert lookup.name == "test-deployment"
 
 
+@pytest.mark.parametrize(
+    "deployment_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_deployment_by_any_tag(
+    prefect_client, deployment_tags, filter_tags, expected_match
+):
+    @flow
+    def moo_deng():
+        pass
+
+    flow_id = await prefect_client.create_flow(moo_deng)
+
+    await prefect_client.create_deployment(
+        flow_id=flow_id,
+        name="moisturized-deployment",
+        tags=deployment_tags,
+    )
+    deployment_responses = await prefect_client.read_deployments(
+        deployment_filter=DeploymentFilter(tags=DeploymentFilterTags(any_=filter_tags))
+    )
+    if expected_match:
+        assert len(deployment_responses) == 1
+        assert deployment_responses[0].name == "moisturized-deployment"
+    else:
+        assert len(deployment_responses) == 0
+
+
 async def test_read_deployment_by_name_fails_with_helpful_suggestion(prefect_client):
     """this is a regression test for https://github.com/PrefectHQ/prefect/issues/15571"""
 

ui-v2/src/api/prefect.ts

@@ -5382,6 +5382,11 @@ export interface components {
              * @description A list of tags. Deployments will be returned only if their tags are a superset of the list
              */
             all_?: string[] | null;
+            /**
+             * Any
+             * @description A list of tags to include
+             */
+            any_?: string[] | null;
             /**
              * Is Null
              * @description If true, only include deployments without tags