Query programs by name

qiskit_ibm_runtime/api/clients/runtime.py

@@ -42,17 +42,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

@@ -51,18 +51,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

@@ -171,6 +171,7 @@ def __init__(
         )
         self._backends: Dict[str, "ibm_backend.IBMBackend"] = {}
         self._api_client = None
+        self._programs_first_call = True
         hgps = self._get_hgps()
         for hgp in hgps:
             for name, backend in hgp.backends.items():
@@ -811,6 +812,7 @@ def pprint_programs(
         self,
         refresh: bool = False,
         detailed: bool = False,
+        name: Optional[str] = "",
         limit: int = 20,
         skip: int = 0,
     ) -> None:
@@ -820,11 +822,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:
@@ -837,7 +840,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.
 
@@ -846,21 +853,24 @@ 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.
         """
+        refreshed = False
         if skip is None:
             skip = 0
         if not self._programs or refresh:
             self._programs = {}
             current_page_limit = 20
             offset = 0
+            program_name = "" if self._programs_first_call else name
             while True:
                 response = self._api_client.list_programs(
-                    limit=current_page_limit, skip=offset
+                    name=program_name, limit=current_page_limit, skip=offset
                 )
                 program_page = response.get("programs", [])
                 # count is the total number of programs that would be returned if
@@ -870,11 +880,22 @@ def programs(
                     program = self._to_program(prog_dict)
                     self._programs[program.program_id] = program
                 if len(self._programs) == count:
+                    if not self._programs_first_call:
+                        refreshed = True
+                    self._programs_first_call = False
                     # Stop if there are no more programs returned by the server.
                     break
                 offset += len(program_page)
         if limit is None:
             limit = len(self._programs)
+        if name:
+            if refreshed:
+                return list(self._programs.values())[skip : limit + skip]
+            matched_programs = []
+            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:

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/ibm/runtime/fake_runtime_client.py

@@ -260,11 +260,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/ibm/runtime/test_runtime.py

@@ -329,6 +329,16 @@ def test_list_programs(self):
         all_ids = [prog.program_id for prog in programs]
         self.assertIn(program_id, all_ids)
 
+    def test_filter_programs_with_program_name(self):
+        """Test filter programs with the program name"""
+        program_id = self._upload_program(name="qiskit-test-sample")
+        programs = self.service.programs(name="qiskit-test-sample")
+        all_ids = [prog.program_id for prog in programs]
+        self.assertIn(program_id, all_ids)
+        programs = self.service.programs(name="qiskit-test")
+        all_ids = [prog.program_id for prog in programs]
+        self.assertNotIn(program_id, all_ids)
+
     def test_list_programs_with_limit_skip(self):
         """Test listing programs with limit and skip."""
         program_1 = self._upload_program()

test/ibm/runtime/test_runtime_integration.py

@@ -139,6 +139,16 @@ def test_list_programs(self):
                 found = True
         self.assertTrue(found, f"Program {self.program_id} not found!")
 
+    def test_filter_programs_with_program_name(self):
+        """Test filter programs with program name."""
+        program_id = self._upload_program(name="qiskit-test-sample")
+        programs = self.service.programs(name="qiskit-test-sample")
+        all_ids = [prog.program_id for prog in programs]
+        self.assertIn(program_id, all_ids)
+        programs = self.service.programs(name="qiskit-test")
+        all_ids = [prog.program_id for prog in programs]
+        self.assertNotIn(program_id, all_ids)
+
     def test_list_programs_with_limit_skip(self):
         """Test listing programs with limit and skip."""
         self._upload_program()