Allow updating runtime metadata in place

qiskit_ibm/api/clients/runtime.py

@@ -87,17 +87,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.
 
@@ -145,14 +134,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 (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.

qiskit_ibm/api/rest/runtime.py

@@ -194,15 +194,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')
@@ -222,8 +213,8 @@ 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 (base64 encoded).
@@ -232,6 +223,34 @@ def update(self, program_data: str) -> None:
         self.session.put(url, data=program_data,
                          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):
     """Rest adapter for program job related endpoints."""

qiskit_ibm/runtime/ibm_runtime_service.py

@@ -16,6 +16,7 @@
 from typing import Dict, Callable, Optional, Union, List, Any, Type
 import json
 import re
+import warnings
 
 from qiskit.providers.exceptions import QiskitBackendNotFoundError
 from qiskit_ibm import ibm_provider  # pylint: disable=unused-import
@@ -373,30 +374,83 @@ def _read_metadata(
     def update_program(
             self,
             program_id: str,
-            data: str,
+            data: str = None,
+            metadata: Optional[Union[Dict, str]] = None,
+            name: str = None,
+            description: str = None,
+            max_execution_time: int = None,
+            spec: Optional[Dict] = None
     ) -> None:
         """Update a runtime program.
 
+        Program metadata can be specified using the `metadata` parameter or
+        individual parameters, such as `name` and `description`. If the
+        same metadata field is specified in both places, the individual parameter
+        takes precedence.
+
         Args:
             program_id: Program ID.
             data: Program data or path of the file containing program data to upload.
+            metadata: Name of the program metadata file or metadata dictionary.
+            name: New program name.
+            description: New program description.
+            max_execution_time: New maximum execution time.
+            spec: New specifications for backend characteristics, input parameters,
+                interim results and final result.
 
         Raises:
             RuntimeProgramNotFound: If the program doesn't exist.
             QiskitRuntimeError: If the request failed.
         """
-        if "def main(" not in data:
-            # This is the program file
-            with open(data, "r") as file:
-                data = file.read()
+        if not any([data, metadata, name, description, max_execution_time, spec]):
+            warnings.warn("None of the 'data', 'metadata', 'name', 'description', "
+                          "'max_execution_time', or 'spec' parameters is specified. "
+                          "No update is made.")
+            return
+
+        if data:
+            if "def main(" not in data:
+                # This is the program file
+                with open(data, "r") as file:
+                    data = file.read()
+            data = to_base64_string(data)
+
+        if metadata:
+            metadata = self._read_metadata(metadata=metadata)
+        combined_metadata = self._merge_metadata(
+            metadata=metadata, name=name, description=description,
+            max_execution_time=max_execution_time, spec=spec)
+
         try:
-            program_data = to_base64_string(data)
-            self._api_client.program_update(program_id, program_data)
+            self._api_client.program_update(
+                program_id, program_data=data, **combined_metadata)
         except RequestsApiError as ex:
             if ex.status_code == 404:
                 raise RuntimeProgramNotFound(f"Program not found: {ex.message}") from None
             raise QiskitRuntimeError(f"Failed to update program: {ex}") from None
 
+    def _merge_metadata(
+            self,
+            metadata: Optional[Dict] = None,
+            **kwargs: Any
+    ) -> Dict:
+        """Merge multiple copies of metadata.
+        Args:
+            metadata: Program metadata.
+            **kwargs: Additional metadata fields to overwrite.
+        Returns:
+            Merged metadata.
+        """
+        merged = {}
+        metadata = metadata or {}
+        metadata_keys = ['name', 'max_execution_time', 'description', 'spec']
+        for key in metadata_keys:
+            if kwargs.get(key, None) is not None:
+                merged[key] = kwargs[key]
+            elif key in metadata.keys():
+                merged[key] = metadata[key]
+        return merged
+
     def delete_program(self, program_id: str) -> None:
         """Delete a runtime program.
 

releasenotes/notes/update-runtime-metadata-d2ddbcfc0d034530.yaml

@@ -0,0 +1,9 @@
+---
+features:
+  - |
+    You can now use the :meth:`qiskit_ibm.runtime.IBMRuntimeService.update_program`
+    method to update the metadata for a Qiskit Runtime program.
+    Program metadata can be specified using the ``metadata`` parameter or
+    individual parameters, such as ``name`` and ``description``. If the
+    same metadata field is specified in both places, the individual parameter
+    takes precedence.

test/ibm/runtime/fake_runtime_client.py

@@ -15,7 +15,8 @@
 import time
 import uuid
 import json
-from typing import Optional
+import base64
+from typing import Optional, Dict
 from concurrent.futures import ThreadPoolExecutor
 
 from qiskit_ibm.credentials import Credentials
@@ -51,7 +52,7 @@ def to_dict(self, include_data=False):
                'creation_date': '2021-09-13T17:27:42Z',
                'update_date': '2021-09-14T19:25:32Z'}
         if include_data:
-            out['data'] = self._data
+            out['data'] = base64.standard_b64decode(self._data).decode()
         out['spec'] = {}
         if self._backend_requirements:
             out['spec']['backend_requirements'] = self._backend_requirements
@@ -255,15 +256,35 @@ def program_create(self, program_data, name, description, max_execution_time,
             is_public=is_public)
         return {'id': program_id}
 
+    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."""
+        if program_id not in self._programs:
+            raise RequestsApiError("Program not found", status_code=404)
+        program = self._programs[program_id]
+        program._data = program_data or program._data
+        program._name = name or program._name
+        program._description = description or program._description
+        program._cost = max_execution_time or program._cost
+        if spec:
+            program._backend_requirements = \
+                spec.get("backend_requirements") or program._backend_requirements
+            program._parameters = spec.get("parameters") or program._parameters
+            program._return_values = spec.get("return_values") or program._return_values
+            program._interim_results = spec.get("interim_results") or program._interim_results
+
     def program_get(self, program_id: str):
         """Return a specific program."""
         if program_id not in self._programs:
             raise RequestsApiError("Program not found", status_code=404)
-        return self._programs[program_id].to_dict()
-
-    def program_get_data(self, program_id: str):
-        """Return a specific program and its data."""
-        return self._programs[program_id].to_dict(iclude_data=True)
+        return self._programs[program_id].to_dict(include_data=True)
 
     def program_run(
             self,

test/ibm/runtime/test_runtime.py

@@ -340,6 +340,53 @@ def test_upload_program(self):
         self.assertEqual(max_execution_time, program.max_execution_time)
         self.assertEqual(program.is_public, is_public)
 
+    def test_update_program(self):
+        """Test updating program."""
+        new_data = "def main() {foo=bar}"
+        new_metadata = copy.deepcopy(self.DEFAULT_METADATA)
+        new_metadata["name"] = "test_update_program"
+        new_name = "name2"
+        new_description = "some other description"
+        new_cost = self.DEFAULT_METADATA["max_execution_time"] + 100
+        new_spec = copy.deepcopy(self.DEFAULT_METADATA["spec"])
+        new_spec["backend_requirements"] = {"input_allowed": "runtime"}
+
+        sub_tests = [
+            {"data": new_data},
+            {"metadata": new_metadata},
+            {"data": new_data, "metadata": new_metadata},
+            {"metadata": new_metadata, "name": new_name},
+            {"data": new_data, "metadata": new_metadata, "description": new_description},
+            {"max_execution_time": new_cost, "spec": new_spec}
+        ]
+
+        for new_vals in sub_tests:
+            with self.subTest(new_vals=new_vals.keys()):
+                program_id = self._upload_program()
+                self.runtime.update_program(program_id=program_id, **new_vals)
+                updated = self.runtime.program(program_id, refresh=True)
+                if "data" in new_vals:
+                    raw_program = self.runtime._api_client.program_get(program_id)
+                    self.assertEqual(new_data, raw_program["data"])
+                if "metadata" in new_vals and "name" not in new_vals:
+                    self.assertEqual(new_metadata["name"], updated.name)
+                if "name" in new_vals:
+                    self.assertEqual(new_name, updated.name)
+                if "description" in new_vals:
+                    self.assertEqual(new_description, updated.description)
+                if "max_execution_time" in new_vals:
+                    self.assertEqual(new_cost, updated.max_execution_time)
+                if "spec" in new_vals:
+                    raw_program = self.runtime._api_client.program_get(program_id)
+                    self.assertEqual(new_spec, raw_program["spec"])
+
+    def test_update_program_no_new_fields(self):
+        """Test updating a program without any new data."""
+        program_id = self._upload_program()
+        with warnings.catch_warnings(record=True) as warn_cm:
+            self.runtime.update_program(program_id=program_id)
+            self.assertEqual(len(warn_cm), 1)
+
     def test_delete_program(self):
         """Test deleting program."""
         program_id = self._upload_program()

test/ibm/runtime/test_runtime_integration.py

@@ -216,8 +216,8 @@ def test_double_delete_program(self):
         with self.assertRaises(RuntimeProgramNotFound):
             self.provider.runtime.delete_program(program_id)
 
-    def test_update_program(self):
-        """Test updating a program."""
+    def test_update_program_data(self):
+        """Test updating program data."""
         program_v1 = """
 def main(backend, user_messenger, **kwargs):
     return "version 1"
@@ -226,13 +226,36 @@ def main(backend, user_messenger, **kwargs):
 def main(backend, user_messenger, **kwargs):
     return "version 2"
         """
+        # TODO retrieve program data instead of run program when #66 is merged
         program_id = self._upload_program(data=program_v1)
         job = self._run_program(program_id=program_id)
         self.assertEqual("version 1", job.result())
         self.provider.runtime.update_program(program_id=program_id, data=program_v2)
         job = self._run_program(program_id=program_id)
         self.assertEqual("version 2", job.result())
 
+    def test_update_program_metadata(self):
+        """Test updating program metadata."""
+        program_id = self._upload_program()
+        original = self.provider.runtime.program(program_id)
+        new_metadata = {
+            "name": self._get_program_name(),
+            "description": "test_update_program_metadata",
+            "max_execution_time": original.max_execution_time + 100,
+            "spec": {
+                "return_values": {
+                    "type": "object",
+                    "description": "Some return value"
+                }
+            }
+        }
+        self.provider.runtime.update_program(program_id=program_id, metadata=new_metadata)
+        updated = self.provider.runtime.program(program_id, refresh=True)
+        self.assertEqual(new_metadata["name"], updated.name)
+        self.assertEqual(new_metadata["description"], updated.description)
+        self.assertEqual(new_metadata["max_execution_time"], updated.max_execution_time)
+        self.assertEqual(new_metadata["spec"]["return_values"], updated.return_values)
+
     def test_run_program(self):
         """Test running a program."""
         job = self._run_program(final_result="foo")