Allow deployments to be sorted in the api (#7187)

Co-authored-by: zangell44 <zachary.james.angell@gmail.com>
Co-authored-by: Michael Adkins <michael@prefect.io>

src/prefect/client/orion.py

@@ -1330,6 +1330,7 @@ async def read_deployments(
         task_run_filter: schemas.filters.TaskRunFilter = None,
         deployment_filter: schemas.filters.DeploymentFilter = None,
         limit: int = None,
+        sort: schemas.sorting.DeploymentSort = None,
         offset: int = 0,
     ) -> schemas.core.Deployment:
         """
@@ -1363,6 +1364,7 @@ async def read_deployments(
             ),
             "limit": limit,
             "offset": offset,
+            "sort": sort,
         }
         response = await self._client.post(f"/deployments/filter", json=body)
         return pydantic.parse_obj_as(List[schemas.core.Deployment], response.json())

src/prefect/orion/api/deployments.py

@@ -136,6 +136,9 @@ async def read_deployments(
     flow_runs: schemas.filters.FlowRunFilter = None,
     task_runs: schemas.filters.TaskRunFilter = None,
     deployments: schemas.filters.DeploymentFilter = None,
+    sort: schemas.sorting.DeploymentSort = Body(
+        schemas.sorting.DeploymentSort.NAME_ASC
+    ),
     db: OrionDBInterface = Depends(provide_database_interface),
 ) -> List[schemas.core.Deployment]:
     """
@@ -145,6 +148,7 @@ async def read_deployments(
         return await models.deployments.read_deployments(
             session=session,
             offset=offset,
+            sort=sort,
             limit=limit,
             flow_filter=flows,
             flow_run_filter=flow_runs,

src/prefect/orion/database/migrations/versions/postgresql/2022_10_14_172612_ad4b1b4d1e9d_index_deployment_created.py

@@ -0,0 +1,25 @@
+"""Index deployment.created
+
+Revision ID: ad4b1b4d1e9d
+Revises: 22b7cb02e593
+Create Date: 2022-10-14 17:26:12.326496
+
+"""
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "ad4b1b4d1e9d"
+down_revision = "22b7cb02e593"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    with op.batch_alter_table("deployment", schema=None) as batch_op:
+        batch_op.create_index("ix_deployment__created", ["created"], unique=False)
+
+
+def downgrade():
+
+    with op.batch_alter_table("deployment", schema=None) as batch_op:
+        batch_op.drop_index("ix_deployment__created")

src/prefect/orion/database/migrations/versions/sqlite/2022_10_14_172612_ad4b1b4d1e9d_index_deployment_created.py

@@ -0,0 +1,25 @@
+"""Index deployment.created
+
+Revision ID: ad4b1b4d1e9d
+Revises: 22b7cb02e593
+Create Date: 2022-10-14 17:26:12.326496
+
+"""
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "ad4b1b4d1e9d"
+down_revision = "22b7cb02e593"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    with op.batch_alter_table("deployment", schema=None) as batch_op:
+        batch_op.create_index("ix_deployment__created", ["created"], unique=False)
+
+
+def downgrade():
+
+    with op.batch_alter_table("deployment", schema=None) as batch_op:
+        batch_op.drop_index("ix_deployment__created")

src/prefect/orion/database/orm_models.py

@@ -712,6 +712,10 @@ def __table_args__(cls):
                 "name",
                 unique=True,
             ),
+            sa.Index(
+                "ix_deployment__created",
+                "created",
+            ),
         )
 
 

src/prefect/orion/models/deployments.py

@@ -262,6 +262,7 @@ async def read_deployments(
     flow_run_filter: schemas.filters.FlowRunFilter = None,
     task_run_filter: schemas.filters.TaskRunFilter = None,
     deployment_filter: schemas.filters.DeploymentFilter = None,
+    sort: schemas.sorting.DeploymentSort = schemas.sorting.DeploymentSort.NAME_ASC,
 ):
     """
     Read deployments.
@@ -274,13 +275,13 @@ async def read_deployments(
         flow_run_filter: only select deployments whose flow runs match these criteria
         task_run_filter: only select deployments whose task runs match these criteria
         deployment_filter: only select deployment that match these filters
-
+        sort: the sort criteria for selected deployments. Defaults to `name` ASC.
 
     Returns:
         List[db.Deployment]: deployments
     """
 
-    query = select(db.Deployment).order_by(db.Deployment.name)
+    query = select(db.Deployment).order_by(sort.as_sql_sort(db=db))
 
     query = await _apply_deployment_filters(
         query=query,

src/prefect/orion/schemas/sorting.py

@@ -102,11 +102,30 @@ class FlowSort(AutoEnum):
     NAME_DESC = AutoEnum.auto()
 
     def as_sql_sort(self, db: "OrionDBInterface") -> ColumnElement:
-        """Return an expression used to sort flow runs"""
+        """Return an expression used to sort flows"""
         sort_mapping = {
             "CREATED_DESC": db.Flow.created.desc(),
             "UPDATED_DESC": db.Flow.updated.desc(),
             "NAME_ASC": db.Flow.name.asc(),
             "NAME_DESC": db.Flow.name.desc(),
         }
         return sort_mapping[self.value]
+
+
+class DeploymentSort(AutoEnum):
+    """Defines deployment sorting options."""
+
+    CREATED_DESC = AutoEnum.auto()
+    UPDATED_DESC = AutoEnum.auto()
+    NAME_ASC = AutoEnum.auto()
+    NAME_DESC = AutoEnum.auto()
+
+    def as_sql_sort(self, db: "OrionDBInterface") -> ColumnElement:
+        """Return an expression used to sort deployments"""
+        sort_mapping = {
+            "CREATED_DESC": db.Deployment.created.desc(),
+            "UPDATED_DESC": db.Deployment.updated.desc(),
+            "NAME_ASC": db.Deployment.name.asc(),
+            "NAME_DESC": db.Deployment.name.desc(),
+        }
+        return sort_mapping[self.value]

tests/orion/api/test_deployments.py

@@ -547,6 +547,21 @@ async def test_read_deployments_offset(self, deployments, client, session):
         # sorted by name by default
         assert response.json()[0]["name"] == "My Deployment Y"
 
+    async def test_read_deployments_sort(self, deployments, client):
+        response = await client.post(
+            "/deployments/filter",
+            json=dict(sort=schemas.sorting.DeploymentSort.NAME_ASC),
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert response.json()[0]["name"] == "My Deployment X"
+
+        response_desc = await client.post(
+            "/deployments/filter",
+            json=dict(sort=schemas.sorting.DeploymentSort.NAME_DESC),
+        )
+        assert response_desc.status_code == status.HTTP_200_OK
+        assert response_desc.json()[0]["name"] == "My Deployment Y"
+
     async def test_read_deployments_returns_empty_list(self, client):
         response = await client.post("/deployments/filter")
         assert response.status_code == status.HTTP_200_OK