Add deployment pagination

coriolis/api/v1/deployments.py

@@ -4,6 +4,7 @@
 from oslo_log import log as logging
 from webob import exc
 
+from coriolis.api import common
 from coriolis.api.v1 import utils as api_utils
 from coriolis.api.v1.views import deployment_view
 from coriolis.api import wsgi as api_wsgi
@@ -43,11 +44,17 @@ class DeploymentsController(api_wsgi.Controller):
         context.can(deployment_policies.get_deployments_policy_label("list"))
         include_task_info = api_utils.get_bool_url_arg(
             req, "include_task_info", default=False)
+
+        marker, limit = common.get_paging_params(req)
+        sort_keys, sort_dirs = common.get_sort_params(req)
+
         return deployment_view.collection(
             self._deployment_api.get_deployments(
                 context,
                 include_tasks=include_task_info,
-                include_task_info=include_task_info
+                include_task_info=include_task_info,
+                marker=marker, limit=limit,
+                sort_keys=sort_keys, sort_dirs=sort_dirs,
             ))
 
     def index(self, req):

coriolis/conductor/rpc/client.py

@@ -226,10 +226,17 @@ class ConductorClient(rpc.BaseRPCClient):
             ctxt, 'delete_transfer_disks', transfer_id=transfer_id)
 
     def get_deployments(self, ctxt, include_tasks=False,
-                        include_task_info=False):
+                        include_task_info=False,
+                        marker=None, limit=None,
+                        sort_keys=None, sort_dirs=None):
         return self._call(
             ctxt, 'get_deployments', include_tasks=include_tasks,
-            include_task_info=include_task_info)
+            include_task_info=include_task_info,
+            marker=marker,
+            limit=limit,
+            sort_keys=sort_keys,
+            sort_dirs=sort_dirs,
+        )
 
     def get_deployment(self, ctxt, deployment_id, include_task_info=False):
         return self._call(

coriolis/conductor/rpc/server.py

@@ -1374,10 +1374,16 @@ class ConductorServerEndpoint(object):
         return transfer
 
     @staticmethod
-    def get_deployments(ctxt, include_tasks, include_task_info=False):
+    def get_deployments(ctxt, include_tasks, include_task_info=False,
+                        marker=None, limit=None,
+                        sort_keys=None, sort_dirs=None):
         return db_api.get_deployments(
             ctxt, include_tasks,
             include_task_info=include_task_info,
+            marker=marker,
+            limit=limit,
+            sort_keys=sort_keys,
+            sort_dirs=sort_dirs,
             to_dict=True)
 
     @deployment_synchronized

coriolis/db/api.py

@@ -563,6 +563,10 @@ def get_transfer_deployments(context, transfer_id):
 def get_deployments(context,
                     include_tasks=False,
                     include_task_info=False,
+                    marker=None,
+                    limit=None,
+                    sort_keys: list[str] | None = None,
+                    sort_dirs: list[str] | None = None,
                     to_dict=False):
     q = _soft_delete_aware_query(context, models.Deployment)
     if include_tasks:
@@ -572,10 +576,26 @@ def get_deployments(context,
     if include_task_info:
         q = q.options(orm.undefer('info'))
 
-    args = {}
     if is_user_context(context):
-        args["project_id"] = context.project_id
-    result = q.filter_by(**args).all()
+        q = q.filter_by(project_id=context.project_id)
+
+    sort_keys, sort_dirs = process_sort_params(
+        sort_keys,
+        sort_dirs,
+    )
+    if marker:
+        try:
+            marker = get_deployment(context, marker)
+        except exception.NotFound:
+            raise exception.MarkerNotFound(marker=marker)
+    q = sqlalchemy_utils.paginate_query(
+        q, models.Deployment, limit,
+        sort_keys=sort_keys,
+        sort_dirs=sort_dirs,
+        marker=marker,
+    )
+
+    result = q.all()
     if to_dict:
         return [i.to_dict(
             include_task_info=include_task_info,

coriolis/deployments/api.py

@@ -26,9 +26,13 @@ class API(object):
         self._rpc_client.cancel_deployment(ctxt, deployment_id, force)
 
     def get_deployments(self, ctxt, include_tasks=False,
-                        include_task_info=False):
+                        include_task_info=False,
+                        marker=None, limit=None,
+                        sort_keys=None, sort_dirs=None):
         return self._rpc_client.get_deployments(
-            ctxt, include_tasks, include_task_info=include_task_info)
+            ctxt, include_tasks, include_task_info=include_task_info,
+            marker=marker, limit=limit,
+            sort_keys=sort_keys, sort_dirs=sort_dirs)
 
     def get_deployment(self, ctxt, deployment_id, include_task_info=False):
         return self._rpc_client.get_deployment(

coriolis/tests/integration/test_pagination.py

@@ -23,6 +23,7 @@ class PaginationTest(base.CoriolisIntegrationTestBase):
 
     TRANSFER_COUNT = 5
     EXECUTIONS_PER_TRANSFER = 5
+    DEPLOYMENTS_PER_TRANSFER = 5
 
     @classmethod
     def setUpClass(cls):
@@ -84,6 +85,23 @@ class PaginationTest(base.CoriolisIntegrationTestBase):
         db_api.add_endpoint(cls._admin_ctx, endpoint)
         return endpoint
 
+    @classmethod
+    def _create_deployment(
+        cls,
+        transfer_id,
+        origin_endpoint_id: str,
+        destination_endpoint_id: str,
+        **kwargs,
+    ) -> models.Deployment:
+        kwargs["id"] = kwargs.get("id", str(uuid.uuid4()))
+        kwargs["transfer_id"] = transfer_id
+        kwargs["origin_endpoint_id"] = origin_endpoint_id
+        kwargs["destination_endpoint_id"] = destination_endpoint_id
+        deployment = models.Deployment(
+            **kwargs)
+        db_api.add_deployment(cls._admin_ctx, deployment)
+        return deployment
+
     @classmethod
     def _setup_mocks(cls):
         cls._src_endpoint = cls._create_endpoint()
@@ -91,6 +109,8 @@ class PaginationTest(base.CoriolisIntegrationTestBase):
 
         cls._transfers = []
         cls._executions = {}
+        cls._deployments = {}
+        cls._all_deployments = []
         for transfer_idx in range(cls.TRANSFER_COUNT):
             # For testing purposes, we'll set the "created_at" field
             # explicitly, adding a small time delta between records.
@@ -109,6 +129,17 @@ class PaginationTest(base.CoriolisIntegrationTestBase):
                         seconds=execution_idx))
                 cls._executions[transfer.id].append(execution)
 
+            cls._deployments[transfer.id] = []
+            for deployment_idx in range(cls.DEPLOYMENTS_PER_TRANSFER):
+                deployment = cls._create_deployment(
+                    transfer_id=transfer.id,
+                    origin_endpoint_id=cls._src_endpoint.id,
+                    destination_endpoint_id=cls._dst_endpoint.id,
+                    created_at=timeutils.utcnow() + datetime.timedelta(
+                        seconds=deployment_idx))
+                cls._deployments[transfer.id].append(deployment)
+                cls._all_deployments.append(deployment)
+
     @staticmethod
     def _get_record_summary(record):
         # Extract a few fields from the db records and entries returned by
@@ -170,4 +201,41 @@ class PaginationTest(base.CoriolisIntegrationTestBase):
 
         exp_sorted_exec_summary = [
             self._get_record_summary(e) for e in sorted_exp_exec][2:4]
-        self.assertEqual(exp_sorted_exec_summary, ret_exec_summary)
+        self.assertEqual(exp_sorted_exec_summary, ret_exec_summary)
+
+    def test_deployment_list(self):
+        deployments = self._client.deployments.list()
+        ret_depl_summary = [self._get_record_summary(d) for d in deployments]
+
+        exp_sorted_depl_summary = [
+            self._get_record_summary(d) for d in self._all_deployments]
+        exp_sorted_depl_summary = sorted(
+            exp_sorted_depl_summary,
+            key=lambda x: (x["created_at"], x["id"]),
+            reverse=True)
+        self.assertEqual(exp_sorted_depl_summary, ret_depl_summary)
+
+    def test_deployment_list_pagination(self):
+        # Get the first 2 entries, sorted by ID in ascending order.
+        deployments = self._client.deployments.list(
+            limit=2,
+            sort_keys=['id'],
+            sort_dirs=['asc'])
+        ret_depl_summary = [self._get_record_summary(d) for d in deployments]
+
+        exp_sorted_depl_summary = [
+            self._get_record_summary(d) for d in self._all_deployments]
+        exp_sorted_depl_summary = sorted(
+            exp_sorted_depl_summary,
+            key=lambda x: x["id"])
+        self.assertEqual(exp_sorted_depl_summary[:2], ret_depl_summary)
+
+        # Get the next 2 entries.
+        deployments = self._client.deployments.list(
+            limit=2,
+            sort_keys=['id'],
+            sort_dirs=['asc'],
+            marker=ret_depl_summary[-1]["id"],
+        )
+        ret_depl_summary = [self._get_record_summary(d) for d in deployments]
+        self.assertEqual(exp_sorted_depl_summary[2:4], ret_depl_summary)