Query programs by name and search

qiskit_ibm_runtime/api/clients/runtime.py

@@ -45,17 +45,19 @@ def __init__(
         )
         self._api = Runtime(self._session)
 
-    def list_programs(self, limit: int = None, skip: int = None) -> Dict[str, Any]:
+    def list_programs(
+        self, name: str = "", limit: int = None, skip: int = None
+    ) -> Dict[str, Any]:
         """Return a list of runtime programs.
 
         Args:
+            name: Name of the program.
             limit: The number of programs to return.
             skip: The number of programs to skip.
-
         Returns:
             A list of runtime programs.
         """
-        return self._api.list_programs(limit, skip)
+        return self._api.list_programs(name, limit, skip)
 
     def program_create(
         self,

qiskit_ibm_runtime/api/rest/runtime.py

@@ -56,18 +56,22 @@ def program_job(self, job_id: str) -> "ProgramJob":
         """
         return ProgramJob(self.session, job_id)
 
-    def list_programs(self, limit: int = None, skip: int = None) -> Dict[str, Any]:
+    def list_programs(
+        self, name: str = "", limit: int = None, skip: int = None
+    ) -> Dict[str, Any]:
         """Return a list of runtime programs.
 
         Args:
+            name: Name of the program.
             limit: The number of programs to return.
             skip: The number of programs to skip.
-
         Returns:
             A list of runtime programs.
         """
         url = self.get_url("programs")
-        payload: Dict[str, int] = {}
+        payload: Dict[str, Union[int, str]] = {}
+        if name:
+            payload["name"] = name
         if limit:
             payload["limit"] = limit
         if skip:

qiskit_ibm_runtime/ibm_runtime_service.py

@@ -632,6 +632,7 @@ def pprint_programs(
         self,
         refresh: bool = False,
         detailed: bool = False,
+        name: Optional[str] = "",
         limit: int = 20,
         skip: int = 0,
     ) -> None:
@@ -641,11 +642,12 @@ def pprint_programs(
             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.
+            name: Only retrieve programs with the exact program name given.
             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, limit, skip)
+        programs = self.programs(refresh, name, limit, skip)
         for prog in programs:
             print("=" * 50)
             if detailed:
@@ -658,7 +660,11 @@ def pprint_programs(
                 print(f"  Description: {prog.description}")
 
     def programs(
-        self, refresh: bool = False, limit: int = 20, skip: int = 0
+        self,
+        refresh: bool = False,
+        name: Optional[str] = "",
+        limit: int = 20,
+        skip: int = 0,
     ) -> List[RuntimeProgram]:
         """Return available runtime programs.
 
@@ -667,35 +673,30 @@ def programs(
         Args:
             refresh: If ``True``, re-query the server for the programs. Otherwise
                 return the cached value.
+            name: Only retrieve programs with the exact program name given.
             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.
         """
+        already_retrieved = False
         if skip is None:
             skip = 0
-        if not self._programs or refresh:
-            self._programs = {}
-            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 not self._programs or (refresh and not name):
+            self._programs = self._retrieve_programs()
+            already_retrieved = True
         if limit is None:
             limit = len(self._programs)
+        if name:
+            matched_programs = []
+            if refresh and not already_retrieved:
+                matched_programs = list(self._retrieve_programs(name).values())
+            else:
+                for program in list(self._programs.values()):
+                    if program.name == name:
+                        matched_programs.append(program)
+            return matched_programs[skip : limit + skip]
         return list(self._programs.values())[skip : limit + skip]
 
     def program(self, program_id: str, refresh: bool = False) -> RuntimeProgram:
@@ -729,6 +730,35 @@ def program(self, program_id: str, refresh: bool = False) -> RuntimeProgram:
 
         return self._programs[program_id]
 
+    def _retrieve_programs(self, name: str = "") -> Dict[str, RuntimeProgram]:
+        """Make an API call to fetch programs.
+
+        Args:
+            name: Name of the program.
+
+        Returns:
+            A dict of ``RuntimeProgram`` instances, keyed by program name.
+        """
+        programs = {}
+        current_page_limit = 20
+        offset = 0
+        while True:
+            response = self._api_client.list_programs(
+                name=name, 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)
+                programs[program.program_id] = program
+            if len(programs) == count:
+                # Stop if there are no more programs returned by the server.
+                break
+            offset += len(program_page)
+        return programs
+
     def _to_program(self, response: Dict) -> RuntimeProgram:
         """Convert server response to ``RuntimeProgram`` instances.
 

releasenotes/notes/query-program-name-22b97d6b4c5731d2.yaml

@@ -0,0 +1,8 @@
+---
+upgrade:
+  - |
+    The ``name`` parameter has been added to 
+    :meth:`qiskit_ibm_runtime.IBMRuntimeService.programs` and 
+    :meth:`qiskit_ibm_runtime.IBMRuntimeService.pprint_programs`
+    which can be used to filter by a specific program name. The
+    ``name`` given must be an exact match with an actual program name. 

test/mock/fake_runtime_client.py

@@ -271,11 +271,14 @@ 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, limit, skip):
+    def list_programs(self, name, limit, skip):
         """List all programs."""
         programs = []
         for prog in self._programs.values():
-            programs.append(prog.to_dict())
+            if not name:
+                programs.append(prog.to_dict())
+            if name == prog.to_dict()["name"]:
+                programs.append(prog.to_dict())
         return {"programs": programs[skip : limit + skip], "count": len(self._programs)}
 
     def program_create(

test/test_integration_program.py

@@ -34,7 +34,7 @@ class TestIntegrationProgram(IBMIntegrationTestCase):
     def test_list_programs(self, service):
         """Test listing programs."""
         program_id = self._upload_program(service)
-        programs = service.programs()
+        programs = service.programs(refresh=True)
         self.assertTrue(programs)
         found = False
         for prog in programs:
@@ -58,6 +58,17 @@ def test_list_programs_with_limit_skip(self, service):
         self.assertIn(all_ids[1], some_ids)
         self.assertIn(all_ids[2], some_ids)
 
+    @run_integration_test
+    def test_filter_programs_with_program_name(self, service):
+        """Test filter programs with the program name"""
+        program_id = self._upload_program(service, name="qiskit-test-sample")
+        programs = service.programs(name="qiskit-test-sample")
+        all_ids = [prog.program_id for prog in programs]
+        self.assertIn(program_id, all_ids)
+        programs = service.programs(name="qiskit-test")
+        all_ids = [prog.program_id for prog in programs]
+        self.assertNotIn(program_id, all_ids)
+
     @run_integration_test
     def test_list_program(self, service):
         """Test listing a single program."""

test/test_programs.py

@@ -55,6 +55,17 @@ def test_list_programs_with_limit_skip(self, service):
         all_ids = [prog.program_id for prog in programs]
         self.assertIn(program_ids[0], all_ids)
 
+    @run_legacy_and_cloud_fake
+    def test_filter_programs_with_program_name(self, service):
+        """Test filter programs with the program name"""
+        program_id = upload_program(service, name="qiskit-test-sample")
+        programs = service.programs(name="qiskit-test-sample")
+        all_ids = [prog.program_id for prog in programs]
+        self.assertIn(program_id, all_ids)
+        programs = service.programs(name="qiskit-test")
+        all_ids = [prog.program_id for prog in programs]
+        self.assertNotIn(program_id, all_ids)
+
     @run_legacy_and_cloud_fake
     def test_list_program(self, service):
         """Test listing a single program."""