Add transfer pagination

coriolis/api/v1/transfers.py

@@ -1,6 +1,7 @@
 # Copyright 2016 Cloudbase Solutions Srl
 # All Rights Reserved.
 
+from coriolis.api import common
 from coriolis.api.v1 import utils as api_utils
 from coriolis.api.v1.views import transfer_tasks_execution_view
 from coriolis.api.v1.views import transfer_view
@@ -49,11 +50,16 @@ class TransferController(api_wsgi.Controller):
         context.can(transfer_policies.get_transfers_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 transfer_view.collection(
             self._transfer_api.get_transfers(
                 context,
                 include_tasks_executions=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):
         return self._list(req)

coriolis/conductor/rpc/client.py

@@ -206,11 +206,18 @@ class ConductorClient(rpc.BaseRPCClient):
             skip_os_morphing=skip_os_morphing)
 
     def get_transfers(self, ctxt, include_tasks_executions=False,
-                      include_task_info=False):
+                      include_task_info=False,
+                      marker=None, limit=None,
+                      sort_keys=None, sort_dirs=None):
         return self._call(
             ctxt, 'get_transfers',
             include_tasks_executions=include_tasks_executions,
-            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_transfer(self, ctxt, transfer_id, include_task_info=False):
         return self._call(

coriolis/conductor/rpc/server.py

@@ -1215,10 +1215,17 @@ class ConductorServerEndpoint(object):
 
     @staticmethod
     def get_transfers(ctxt, include_tasks_executions=False,
-                      include_task_info=False):
+                      include_task_info=False,
+                      marker=None, limit=None,
+                      sort_keys=None, sort_dirs=None):
         return db_api.get_transfers(
             ctxt, include_tasks_executions=include_tasks_executions,
-            include_task_info=include_task_info, to_dict=True)
+            include_task_info=include_task_info,
+            marker=marker,
+            limit=limit,
+            sort_keys=sort_keys,
+            sort_dirs=sort_dirs,
+            to_dict=True)
 
     @transfer_synchronized
     def get_transfer(self, ctxt, transfer_id, include_task_info=False):

coriolis/db/api.py

@@ -454,6 +454,10 @@ def get_transfers(context,
                   transfer_scenario=None,
                   include_tasks_executions=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.Transfer)
     if include_tasks_executions:
@@ -466,6 +470,23 @@ def get_transfers(context,
     if is_user_context(context):
         q = q.filter(
             models.Transfer.project_id == context.project_id)
+
+    sort_keys, sort_dirs = process_sort_params(
+        sort_keys,
+        sort_dirs,
+    )
+    if marker:
+        try:
+            marker = get_transfer(context, marker)
+        except exception.NotFound:
+            raise exception.MarkerNotFound(marker=marker)
+    q = sqlalchemy_utils.paginate_query(
+        q, models.Transfer, limit,
+        sort_keys=sort_keys,
+        sort_dirs=sort_dirs,
+        marker=marker,
+    )
+
     db_result = q.all()
     if to_dict:
         return [

coriolis/tests/integration/test_pagination.py

@@ -239,3 +239,39 @@ class PaginationTest(base.CoriolisIntegrationTestBase):
         )
         ret_depl_summary = [self._get_record_summary(d) for d in deployments]
         self.assertEqual(exp_sorted_depl_summary[2:4], ret_depl_summary)
+
+    def test_transfer_list(self):
+        transfers = self._client.transfers.list()
+        ret_transfer_summary = [self._get_record_summary(t) for t in transfers]
+
+        exp_sorted_transfer_summary = [
+            self._get_record_summary(d) for d in self._transfers]
+        exp_sorted_transfer_summary = sorted(
+            exp_sorted_transfer_summary,
+            key=lambda x: (x["created_at"], x["id"]),
+            reverse=True)
+        self.assertEqual(exp_sorted_transfer_summary, ret_transfer_summary)
+
+    def test_transfer_list_pagination(self):
+        # Get the first 2 entries, sorted by ID in ascending order.
+        transfers = self._client.transfers.list(
+            limit=2,
+            sort_keys=['id'],
+            sort_dirs=['asc'])
+        ret_transfer_summary = [self._get_record_summary(t) for t in transfers]
+
+        exp_sorted_transfer_summary = [
+            self._get_record_summary(d) for d in self._transfers]
+        exp_sorted_transfer_summary = sorted(
+            exp_sorted_transfer_summary,
+            key=lambda x: x["id"])
+        self.assertEqual(exp_sorted_transfer_summary[:2], ret_transfer_summary)
+
+        # Get the next 2 entries.
+        transfers = self._client.transfers.list(
+            limit=2,
+            sort_keys=['id'],
+            sort_dirs=['asc'],
+            marker=transfers[-1].id)
+        ret_transfer_summary = [self._get_record_summary(t) for t in transfers]
+        self.assertEqual(exp_sorted_transfer_summary[2:4], ret_transfer_summary)

coriolis/transfers/api.py

@@ -32,10 +32,14 @@ class API(object):
         self._rpc_client.delete_transfer(ctxt, transfer_id)
 
     def get_transfers(self, ctxt, include_tasks_executions=False,
-                      include_task_info=False):
+                      include_task_info=False,
+                      marker=None, limit=None,
+                      sort_keys=None, sort_dirs=None):
         return self._rpc_client.get_transfers(
             ctxt, include_tasks_executions,
-            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_transfer(self, ctxt, transfer_id, include_task_info=False):
         return self._rpc_client.get_transfer(