Allow runtime program authors to retrieve program data

qiskit_ibm/runtime/ibm_runtime_service.py

@@ -203,7 +203,9 @@ def _to_program(self, response: Dict) -> RuntimeProgram:
                               creation_date=response.get('creation_date', ""),
                               update_date=response.get('update_date', ""),
                               backend_requirements=backend_requirements,
-                              is_public=response.get('is_public', False))
+                              is_public=response.get('is_public', False),
+                              data=response.get('data', ""),
+                              api_client=self._api_client)
 
     def run(
             self,

qiskit_ibm/runtime/runtime_program.py

@@ -16,8 +16,8 @@
 import re
 from typing import Optional, List, Dict
 from types import SimpleNamespace
-from qiskit_ibm.exceptions import IBMInputValueError
-
+from qiskit_ibm.exceptions import IBMInputValueError, IBMNotAuthorizedError
+from ..api.clients.runtime import RuntimeClient
 
 logger = logging.getLogger(__name__)
 
@@ -55,7 +55,9 @@ def __init__(
             backend_requirements: Optional[Dict] = None,
             creation_date: str = "",
             update_date: str = "",
-            is_public: Optional[bool] = False
+            is_public: Optional[bool] = False,
+            data: str = "",
+            api_client: Optional[RuntimeClient] = None
     ) -> None:
         """RuntimeProgram constructor.
 
@@ -71,6 +73,8 @@ def __init__(
             creation_date: Program creation date.
             update_date: Program last updated date.
             is_public: ``True`` if program is visible to all. ``False`` if it's only visible to you.
+            data: Program data.
+            api_client: Runtime api client.
         """
         self._name = program_name
         self._id = program_id
@@ -83,6 +87,8 @@ def __init__(
         self._creation_date = creation_date
         self._update_date = update_date
         self._is_public = is_public
+        self._data = data
+        self._api_client = api_client
 
     def __str__(self) -> str:
         def _format_common(schema: Dict) -> None:
@@ -258,6 +264,41 @@ def is_public(self) -> bool:
         """
         return self._is_public
 
+    @property
+    def data(self) -> str:
+        """Program data.
+
+        Returns:
+            Program data.
+
+        Raises:
+            IBMNotAuthorizedError: if user is not the program author.
+        """
+        if not self._data:
+            response = self._api_client.program_get(self._id)
+            self._backend_requirements = {}
+            self._parameters = {}
+            self._return_values = {}
+            self._interim_results = {}
+            if "spec" in response:
+                self._backend_requirements = response["spec"].get('backend_requirements', {})
+                self._parameters = response["spec"].get('parameters', {})
+                self._return_values = response["spec"].get('return_values', {})
+                self._interim_results = response["spec"].get('interim_results', {})
+            self._name = response['name']
+            self._id = response['id']
+            self._description = response.get('description', "")
+            self._max_execution_time = response.get('cost', 0)
+            self._creation_date = response.get('creation_date', "")
+            self._update_date = response.get('update_date', "")
+            self._is_public = response.get('is_public', False)
+            if 'data' in response:
+                self._data = response['data']
+            else:
+                raise IBMNotAuthorizedError(
+                    'Only program authors are authorized to retrieve program data')
+        return self._data
+
 
 class ParameterNamespace(SimpleNamespace):
     """ A namespace for program parameters with validation.

releasenotes/notes/program-data-retrieval-9a9782eb16274593.yaml

@@ -0,0 +1,6 @@
+---
+features:
+  - |
+    If you are the author of a runtime program, 
+    you can now use :attr:`qiskit_ibm.runtime.RuntimeProgram.data`
+    property to retrieve the program data as a string.

test/ibm/runtime/test_runtime.py

@@ -67,6 +67,7 @@
 class TestRuntime(IBMTestCase):
     """Class for testing runtime modules."""
 
+    DEFAULT_DATA = "def main() {}"
     DEFAULT_METADATA = {
         "name": "qiskit-test",
         "description": "Test program.",
@@ -427,10 +428,18 @@ def test_run_program_with_custom_runtime_image(self):
         self.assertTrue(job.result())
         self.assertEqual(job.image, image)
 
+    def test_retrieve_program_data(self):
+        """Test retrieving program data"""
+        program_id = self._upload_program(name="qiskit-test")
+        self.runtime.programs()
+        program = self.runtime.program(program_id)
+        self.assertEqual(program.data, self.DEFAULT_DATA)
+        self._validate_program(program)
+
     def test_program_params_validation(self):
         """Test program parameters validation process"""
         program_id = self.runtime.upload_program(
-            data="def main() {}", metadata=self.DEFAULT_METADATA)
+            data=self.DEFAULT_DATA, metadata=self.DEFAULT_METADATA)
         program = self.runtime.program(program_id)
         params: ParameterNamespace = program.parameters()
         params.param1 = 'Hello, World'
@@ -448,7 +457,7 @@ def test_program_params_validation(self):
     def test_program_params_namespace(self):
         """Test running a program using parameter namespace."""
         program_id = self.runtime.upload_program(
-            data="def main() {}", metadata=self.DEFAULT_METADATA)
+            data=self.DEFAULT_DATA, metadata=self.DEFAULT_METADATA)
         params = self.runtime.program(program_id).parameters()
         params.param1 = "Hello World"
         self._run_program(program_id, inputs=params)
@@ -686,23 +695,10 @@ def test_program_metadata(self):
 
         for metadata in sub_tests:
             with self.subTest(metadata_type=type(metadata)):
-                program_id = self.runtime.upload_program(data="def main() {}", metadata=metadata)
+                program_id = self.runtime.upload_program(data=self.DEFAULT_DATA, metadata=metadata)
                 program = self.runtime.program(program_id)
                 self.runtime.delete_program(program_id)
-                self.assertEqual(self.DEFAULT_METADATA['name'], program.name)
-                self.assertEqual(self.DEFAULT_METADATA['description'], program.description)
-                self.assertEqual(self.DEFAULT_METADATA['max_execution_time'],
-                                 program.max_execution_time)
-                self.assertTrue(program.creation_date)
-                self.assertTrue(program.update_date)
-                self.assertEqual(self.DEFAULT_METADATA['spec']['backend_requirements'],
-                                 program.backend_requirements)
-                self.assertEqual(self.DEFAULT_METADATA['spec']['parameters'],
-                                 program.parameters().metadata)
-                self.assertEqual(self.DEFAULT_METADATA['spec']['return_values'],
-                                 program.return_values)
-                self.assertEqual(self.DEFAULT_METADATA['spec']['interim_results'],
-                                 program.interim_results)
+                self._validate_program(program)
 
     def test_different_providers(self):
         """Test retrieving job submitted with different provider."""
@@ -718,7 +714,7 @@ def _upload_program(self, name=None, max_execution_time=300,
                         is_public: bool = False):
         """Upload a new program."""
         name = name or uuid.uuid4().hex
-        data = "def main() {}"
+        data = self.DEFAULT_DATA
         metadata = copy.deepcopy(self.DEFAULT_METADATA)
         metadata.update(name=name)
         metadata.update(is_public=is_public)
@@ -762,3 +758,20 @@ def _populate_jobs_with_all_statuses(self, jobs, program_id):
             jobs.append(self._run_program(program_id, final_status='CANCELLED'))
             returned_jobs_count += 1
         return (jobs, pending_jobs_count, returned_jobs_count)
+
+    def _validate_program(self, program):
+        """Validate a program."""
+        self.assertEqual(self.DEFAULT_METADATA['name'], program.name)
+        self.assertEqual(self.DEFAULT_METADATA['description'], program.description)
+        self.assertEqual(self.DEFAULT_METADATA['max_execution_time'],
+                         program.max_execution_time)
+        self.assertTrue(program.creation_date)
+        self.assertTrue(program.update_date)
+        self.assertEqual(self.DEFAULT_METADATA['spec']['backend_requirements'],
+                         program.backend_requirements)
+        self.assertEqual(self.DEFAULT_METADATA['spec']['parameters'],
+                         program.parameters().metadata)
+        self.assertEqual(self.DEFAULT_METADATA['spec']['return_values'],
+                         program.return_values)
+        self.assertEqual(self.DEFAULT_METADATA['spec']['interim_results'],
+                         program.interim_results)

test/ibm/runtime/test_runtime_integration.py

@@ -147,6 +147,19 @@ def test_list_program(self):
         self.assertEqual(self.program_id, program.program_id)
         self._validate_program(program)
 
+    def test_retrieve_program_data(self):
+        """Test retrieving program data"""
+        program = self.provider.runtime.program(self.program_id)
+        self.assertEqual(self.RUNTIME_PROGRAM, program.data)
+        self._validate_program(program)
+
+    def test_retrieve_unauthorized_program_data(self):
+        """Test retrieving program data when user is not the program author"""
+        program = self.provider.runtime.program('sample-program')
+        self._validate_program(program)
+        with self.assertRaises(IBMNotAuthorizedError):
+            return program.data
+
     def test_upload_program(self):
         """Test uploading a program."""
         max_execution_time = 3000