Query programs by name

qiskit_ibm/api/clients/runtime.py

@@ -13,7 +13,7 @@
 """Client for accessing IBM Quantum runtime service."""
 
 import logging
-from typing import List, Dict, Union, Optional
+from typing import Any, List, Dict, Union, Optional
 
 from qiskit_ibm.credentials import Credentials
 from qiskit_ibm.api.session import RetrySession
@@ -39,40 +39,35 @@ def __init__(
                                      **credentials.connection_parameters())
         self.api = Runtime(self._session)
 
-    def list_programs(self) -> List[Dict]:
+    def list_programs(self, name: str) -> Dict[str, Any]:
         """Return a list of runtime programs.
 
+        Args:
+            name: Name of the program.
+
         Returns:
             A list of quantum programs.
         """
-        return self.api.list_programs()
+        return self.api.list_programs(name)
 
     def program_create(
             self,
-            program_data: bytes,
+            program_data: str,
             name: str,
             description: str,
             max_execution_time: int,
             is_public: Optional[bool] = False,
-            version: Optional[str] = None,
-            backend_requirements: Optional[Dict] = None,
-            parameters: Optional[Dict] = None,
-            return_values: Optional[List] = None,
-            interim_results: Optional[List] = None
+            spec: Optional[Dict] = None
     ) -> Dict:
         """Create a new program.
 
         Args:
             name: Name of the program.
-            program_data: Program data.
+            program_data: Program data (base64 encoded).
             description: Program description.
             max_execution_time: Maximum execution time.
             is_public: Whether the program should be public.
-            version: Program version.
-            backend_requirements: Backend requirements.
-            parameters: Program parameters.
-            return_values: Program return values.
-            interim_results: Program interim results.
+            spec: Backend requirements, parameters, interim results, return values, etc.
 
         Returns:
             Server response.
@@ -81,9 +76,7 @@ def program_create(
             program_data=program_data,
             name=name,
             description=description, max_execution_time=max_execution_time,
-            is_public=is_public, version=version, backend_requirements=backend_requirements,
-            parameters=parameters, return_values=return_values,
-            interim_results=interim_results
+            is_public=is_public, spec=spec
         )
 
     def program_get(self, program_id: str) -> Dict:
@@ -97,17 +90,6 @@ def program_get(self, program_id: str) -> Dict:
         """
         return self.api.program(program_id).get()
 
-    def program_get_data(self, program_id: str) -> Dict:
-        """Return a specific program and its data.
-
-        Args:
-            program_id: Program ID.
-
-        Returns:
-            Program information, including data.
-        """
-        return self.api.program(program_id).get_data()
-
     def set_program_visibility(self, program_id: str, public: bool) -> None:
         """Sets a program's visibility.
 
@@ -127,7 +109,7 @@ def program_run(
             program_id: str,
             credentials: Credentials,
             backend_name: str,
-            params: str,
+            params: Dict,
             image: str
     ) -> Dict:
         """Run the specified program.
@@ -155,14 +137,32 @@ def program_delete(self, program_id: str) -> None:
         """
         self.api.program(program_id).delete()
 
-    def program_update(self, program_id: str, program_data: str) -> None:
+    def program_update(
+            self,
+            program_id: str,
+            program_data: str = None,
+            name: str = None,
+            description: str = None,
+            max_execution_time: int = None,
+            spec: Optional[Dict] = None
+    ) -> None:
         """Update a program.
 
         Args:
             program_id: Program ID.
-            program_data: Program data.
+            program_data: Program data (base64 encoded).
+            name: Name of the program.
+            description: Program description.
+            max_execution_time: Maximum execution time.
+            spec: Backend requirements, parameters, interim results, return values, etc.
         """
-        self.api.program(program_id).update(program_data)
+        if program_data:
+            self.api.program(program_id).update_data(program_data)
+
+        if any([name, description, max_execution_time, spec]):
+            self.api.program(program_id).update_metadata(
+                name=name, description=description,
+                max_execution_time=max_execution_time, spec=spec)
 
     def job_get(self, job_id: str) -> Dict:
         """Get job data.
@@ -177,19 +177,26 @@ def job_get(self, job_id: str) -> Dict:
         logger.debug("Runtime job get response: %s", response)
         return response
 
-    def jobs_get(self, limit: int = None, skip: int = None, pending: bool = None) -> Dict:
+    def jobs_get(
+            self,
+            limit: int = None,
+            skip: int = None,
+            pending: bool = None,
+            program_id: str = None
+    ) -> Dict:
         """Get job data for all jobs.
 
         Args:
             limit: Number of results to return.
             skip: Number of results to skip.
             pending: Returns 'QUEUED' and 'RUNNING' jobs if True,
                 returns 'DONE', 'CANCELLED' and 'ERROR' jobs if False.
+            program_id: Filter by Program ID.
 
         Returns:
             JSON response.
         """
-        return self.api.jobs_get(limit=limit, skip=skip, pending=pending)
+        return self.api.jobs_get(limit=limit, skip=skip, pending=pending, program_id=program_id)
 
     def job_results(self, job_id: str) -> str:
         """Get the results of a program job.

qiskit_ibm/api/rest/runtime.py

@@ -19,6 +19,7 @@
 
 from .base import RestAdapterBase
 from ..session import RetrySession
+from ...runtime.utils import RuntimeEncoder
 
 logger = logging.getLogger(__name__)
 
@@ -54,65 +55,54 @@ def program_job(self, job_id: str) -> 'ProgramJob':
         """
         return ProgramJob(self.session, job_id)
 
-    def list_programs(self) -> List[Dict]:
+    def list_programs(self, name: str) -> Dict[str, Any]:
         """Return a list of runtime programs.
 
+        Args:
+            name: Name of the program.
+
         Returns:
             JSON response.
         """
         url = self.get_url('programs')
-        return self.session.get(url).json()
+        payload: Dict[str, Union[str]] = {}
+        if name:
+            payload['name'] = name
+        response = self.session.get(url, params=payload).json()
+        return response
 
     def create_program(
             self,
-            program_data: bytes,
+            program_data: str,
             name: str,
             description: str,
             max_execution_time: int,
             is_public: Optional[bool] = False,
-            version: Optional[str] = None,
-            backend_requirements: Optional[Dict] = None,
-            parameters: Optional[Dict] = None,
-            return_values: Optional[List] = None,
-            interim_results: Optional[List] = None
+            spec: Optional[Dict] = None
     ) -> Dict:
         """Upload a new program.
 
         Args:
-            program_data: Program data.
+            program_data: Program data (base64 encoded).
             name: Name of the program.
             description: Program description.
             max_execution_time: Maximum execution time.
             is_public: Whether the program should be public.
-            version: Program version.
-            backend_requirements: Backend requirements.
-            parameters: Program parameters.
-            return_values: Program return values.
-            interim_results: Program interim results.
+            spec: Backend requirements, parameters, interim results, return values, etc.
 
         Returns:
             JSON response.
         """
         url = self.get_url('programs')
-        data = {'name': name,
-                'cost': str(max_execution_time),
-                'description': description.encode(),
-                'max_execution_time': max_execution_time,
-                'isPublic': is_public}
-        if version is not None:
-            data['version'] = version
-        if backend_requirements:
-            data['backendRequirements'] = json.dumps(backend_requirements)
-        if parameters:
-            data['parameters'] = json.dumps({"doc": parameters})
-        if return_values:
-            data['returnValues'] = json.dumps(return_values)
-        if interim_results:
-            data['interimResults'] = json.dumps(interim_results)
-
-        files = {'program': (name, program_data)}  # type: ignore[dict-item]
-        response = self.session.post(url, data=data, files=files).json()
-        return response
+        payload = {'name': name,
+                   'data': program_data,
+                   'cost': max_execution_time,
+                   'description': description,
+                   'is_public': is_public}
+        if spec is not None:
+            payload['spec'] = spec
+        data = json.dumps(payload)
+        return self.session.post(url, data=data).json()
 
     def program_run(
             self,
@@ -121,7 +111,7 @@ def program_run(
             group: str,
             project: str,
             backend_name: str,
-            params: str,
+            params: Dict,
             image: str
     ) -> Dict:
         """Execute the program.
@@ -140,25 +130,32 @@ def program_run(
         """
         url = self.get_url('jobs')
         payload = {
-            'programId': program_id,
+            'program_id': program_id,
             'hub': hub,
             'group': group,
             'project': project,
             'backend': backend_name,
-            'params': [params],
+            'params': params,
             'runtime': image
         }
-        data = json.dumps(payload)
+        data = json.dumps(payload, cls=RuntimeEncoder)
         return self.session.post(url, data=data).json()
 
-    def jobs_get(self, limit: int = None, skip: int = None, pending: bool = None) -> Dict:
+    def jobs_get(
+            self,
+            limit: int = None,
+            skip: int = None,
+            pending: bool = None,
+            program_id: str = None
+    ) -> Dict:
         """Get a list of job data.
 
         Args:
             limit: Number of results to return.
             skip: Number of results to skip.
             pending: Returns 'QUEUED' and 'RUNNING' jobs if True,
                 returns 'DONE', 'CANCELLED' and 'ERROR' jobs if False.
+            program_id: Filter by Program ID.
 
         Returns:
             JSON response.
@@ -171,6 +168,8 @@ def jobs_get(self, limit: int = None, skip: int = None, pending: bool = None) ->
             payload['offset'] = skip
         if pending is not None:
             payload['pending'] = 'true' if pending else 'false'
+        if program_id:
+            payload['program'] = program_id
         return self.session.get(url, params=payload).json()
 
     def logout(self) -> None:
@@ -211,15 +210,6 @@ def get(self) -> Dict[str, Any]:
         url = self.get_url('self')
         return self.session.get(url).json()
 
-    def get_data(self) -> Dict[str, Any]:
-        """Return program information, including data.
-
-        Returns:
-            JSON response.
-        """
-        url = self.get_url('data')
-        return self.session.get(url).json()
-
     def make_public(self) -> None:
         """Sets a runtime program's visibility to public."""
         url = self.get_url('public')
@@ -239,15 +229,43 @@ def delete(self) -> None:
         url = self.get_url('self')
         self.session.delete(url)
 
-    def update(self, program_data: str) -> None:
-        """Update a program.
+    def update_data(self, program_data: str) -> None:
+        """Update program data.
 
         Args:
-            program_data: Program data.
+            program_data: Program data (base64 encoded).
         """
         url = self.get_url("data")
         self.session.put(url, data=program_data,
-                         headers={'Content-Type': 'text/plain'})
+                         headers={'Content-Type': 'application/octet-stream'})
+
+    def update_metadata(
+            self,
+            name: str = None,
+            description: str = None,
+            max_execution_time: int = None,
+            spec: Optional[Dict] = None
+    ) -> None:
+        """Update program metadata.
+
+        Args:
+            name: Name of the program.
+            description: Program description.
+            max_execution_time: Maximum execution time.
+            spec: Backend requirements, parameters, interim results, return values, etc.
+        """
+        url = self.get_url("self")
+        payload: Dict = {}
+        if name:
+            payload["name"] = name
+        if description:
+            payload["description"] = description
+        if max_execution_time:
+            payload["cost"] = max_execution_time
+        if spec:
+            payload["spec"] = spec
+
+        self.session.patch(url, json=payload)
 
 
 class ProgramJob(RestAdapterBase):

qiskit_ibm/runtime/__init__.py

@@ -191,14 +191,11 @@ def interim_result_callback(job_id, interim_result):
     provider = IBMProvider()
     program_id = provider.runtime.upload_program(
                     data="my_vqe.py",
-                    metadata="my_vqe_metadata.json",
-                    version="1.2"
+                    metadata="my_vqe_metadata.json"
                 )
 
 In the example above, the file ``my_vqe.py`` contains the program data, and
-``my_vqe_metadata.json`` contains the program metadata. An additional
-parameter ``version`` is also specified, which takes precedence over any
-``version`` value specified in ``my_vqe_metadata.json``.
+``my_vqe_metadata.json`` contains the program metadata.
 
 Method :meth:`IBMRuntimeService.delete_program` allows you to delete a
 program.