Support pagination for retrieving runtime programs

qiskit_ibm/api/clients/runtime.py

@@ -39,13 +39,17 @@ def __init__(
                                      **credentials.connection_parameters())
         self.api = Runtime(self._session)
 
-    def list_programs(self) -> Dict[str, Any]:
+    def list_programs(self, limit: int = None, skip: int = None) -> Dict[str, Any]:
         """Return a list of runtime programs.
 
+        Args:
+            limit: The number of programs to return.
+            skip: The number of programs to skip.
+
         Returns:
-            A list of quantum programs.
+            A list of runtime programs.
         """
-        return self.api.list_programs()
+        return self.api.list_programs(limit, skip)
 
     def program_create(
             self,

qiskit_ibm/api/rest/runtime.py

@@ -55,14 +55,23 @@ def program_job(self, job_id: str) -> 'ProgramJob':
         """
         return ProgramJob(self.session, job_id)
 
-    def list_programs(self) -> Dict[str, Any]:
+    def list_programs(self, limit: int = None, skip: int = None) -> Dict[str, Any]:
         """Return a list of runtime programs.
 
+        Args:
+            limit: The number of programs to return.
+            skip: The number of programs to skip.
+
         Returns:
-            JSON response.
+            A list of runtime programs.
         """
         url = self.get_url('programs')
-        return self.session.get(url).json()
+        payload: Dict[str, int] = {}
+        if limit:
+            payload['limit'] = limit
+        if skip:
+            payload['offset'] = skip
+        return self.session.get(url, params=payload).json()
 
     def create_program(
             self,

qiskit_ibm/runtime/ibm_runtime_service.py

@@ -107,15 +107,19 @@ def __init__(self, provider: 'ibm_provider.IBMProvider') -> None:
         self._ws_url = provider.credentials.runtime_url.replace('https', 'wss')
         self._programs = {}  # type: Dict
 
-    def pprint_programs(self, refresh: bool = False, detailed: bool = False) -> None:
+    def pprint_programs(self, refresh: bool = False, detailed: bool = False,
+                        limit: int = 20, skip: int = 0) -> None:
         """Pretty print information about available runtime programs.
 
         Args:
             refresh: If ``True``, re-query the server for the programs. Otherwise
                 return the cached value.
             detailed: If ``True`` print all details about available runtime programs.
+            limit: The number of programs returned at a time. Default and maximum
+                value of 20.
+            skip: The number of programs to skip.
         """
-        programs = self.programs(refresh)
+        programs = self.programs(refresh, limit, skip)
         for prog in programs:
             print("="*50)
             if detailed:
@@ -125,25 +129,43 @@ def pprint_programs(self, refresh: bool = False, detailed: bool = False) -> None
                 print(f"  Name: {prog.name}")
                 print(f"  Description: {prog.description}")
 
-    def programs(self, refresh: bool = False) -> List[RuntimeProgram]:
+    def programs(self, refresh: bool = False,
+                 limit: int = 20, skip: int = 0) -> List[RuntimeProgram]:
         """Return available runtime programs.
 
         Currently only program metadata is returned.
 
         Args:
             refresh: If ``True``, re-query the server for the programs. Otherwise
                 return the cached value.
+            limit: The number of programs returned at a time. ``None`` means no limit.
+            skip: The number of programs to skip.
 
         Returns:
             A list of runtime programs.
         """
+        if skip is None:
+            skip = 0
         if not self._programs or refresh:
             self._programs = {}
-            response = self._api_client.list_programs()
-            for prog_dict in response.get("programs", []):
-                program = self._to_program(prog_dict)
-                self._programs[program.program_id] = program
-        return list(self._programs.values())
+            current_page_limit = 20
+            offset = 0
+            while True:
+                response = self._api_client.list_programs(limit=current_page_limit, skip=offset)
+                program_page = response.get("programs", [])
+                # count is the total number of programs that would be returned if
+                # there was no limit or skip
+                count = response.get("count", 0)
+                for prog_dict in program_page:
+                    program = self._to_program(prog_dict)
+                    self._programs[program.program_id] = program
+                if len(self._programs) == count:
+                    # Stop if there are no more programs returned by the server.
+                    break
+                offset += len(program_page)
+        if limit is None:
+            limit = len(self._programs)
+        return list(self._programs.values())[skip:limit+skip]
 
     def program(self, program_id: str, refresh: bool = False) -> RuntimeProgram:
         """Retrieve a runtime program.

releasenotes/notes/runtime-program-pagination-8d599ae984a5ce33.yaml

@@ -0,0 +1,9 @@
+---
+upgrade:
+  - |
+    ``limit`` and ``skip`` parameters have been added to 
+    :meth:`qiskit_ibm.runtime.IBMRuntimeService.programs` and 
+    :meth:`qiskit_ibm.runtime.IBMRuntimeService.pprint_programs`.
+    ``limit`` can be used to set the number of runtime programs returned 
+    and ``skip`` is the number of programs to skip when retrieving 
+    programs.

test/ibm/runtime/fake_runtime_client.py

@@ -242,12 +242,12 @@ def set_final_status(self, final_status):
         """Set job status to passed in final status instantly."""
         self._final_status = final_status
 
-    def list_programs(self):
+    def list_programs(self, limit, skip):
         """List all programs."""
         programs = []
         for prog in self._programs.values():
             programs.append(prog.to_dict())
-        return {"programs": programs}
+        return {"programs": programs[skip:limit+skip], "count": len(self._programs)}
 
     def program_create(self, program_data, name, description, max_execution_time,
                        spec=None, is_public=False):

test/ibm/runtime/test_runtime.py

@@ -302,6 +302,20 @@ def test_list_programs(self):
         all_ids = [prog.program_id for prog in programs]
         self.assertIn(program_id, all_ids)
 
+    def test_list_programs_with_limit_skip(self):
+        """Test listing programs with limit and skip."""
+        program_1 = self._upload_program()
+        program_2 = self._upload_program()
+        program_3 = self._upload_program()
+        programs = self.runtime.programs(limit=2, skip=1)
+        all_ids = [prog.program_id for prog in programs]
+        self.assertNotIn(program_1, all_ids)
+        self.assertIn(program_2, all_ids)
+        self.assertIn(program_3, all_ids)
+        programs = self.runtime.programs(limit=3)
+        all_ids = [prog.program_id for prog in programs]
+        self.assertIn(program_1, all_ids)
+
     def test_list_program(self):
         """Test listing a single program."""
         program_id = self._upload_program()

test/ibm/runtime/test_runtime_integration.py

@@ -141,6 +141,21 @@ def test_list_programs(self):
                 found = True
         self.assertTrue(found, f"Program {self.program_id} not found!")
 
+    def test_list_programs_with_limit_skip(self):
+        """Test listing programs with limit and skip."""
+        self._upload_program()
+        self._upload_program()
+        self._upload_program()
+        programs = self.provider.runtime.programs(limit=3)
+        all_ids = [prog.program_id for prog in programs]
+        self.assertEqual(len(all_ids), 3)
+        programs = self.provider.runtime.programs(limit=2, skip=1)
+        some_ids = [prog.program_id for prog in programs]
+        self.assertEqual(len(some_ids), 2)
+        self.assertNotIn(all_ids[0], some_ids)
+        self.assertIn(all_ids[1], some_ids)
+        self.assertIn(all_ids[2], some_ids)
+
     def test_list_program(self):
         """Test listing a single program."""
         program = self.provider.runtime.program(self.program_id)