Query programs by name and search

qiskit_ibm_runtime/api/clients/runtime.py

@@ -45,17 +45,20 @@ 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 = "", search: str = "", limit: int = None, skip: int = None
+    ) -> Dict[str, Any]:
         """Return a list of runtime programs.
 
         Args:
+            name: Name of the program.
+            search: Returns programs containing search word in name or description.
             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, search, limit, skip)
 
     def program_create(
         self,

qiskit_ibm_runtime/api/rest/runtime.py

@@ -56,18 +56,25 @@ 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 = "", search: str = "", limit: int = None, skip: int = None
+    ) -> Dict[str, Any]:
         """Return a list of runtime programs.
 
         Args:
+            name: Name of the program.
+            search: Returns programs containing search word in name or description.
             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 search:
+            payload["search"] = search
         if limit:
             payload["limit"] = limit
         if skip:

qiskit_ibm_runtime/ibm_runtime_service.py

@@ -641,6 +641,8 @@ def pprint_programs(
         self,
         refresh: bool = False,
         detailed: bool = False,
+        name: Optional[str] = "",
+        search: Optional[str] = "",
         limit: int = 20,
         skip: int = 0,
     ) -> None:
@@ -650,11 +652,13 @@ 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.
+            search: Returns programs containing search word in name or description.
             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, search, limit, skip)
         for prog in programs:
             print("=" * 50)
             if detailed:
@@ -667,7 +671,12 @@ 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] = "",
+        search: Optional[str] = "",
+        limit: int = 20,
+        skip: int = 0,
     ) -> List[RuntimeProgram]:
         """Return available runtime programs.
 
@@ -676,35 +685,52 @@ 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.
+            search: Returns programs containing search word in name or description.
             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 and not search):
+            self._programs = self._retrieve_programs()
+            already_retrieved = True
         if limit is None:
             limit = len(self._programs)
+        if name and search:
+            matched_programs = []
+            if refresh and not already_retrieved:
+                matched_programs = list(self._retrieve_programs(name, search).values())
+            else:
+                for program in list(self._programs.values()):
+                    if program.name == name and (
+                        search in program.name or search in program.description
+                    ):
+                        matched_programs.append(program)
+            return matched_programs[skip : limit + skip]
+        elif 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]
+        elif search:
+            matched_programs = []
+            if refresh and not already_retrieved:
+                matched_programs = list(self._retrieve_programs(search).values())
+            else:
+                for program in list(self._programs.values()):
+                    if search in program.name or search in program.description:
+                        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:
@@ -738,6 +764,38 @@ def program(self, program_id: str, refresh: bool = False) -> RuntimeProgram:
 
         return self._programs[program_id]
 
+    def _retrieve_programs(
+        self, name: str = "", search: str = ""
+    ) -> Dict[str, RuntimeProgram]:
+        """Make an API call to fetch programs.
+
+        Args:
+            name: Name of the program.
+            search: Search query for program name and description.
+
+        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, search=search, 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`` and ``search`` parameters have been added to 
+    :meth:`qiskit_ibm_runtime.IBMRuntimeService.programs` and 
+    :meth:`qiskit_ibm_runtime.IBMRuntimeService.pprint_programs`.
+    ``name`` can be used to filter by program names and ``search`` can be used to 
+    filter by keywords in the program name and description. 

test/integration/test_program.py

@@ -33,7 +33,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:
@@ -57,6 +57,43 @@ 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", refresh=True)
+        all_ids = [prog.program_id for prog in programs]
+        self.assertIn(program_id, all_ids)
+        programs = service.programs(name="qiskit-test", refresh=True)
+        all_ids = [prog.program_id for prog in programs]
+        self.assertNotIn(program_id, all_ids)
+
+    @run_integration_test
+    def test_filter_programs_with_search(self, service):
+        """Test filtering programs with the search parameter"""
+        program_id = self._upload_program(service)
+        programs = service.programs(search="qiskit-test", refresh=True)
+        all_ids = [prog.program_id for prog in programs]
+        self.assertIn(program_id, all_ids)
+        programs = service.programs(search="qiskit-test-not", refresh=True)
+        all_ids = [prog.program_id for prog in programs]
+        self.assertNotIn(program_id, all_ids)
+
+    @run_integration_test
+    def test_filter_programs_with_name_and_search(self, service):
+        """Test filtering programs with both search and name parameter"""
+        program_id = self._upload_program(service, name="qiskit-test-sample")
+        programs = service.programs(
+            search="qiskit-test", name="qiskit-test-sample", refresh=True
+        )
+        all_ids = [prog.program_id for prog in programs]
+        self.assertIn(program_id, all_ids)
+        programs = service.programs(
+            search="qiskit-test", name="qiskit-test", refresh=True
+        )
+        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/unit/mock/fake_runtime_client.py

@@ -270,11 +270,25 @@ 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, search, limit, skip):
         """List all programs."""
         programs = []
         for prog in self._programs.values():
-            programs.append(prog.to_dict())
+            if not name and not search:
+                programs.append(prog.to_dict())
+            elif name == prog.to_dict()["name"] and (
+                search in prog.to_dict()["name"].lower()
+                or search in prog.to_dict()["description"].lower()
+            ):
+                programs.append(prog.to_dict())
+            elif name == prog.to_dict()["name"]:
+                programs.append(prog.to_dict())
+            elif (
+                search in prog.to_dict()["name"].lower()
+                or search in prog.to_dict()["description"].lower()
+            ):
+                programs.append(prog.to_dict())
+
         return {"programs": programs[skip : limit + skip], "count": len(self._programs)}
 
     def program_create(

test/unit/test_programs.py

@@ -54,6 +54,39 @@ 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_filter_programs_with_search(self, service):
+        """Test filtering programs with the search parameter"""
+        program_id = upload_program(service)
+        programs = service.programs(search="Test program")
+        all_ids = [prog.program_id for prog in programs]
+        self.assertIn(program_id, all_ids)
+        programs = service.programs(search="Test sample")
+        all_ids = [prog.program_id for prog in programs]
+        self.assertNotIn(program_id, all_ids)
+
+    @run_legacy_and_cloud_fake
+    def test_filter_programs_with_name_and_search(self, service):
+        """Test filtering programs with both search and name parameter"""
+        program_id = upload_program(service, name="qiskit-test-sample")
+        programs = service.programs(search="qiskit-test", name="qiskit-test-sample")
+        all_ids = [prog.program_id for prog in programs]
+        self.assertIn(program_id, all_ids)
+        programs = service.programs(search="qiskit-test", 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."""