Add automation name filter, ability to read automation by name (#12850)

Author: serinamarie

src/prefect/client/orchestration.py

@@ -25,6 +25,8 @@
     handle_deprecated_infra_overrides_parameter,
 )
 from prefect._internal.pydantic import HAS_PYDANTIC_V2
+from prefect.client.schemas import sorting
+from prefect.events import filters
 from prefect.settings import (
     PREFECT_API_SERVICES_TRIGGERS_ENABLED,
     PREFECT_EXPERIMENTAL_EVENTS,
@@ -3178,6 +3180,45 @@ async def read_automation(self, automation_id: UUID) -> Optional[Automation]:
         response.raise_for_status()
         return Automation.parse_obj(response.json())
 
+    async def read_automation_by_name(self, name: str) -> Optional[Automation]:
+        """
+        Query the Prefect API for an automation by name. Only automations matching the provided name will be returned.
+
+        If more than one automation matches the name, the most recently updated automation will be returned.
+
+        Args:
+            name: the name of the automation to query
+
+        Returns:
+            an Automation model representation of the automation, or None if not found. If more than one automation
+            matches the name, the most recently updated automation will be returned.
+        """
+        if not self.server_type.supports_automations():
+            self._raise_for_unsupported_automations()
+        automation_filter = filters.AutomationFilter(name=dict(any_=[name]))
+
+        response = await self._client.post(
+            "/automations/filter",
+            json={
+                "limit": 1,
+                "sort": sorting.AutomationSort.UPDATED_DESC,
+                "automations": automation_filter.dict(json_compatible=True)
+                if automation_filter
+                else None,
+            },
+        )
+
+        response.raise_for_status()
+
+        if not response.json():
+            return None
+
+        else:
+            # normally a `/filter` endpoint would return a list of objects, but read_x_by_name
+            # methods return a single object in all other methods in the client, so
+            # we're ensuring parity there
+            return Automation.parse_obj(response.json()[0])
+
     async def pause_automation(self, automation_id: UUID):
         if not self.server_type.supports_automations():
             self._raise_for_unsupported_automations()

src/prefect/client/schemas/sorting.py

@@ -27,6 +27,15 @@ class TaskRunSort(AutoEnum):
     END_TIME_DESC = AutoEnum.auto()
 
 
+class AutomationSort(AutoEnum):
+    """Defines automation sorting options."""
+
+    CREATED_DESC = AutoEnum.auto()
+    UPDATED_DESC = AutoEnum.auto()
+    NAME_ASC = AutoEnum.auto()
+    NAME_DESC = AutoEnum.auto()
+
+
 class LogSort(AutoEnum):
     """Defines log sorting options."""
 

src/prefect/events/filters.py

@@ -16,6 +16,33 @@
     from pydantic import Field, PrivateAttr  # type: ignore
 
 
+class AutomationFilterCreated(PrefectBaseModel):
+    """Filter by `Automation.created`."""
+
+    before_: Optional[DateTimeTZ] = Field(
+        default=None,
+        description="Only include automations created before this datetime",
+    )
+
+
+class AutomationFilterName(PrefectBaseModel):
+    """Filter by `Automation.created`."""
+
+    any_: Optional[List[str]] = Field(
+        default=None,
+        description="Only include automations with names that match any of these strings",
+    )
+
+
+class AutomationFilter(PrefectBaseModel):
+    name: Optional[AutomationFilterName] = Field(
+        default=None, description="Filter criteria for `Automation.name`"
+    )
+    created: Optional[AutomationFilterCreated] = Field(
+        default=None, description="Filter criteria for `Automation.created`"
+    )
+
+
 class EventDataFilter(PrefectBaseModel, extra="forbid"):
     """A base class for filtering event data."""
 

src/prefect/server/api/automations.py

@@ -1,4 +1,4 @@
-from typing import Sequence
+from typing import Optional, Sequence
 from uuid import UUID
 
 import pendulum
@@ -22,7 +22,7 @@
 from prefect.server.database.dependencies import provide_database_interface
 from prefect.server.database.interface import PrefectDBInterface
 from prefect.server.events import actions
-from prefect.server.events.models import automations
+from prefect.server.events.models import automations as automations_models
 from prefect.server.events.schemas.automations import (
     Automation,
     AutomationCreate,
@@ -133,15 +133,15 @@ async def create_automation(
     owner_resource = automation_dict.pop("owner_resource", None)
 
     async with db.session_context(begin_transaction=True) as session:
-        created_automation = await automations.create_automation(
+        created_automation = await automations_models.create_automation(
             session=session,
             automation=Automation(
                 **automation_dict,
             ),
         )
 
         if owner_resource:
-            await automations.relate_automation_to_resource(
+            await automations_models.relate_automation_to_resource(
                 session,
                 automation_id=created_automation.id,
                 resource_id=owner_resource,
@@ -185,7 +185,7 @@ async def update_automation(
             )
 
     async with db.session_context(begin_transaction=True) as session:
-        updated = await automations.update_automation(
+        updated = await automations_models.update_automation(
             session=session,
             automation_update=automation,
             automation_id=automation_id,
@@ -206,7 +206,7 @@ async def patch_automation(
 ):
     try:
         async with db.session_context(begin_transaction=True) as session:
-            updated = await automations.update_automation(
+            updated = await automations_models.update_automation(
                 session=session,
                 automation_update=automation,
                 automation_id=automation_id,
@@ -230,7 +230,7 @@ async def delete_automation(
     db: PrefectDBInterface = Depends(provide_database_interface),
 ):
     async with db.session_context(begin_transaction=True) as session:
-        deleted = await automations.delete_automation(
+        deleted = await automations_models.delete_automation(
             session=session,
             automation_id=automation_id,
         )
@@ -244,14 +244,16 @@ async def read_automations(
     sort: AutomationSort = Body(AutomationSort.NAME_ASC),
     limit: int = LimitBody(),
     offset: int = Body(0, ge=0),
+    automations: Optional[AutomationFilter] = None,
     db: PrefectDBInterface = Depends(provide_database_interface),
 ) -> Sequence[Automation]:
     async with db.session_context() as session:
-        return await automations.read_automations_for_workspace(
+        return await automations_models.read_automations_for_workspace(
             session=session,
             sort=sort,
             limit=limit,
             offset=offset,
+            automation_filter=automations,
         )
 
 
@@ -260,7 +262,7 @@ async def count_automations(
     db: PrefectDBInterface = Depends(provide_database_interface),
 ) -> int:
     async with db.session_context() as session:
-        return await automations.count_automations_for_workspace(session=session)
+        return await automations_models.count_automations_for_workspace(session=session)
 
 
 @router.get("/{id:uuid}")
@@ -269,7 +271,7 @@ async def read_automation(
     db: PrefectDBInterface = Depends(provide_database_interface),
 ) -> Automation:
     async with db.session_context() as session:
-        automation = await automations.read_automation(
+        automation = await automations_models.read_automation(
             session=session,
             automation_id=automation_id,
         )
@@ -285,7 +287,7 @@ async def read_automations_related_to_resource(
     db: PrefectDBInterface = Depends(provide_database_interface),
 ) -> Sequence[Automation]:
     async with db.session_context() as session:
-        return await automations.read_automations_related_to_resource(
+        return await automations_models.read_automations_related_to_resource(
             session=session,
             resource_id=resource_id,
         )
@@ -297,7 +299,7 @@ async def delete_automations_owned_by_resource(
     db: PrefectDBInterface = Depends(provide_database_interface),
 ):
     async with db.session_context(begin_transaction=True) as session:
-        await automations.delete_automations_owned_by_resource(
+        await automations_models.delete_automations_owned_by_resource(
             session,
             resource_id=resource_id,
             automation_filter=AutomationFilter(

src/prefect/server/events/filters.py

@@ -44,14 +44,34 @@ def _get_filter_list(self, db: PrefectDBInterface) -> list:
         return filters
 
 
+class AutomationFilterName(PrefectFilterBaseModel):
+    """Filter by `Automation.created`."""
+
+    any_: Optional[List[str]] = Field(
+        default=None,
+        description="Only include automations with names that match any of these strings",
+    )
+
+    def _get_filter_list(self, db: PrefectDBInterface) -> List:
+        filters = []
+        if self.any_ is not None:
+            filters.append(db.Automation.name.in_(self.any_))
+        return filters
+
+
 class AutomationFilter(PrefectOperatorFilterBaseModel):
+    name: Optional[AutomationFilterName] = Field(
+        default=None, description="Filter criteria for `Automation.name`"
+    )
     created: Optional[AutomationFilterCreated] = Field(
         default=None, description="Filter criteria for `Automation.created`"
     )
 
     def _get_filter_list(self, db: PrefectDBInterface) -> List:
         filters = []
 
+        if self.name is not None:
+            filters.append(self.name.as_sql_filter(db))
         if self.created is not None:
             filters.append(self.created.as_sql_filter(db))
 

src/prefect/server/events/models/automations.py

@@ -35,11 +35,14 @@ async def read_automations_for_workspace(
     sort: AutomationSort = AutomationSort.NAME_ASC,
     limit: Optional[int] = None,
     offset: Optional[int] = None,
+    automation_filter: Optional[filters.AutomationFilter] = None,
 ) -> Sequence[Automation]:
     query = sa.select(db.Automation)
 
     query = query.order_by(db.Automation.sort_expression(sort))
 
+    if automation_filter:
+        query = query.where(automation_filter.as_sql_filter(db))
     if limit is not None:
         query = query.limit(limit)
     if offset is not None:

tests/client/test_prefect_client.py

@@ -2090,6 +2090,77 @@ async def test_create_automation(self, cloud_client, automation: AutomationCore)
             )
             assert automation_id == UUID(created_automation["id"])
 
+    async def test_read_automation(self, cloud_client, automation: AutomationCore):
+        with respx.mock(base_url=PREFECT_CLOUD_API_URL.value()) as router:
+            created_automation = automation.dict(json_compatible=True)
+            created_automation["id"] = str(uuid4())
+
+            created_automation_id = created_automation["id"]
+
+            read_route = router.get(f"/automations/{created_automation_id}").mock(
+                return_value=httpx.Response(200, json=created_automation)
+            )
+
+            read_automation = await cloud_client.read_automation(created_automation_id)
+
+            assert read_route.called
+            assert read_automation.id == UUID(created_automation["id"])
+
+    async def test_read_automation_not_found(
+        self, cloud_client, automation: AutomationCore
+    ):
+        with respx.mock(base_url=PREFECT_CLOUD_API_URL.value()) as router:
+            created_automation = automation.dict(json_compatible=True)
+            created_automation["id"] = str(uuid4())
+
+            created_automation_id = created_automation["id"]
+
+            read_route = router.get(f"/automations/{created_automation_id}").mock(
+                return_value=httpx.Response(404)
+            )
+
+            with pytest.raises(prefect.exceptions.PrefectHTTPStatusError, match="404"):
+                await cloud_client.read_automation(created_automation_id)
+
+            assert read_route.called
+
+    async def test_read_automation_by_name(
+        self, cloud_client, automation: AutomationCore
+    ):
+        with respx.mock(base_url=PREFECT_CLOUD_API_URL.value()) as router:
+            created_automation = automation.dict(json_compatible=True)
+            created_automation["id"] = str(uuid4())
+            read_route = router.post("/automations/filter").mock(
+                return_value=httpx.Response(200, json=[created_automation])
+            )
+            read_automation = await cloud_client.read_automation_by_name(
+                automation.name
+            )
+
+            assert read_route.called
+            assert isinstance(read_automation, AutomationCore)
+            assert read_automation.id == UUID(created_automation["id"])
+            assert read_automation.name == automation.name == created_automation["name"]
+
+    async def test_read_automation_by_name_not_found(
+        self, cloud_client, automation: AutomationCore
+    ):
+        with respx.mock(base_url=PREFECT_CLOUD_API_URL.value()) as router:
+            created_automation = automation.dict(json_compatible=True)
+            created_automation["id"] = str(uuid4())
+            created_automation["name"] = "nonexistent"
+            read_route = router.post("/automations/filter").mock(
+                return_value=httpx.Response(200, json=[])
+            )
+
+            nonexistent_automation = await cloud_client.read_automation_by_name(
+                name="nonexistent"
+            )
+
+            assert read_route.called
+
+            assert nonexistent_automation is None
+
     async def test_delete_owned_automations_not_cloud_runtime_error(
         self, prefect_client
     ):

tests/events/server/models/test_automations.py

@@ -59,6 +59,34 @@ async def test_reading_automations_by_workspace_by_name(
     ]
 
 
+async def test_reading_automation_by_workspace_by_name_filtered_match(
+    automations_session: AsyncSession,
+    some_workspace_automations: Sequence[Automation],
+):
+    existing = await automations.read_automations_for_workspace(
+        session=automations_session,
+        sort=AutomationSort.NAME_ASC,
+        automation_filter=filters.AutomationFilter(name={"any_": ["automation 2"]}),
+    )
+
+    assert len(existing) == 1
+
+    assert existing[0].name == "automation 2"
+
+
+async def test_reading_automation_by_workspace_by_name_filtered_mismatch(
+    automations_session: AsyncSession,
+    some_workspace_automations: Sequence[Automation],
+):
+    existing = await automations.read_automations_for_workspace(
+        session=automations_session,
+        sort=AutomationSort.NAME_ASC,
+        automation_filter=filters.AutomationFilter(name={"any_": ["automation 5"]}),
+    )
+
+    assert len(existing) == 0
+
+
 async def test_reading_automations_by_workspace_paging(
     automations_session: AsyncSession,
     some_workspace_automations: Sequence[Automation],