Repository: Qiskit/qiskit-ibm-runtime
Pull request #522 (2022-09-07 21:51:00)

Use updated base primitives
<!--
⚠️ The pull request title should be short detailed and understandable for all.
⚠️ Also, please add a release note file using reno if the change needs to be
  documented in the release notes.
⚠️ If your pull request fixes an open issue, please link to the issue.

✅ I have added the tests to cover my changes.
✅ I have updated the documentation accordingly.
✅ I have read the CONTRIBUTING document.
-->

### Summary

Update the base primitives to the (almost) latest version.

The main change was for `RuntimeJob` to inherit the `Job` class in qiskit-terra, which means
- `job_id` and `backend` need to be methods, not attributes. This was done by adding `__call__` to those attributes so both `job.job_id` and `job.job_id()` would return the same thing.
- rename `metadata()` to `metrics()`. Luckily the method was introduced after the last stable release, so it's not really a breaking change
- `wait_for_final_state()` has a different signature from the base class because QRT API doesn't support getting job status from ws streaming. Will open a separate issue on that.

### Details and comments
Related to #482. There are new changes coming to the base primitives in qiskit-terra, so I'll leave that issue open. 


review-notebook-app[bot]: Check out this pull request on&nbsp; <a href="https://app.reviewnb.com/Qiskit/qiskit-ibm-runtime/pull/522"><img align="absmiddle"  alt="ReviewNB" height="28" class="BotMessageButtonImage" src="https://raw.githubusercontent.com/ReviewNB/support/master/images/button_reviewnb.png"/></a> 

 See visual diffs & provide feedback on Jupyter Notebooks. 

---

 <i>Powered by <a href='https://www.reviewnb.com/?utm_source=gh'>ReviewNB</a></i>
coveralls: ## Pull Request Test Coverage Report for [Build 3011048376](https://coveralls.io/builds/52292442)

* **-64** of **177**   **(63.84%)**  changed or added relevant lines in **11** files are covered.
* **10** unchanged lines in **3** files lost coverage.
* Overall coverage decreased (**-2.4%**) to **61.878%**

---

|  Changes Missing Coverage | Covered Lines | Changed/Added Lines | % |
| :-----|--------------|--------|---: |
| [qiskit_ibm_runtime/utils/json.py](https://coveralls.io/builds/52292442/source?filename=qiskit_ibm_runtime%2Futils%2Fjson.py#L238) | 2 | 3 | 66.67%
| [qiskit_ibm_runtime/estimator.py](https://coveralls.io/builds/52292442/source?filename=qiskit_ibm_runtime%2Festimator.py#L219) | 9 | 11 | 81.82%
| [qiskit_ibm_runtime/runtime_session.py](https://coveralls.io/builds/52292442/source?filename=qiskit_ibm_runtime%2Fruntime_session.py#L80) | 0 | 2 | 0.0%
| [qiskit_ibm_runtime/sampler.py](https://coveralls.io/builds/52292442/source?filename=qiskit_ibm_runtime%2Fsampler.py#L180) | 7 | 9 | 77.78%
| [qiskit_ibm_runtime/runtime_job.py](https://coveralls.io/builds/52292442/source?filename=qiskit_ibm_runtime%2Fruntime_job.py#L327) | 18 | 22 | 81.82%
| [qiskit_ibm_runtime/qiskit/primitives/base_sampler.py](https://coveralls.io/builds/52292442/source?filename=qiskit_ibm_runtime%2Fqiskit%2Fprimitives%2Fbase_sampler.py#L132) | 26 | 47 | 55.32%
| [qiskit_ibm_runtime/qiskit/primitives/base_estimator.py](https://coveralls.io/builds/52292442/source?filename=qiskit_ibm_runtime%2Fqiskit%2Fprimitives%2Fbase_estimator.py#L153) | 36 | 68 | 52.94%
<!-- | **Total:** | **113** | **177** | **63.84%** | -->

|  Files with Coverage Reduction | New Missed Lines | % |
| :-----|--------------|--: |
| [qiskit_ibm_runtime/qiskit/primitives/base_sampler.py](https://coveralls.io/builds/52292442/source?filename=qiskit_ibm_runtime%2Fqiskit%2Fprimitives%2Fbase_sampler.py#L252) | 1 | 50.5% |
| [qiskit_ibm_runtime/qiskit/primitives/base_estimator.py](https://coveralls.io/builds/52292442/source?filename=qiskit_ibm_runtime%2Fqiskit%2Fprimitives%2Fbase_estimator.py#L178) | 3 | 45.65% |
| [qiskit_ibm_runtime/qiskit/primitives/utils.py](https://coveralls.io/builds/52292442/source?filename=qiskit_ibm_runtime%2Fqiskit%2Fprimitives%2Futils.py#L12) | 6 | 0% |
<!-- | **Total:** | **10** |  | -->

|  Totals | [![Coverage Status](https://coveralls.io/builds/52292442/badge)](https://coveralls.io/builds/52292442) |
| :-- | --: |
| Change from base [Build 3001239376](https://coveralls.io/builds/52246815): |  -2.4% |
| Covered Lines: | 3110 |
| Relevant Lines: | 5026 |

---
##### 💛  - [Coveralls](https://coveralls.io)


Patch:
diff --git a/README.md b/README.md
index c7e30261..d2a5f938 100644
--- a/README.md
+++ b/README.md
@@ -142,7 +142,7 @@ job = service.run(
     program_id="sampler",
     options=options,
     inputs=program_inputs)
-print(f"job ID: {job.job_id}")
+print(f"job ID: {job.job_id()}")
 result = job.result()
 ```
 
diff --git a/docs/getting_started.rst b/docs/getting_started.rst
index b542c352..3dfdf9de 100644
--- a/docs/getting_started.rst
+++ b/docs/getting_started.rst
@@ -59,7 +59,7 @@ Run the Hello World program to ensure that your environment is set up properly:
                     options=options,
                     inputs=program_inputs
                     )
-    print(f"job id: {job.job_id}")
+    print(f"job id: {job.job_id()}")
     result = job.result()
     print(result)
 
diff --git a/docs/how_to/run_a_job.ipynb b/docs/how_to/run_a_job.ipynb
index 45e00765..e6ec5a40 100644
--- a/docs/how_to/run_a_job.ipynb
+++ b/docs/how_to/run_a_job.ipynb
@@ -72,7 +72,7 @@
     ")\n",
     "\n",
     "# Printing the job ID in case we need to retrieve it later.\n",
-    "print(f\"job id: {job.job_id}\")\n",
+    "print(f\"job id: {job.job_id()}\")\n",
     "\n",
     "# Get the job result - this is blocking and control may not return immediately.\n",
     "result = job.result()\n",
diff --git a/docs/tutorials/01_introduction_ibm_cloud_runtime.ipynb b/docs/tutorials/01_introduction_ibm_cloud_runtime.ipynb
index f1225404..847bf8f3 100644
--- a/docs/tutorials/01_introduction_ibm_cloud_runtime.ipynb
+++ b/docs/tutorials/01_introduction_ibm_cloud_runtime.ipynb
@@ -399,13 +399,13 @@
     ")\n",
     "\n",
     "# Printing the job ID in case we need to retrieve it later.\n",
-    "print(f\"Job ID: {job.job_id}\")\n",
+    "print(f\"Job ID: {job.job_id()}\")\n",
     "\n",
     "# Get the job result - this is blocking and control may not return immediately.\n",
     "result = job.result()\n",
     "print(result)\n",
     "# see which backend the job was executed on\n",
-    "print(job.backend) "
+    "print(job.backend()) "
    ]
   },
   {
@@ -431,11 +431,10 @@
     "- `wait_for_final_state()`: Wait for the job to finish.\n",
     "- `logs()`: Return job logs.\n",
     "- `error_message()`: Returns the reason if the job failed and `None` otherwise.\n",
+    "- `job_id()`: Unique identifier of the job.\n",
+    "- `backend()`: The backend where the job is run.\n",
     "\n",
     "Some of the `RuntimeJob` attributes:\n",
-    "\n",
-    "- `job_id`: Unique identifier of the job.\n",
-    "- `backend`: The backend where the job is run.\n",
     "- `program_id`: ID of the program the execution is for.\n",
     "\n",
     "\n",
@@ -576,7 +575,7 @@
     ")\n",
     "\n",
     "# Printing the job ID in case we need to retrieve it later.\n",
-    "print(f\"Job ID: {job.job_id}\")\n",
+    "print(f\"Job ID: {job.job_id()}\")\n",
     "\n",
     "# Get the job result - this is blocking and control may not return immediately.\n",
     "result = job.result()\n",
@@ -615,12 +614,12 @@
     }
    ],
    "source": [
-    "retrieved_job = service.job(job.job_id)\n",
+    "retrieved_job = service.job(job.job_id())\n",
     "print(\n",
-    "    f\"Job {retrieved_job.job_id} is an execution instance of runtime program {retrieved_job.program_id}.\"\n",
+    "    f\"Job {retrieved_job.job_id()} is an execution instance of runtime program {retrieved_job.program_id}.\"\n",
     ")\n",
     "print(\n",
-    "    f\"This job ran on backend {retrieved_job.backend} and had input parameters {retrieved_job.inputs}\"\n",
+    "    f\"This job ran on backend {retrieved_job.backend()} and had input parameters {retrieved_job.inputs}\"\n",
     ")"
    ]
   },
@@ -649,7 +648,7 @@
    "source": [
     "retrieved_jobs = service.jobs(limit=1)\n",
     "for rjob in retrieved_jobs:\n",
-    "    print(rjob.job_id)"
+    "    print(rjob.job_id())"
    ]
   },
   {
@@ -675,7 +674,7 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "service.delete_job(job.job_id)"
+    "service.delete_job(job.job_id())"
    ]
   },
   {
diff --git a/docs/tutorials/02_introduction_ibm_quantum_runtime.ipynb b/docs/tutorials/02_introduction_ibm_quantum_runtime.ipynb
index 48784d3f..37cf46e4 100644
--- a/docs/tutorials/02_introduction_ibm_quantum_runtime.ipynb
+++ b/docs/tutorials/02_introduction_ibm_quantum_runtime.ipynb
@@ -354,7 +354,7 @@
     ")\n",
     "\n",
     "# Printing the job ID in case we need to retrieve it later.\n",
-    "print(f\"job id: {job.job_id}\")\n",
+    "print(f\"job id: {job.job_id()}\")\n",
     "\n",
     "# Get the job result - this is blocking and control may not return immediately.\n",
     "result = job.result()\n",
@@ -402,11 +402,11 @@
     "- `stream_results()`: Stream interim results and final result. This can be used to start streaming the results if a `callback` function was not passed to the `run()` method. This method can also be used to reconnect a lost websocket connection.\n",
     "- `logs()`: Return job logs.\n",
     "- `error_message()`: Returns the reason if the job failed and `None` otherwise.\n",
+    "- `job_id()`: Unique identifier of the job.\n",
+    "- `backend()`: The backend where the job is run.\n",
     "\n",
     "Some of the `RuntimeJob` attributes:\n",
     "\n",
-    "- `job_id`: Unique identifier of the job.\n",
-    "- `backend`: The backend where the job is run.\n",
     "- `program_id`: ID of the program the execution is for.\n",
     "\n",
     "\n",
@@ -455,12 +455,12 @@
     }
    ],
    "source": [
-    "retrieved_job = service.job(job.job_id)\n",
+    "retrieved_job = service.job(job.job_id())\n",
     "print(\n",
-    "    f\"Job {retrieved_job.job_id} is an execution instance of runtime program {retrieved_job.program_id}.\"\n",
+    "    f\"Job {retrieved_job.job_id()} is an execution instance of runtime program {retrieved_job.program_id}.\"\n",
     ")\n",
     "print(\n",
-    "    f\"This job ran on backend {retrieved_job.backend} and had input parameters {retrieved_job.inputs}\"\n",
+    "    f\"This job ran on backend {retrieved_job.backend()} and had input parameters {retrieved_job.inputs}\"\n",
     ")"
    ]
   },
@@ -489,7 +489,7 @@
    "source": [
     "retrieved_jobs = service.jobs(limit=1)\n",
     "for rjob in retrieved_jobs:\n",
-    "    print(rjob.job_id)"
+    "    print(rjob.job_id())"
    ]
   },
   {
@@ -515,7 +515,7 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "service.delete_job(job.job_id)"
+    "service.delete_job(job.job_id())"
    ]
   },
   {
diff --git a/docs/tutorials/qka.ipynb b/docs/tutorials/qka.ipynb
index b42d817b..d36f0489 100644
--- a/docs/tutorials/qka.ipynb
+++ b/docs/tutorials/qka.ipynb
@@ -493,7 +493,7 @@
     "    callback=interim_result_callback,\n",
     ")\n",
     "\n",
-    "print(job.job_id)\n",
+    "print(job.job_id())\n",
     "result = job.result()"
    ]
   },
diff --git a/docs/tutorials/vqe.ipynb b/docs/tutorials/vqe.ipynb
index 5fc85d57..2b1ae2d1 100644
--- a/docs/tutorials/vqe.ipynb
+++ b/docs/tutorials/vqe.ipynb
@@ -385,7 +385,7 @@
     "job = service.run(\n",
     "    program_id=\"vqe\", inputs=vqe_inputs, options=backend_options, callback=raw_callback\n",
     ")\n",
-    "print(\"Job ID:\", job.job_id)"
+    "print(\"Job ID:\", job.job_id())"
    ]
   },
   {
diff --git a/qiskit_ibm_runtime/estimator.py b/qiskit_ibm_runtime/estimator.py
index 5bdee2df..426d1857 100644
--- a/qiskit_ibm_runtime/estimator.py
+++ b/qiskit_ibm_runtime/estimator.py
@@ -18,6 +18,7 @@ from typing import Iterable, Optional, Dict, Sequence, Any, Union
 
 import numpy as np
 from qiskit.circuit import QuantumCircuit, Parameter
+from qiskit.circuit.parametertable import ParameterView
 from qiskit.quantum_info import SparsePauliOp
 from qiskit.opflow import PauliSumOp
 from qiskit.quantum_info.operators.base_operator import BaseOperator
@@ -132,13 +133,6 @@ class Estimator(BaseEstimator):
             parameters=parameters,
         )
 
-        # TODO: Remove deprecation warnings if done in base class
-        if circuits or parameters or observables:
-            deprecate_arguments(
-                "circuits, parameters, and observables",
-                "0.7",
-                f"You can instead specify these inputs using the {self.__class__.__name__}.run method.",
-            )
         if skip_transpilation:
             deprecate_arguments(
                 "skip_transpilation",
@@ -186,12 +180,64 @@ class Estimator(BaseEstimator):
 
     def run(
         self,
-        circuits: Union[QuantumCircuit, Sequence[QuantumCircuit]],
+        circuits: QuantumCircuit | Sequence[QuantumCircuit],
+        observables: BaseOperator | PauliSumOp | Sequence[BaseOperator | PauliSumOp],
+        parameter_values: Sequence[float] | Sequence[Sequence[float]] | None = None,
+        parameters: Sequence[Parameter] | Sequence[Sequence[Parameter]] | None = None,
+        **kwargs: Any,
+    ) -> RuntimeJob:
+        """Submit a request to the estimator primitive program.
+
+        Args:
+            circuits: a (parameterized) :class:`~qiskit.circuit.QuantumCircuit` or
+                a list of (parameterized) :class:`~qiskit.circuit.QuantumCircuit`.
+
+            observables: Observable objects.
+
+            parameter_values: Concrete parameters to be bound.
+
+            parameters: Parameters of quantum circuits, specifying the order in which values
+                will be bound. Defaults to ``[circ.parameters for circ in circuits]``
+
+            **kwargs: Individual options to overwrite the default primitive options.
+
+        Returns:
+            Submitted job.
+
+        Raises:
+            QiskitError: Invalid arguments are given.
+        """
+        if not isinstance(circuits, Sequence):
+            circuits = [circuits]
+        if not isinstance(observables, Sequence):
+            observables = [observables]
+        if (
+            parameter_values is not None
+            and len(parameter_values) > 1
+            and not isinstance(parameter_values[0], Sequence)
+        ):
+            parameter_values = [parameter_values]  # type: ignore[assignment]
+        if (
+            parameters is not None
+            and len(parameters) > 1
+            and not isinstance(parameters[0], Sequence)
+        ):
+            parameters = [parameters]
+
+        return super().run(
+            circuits=circuits,
+            observables=observables,
+            parameter_values=parameter_values,
+            parameters=parameters,
+            **kwargs,
+        )
+
+    def _run(
+        self,
+        circuits: Sequence[QuantumCircuit],
         observables: Sequence[BaseOperator | PauliSumOp],
-        parameter_values: Optional[
-            Union[Sequence[float], Sequence[Sequence[float]]]
-        ] = None,
-        parameters: Sequence[Sequence[Parameter]] | None = None,
+        parameter_values: Sequence[Sequence[float]],
+        parameters: list[ParameterView],
         **kwargs: Any,
     ) -> RuntimeJob:
         """Submit a request to the estimator primitive program.
@@ -213,29 +259,12 @@ class Estimator(BaseEstimator):
 
         Returns:
             Submitted job.
-
-        Raises:
-            ValueError: If the input values are invalid.
         """
-        if isinstance(circuits, Iterable) and not all(
-            isinstance(inst, QuantumCircuit) for inst in circuits
-        ):
-            raise ValueError(
-                "The circuits parameter has to be instances of QuantumCircuit."
-            )
-
-        circ_count = 1 if isinstance(circuits, QuantumCircuit) else len(circuits)
-        obs_count = (
-            1
-            if isinstance(observables, (BaseOperator, PauliSumOp))
-            else len(observables)
-        )
-
         inputs = {
             "circuits": circuits,
-            "circuit_indices": list(range(circ_count)),
+            "circuit_indices": list(range(len(circuits))),
             "observables": observables,
-            "observable_indices": list(range(obs_count)),
+            "observable_indices": list(range(len(observables))),
             "parameters": parameters,
             "parameter_values": parameter_values,
         }
diff --git a/qiskit_ibm_runtime/ibm_backend.py b/qiskit_ibm_runtime/ibm_backend.py
index 90fdfc75..df71c654 100644
--- a/qiskit_ibm_runtime/ibm_backend.py
+++ b/qiskit_ibm_runtime/ibm_backend.py
@@ -481,6 +481,10 @@ class IBMBackend(Backend):
     def __repr__(self) -> str:
         return "<{}('{}')>".format(self.__class__.__name__, self.name)
 
+    def __call__(self) -> "IBMBackend":
+        # For backward compatibility only, can be removed later.
+        return self
+
     def run(
         self,
         circuits: Union[
diff --git a/qiskit_ibm_runtime/options.py b/qiskit_ibm_runtime/options.py
index 0e4668de..856807ce 100644
--- a/qiskit_ibm_runtime/options.py
+++ b/qiskit_ibm_runtime/options.py
@@ -124,10 +124,6 @@ class Options:
             log levels are: ``DEBUG``, ``INFO``, ``WARNING``, ``ERROR``, and ``CRITICAL``.
             The default level is ``WARNING``.
 
-        log_level: logging level to set in the execution environment. The valid
-            log levels are: ``DEBUG``, ``INFO``, ``WARNING``, ``ERROR``, and ``CRITICAL``.
-            The default level is ``WARNING``.
-
         transpilation: Transpilation options.
 
             * skip_transpilation: Whether to skip transpilation.
diff --git a/qiskit_ibm_runtime/qiskit/primitives/base_estimator.py b/qiskit_ibm_runtime/qiskit/primitives/base_estimator.py
index 3e4f94ff..76cd9d2e 100644
--- a/qiskit_ibm_runtime/qiskit/primitives/base_estimator.py
+++ b/qiskit_ibm_runtime/qiskit/primitives/base_estimator.py
@@ -32,9 +32,9 @@ An estimator is initialized with the following elements.
 
 The estimator is called with the following inputs.
 
-* circuit indexes: a list of indexes of the quantum circuits.
+* circuits: a list of the quantum circuits.
 
-* observable indexes: a list of indexes of the observables.
+* observables: a list of the observables.
 
 * parameters: a list of parameters of the quantum circuits.
   (:class:`~qiskit.circuit.parametertable.ParameterView` or
@@ -74,57 +74,50 @@ Here is an example of how estimator is used.
     H2 = SparsePauliOp.from_list([("IZ", 1)])
     H3 = SparsePauliOp.from_list([("ZI", 1), ("ZZ", 1)])
 
-    with Estimator([psi1, psi2], [H1, H2, H3], [params1, params2]) as e:
-        theta1 = [0, 1, 1, 2, 3, 5]
-        theta2 = [0, 1, 1, 2, 3, 5, 8, 13]
-        theta3 = [1, 2, 3, 4, 5, 6]
-
-        # calculate [ <psi1(theta1)|H1|psi1(theta1)> ]
-        result = e([0], [0], [theta1])
-        print(result)
-
-        # calculate [ <psi1(theta1)|H2|psi1(theta1)>, <psi1(theta1)|H3|psi1(theta1)> ]
-        result2 = e([0, 0], [1, 2], [theta1]*2)
-        print(result2)
-
-        # calculate [ <psi2(theta2)|H2|psi2(theta2)> ]
-        result3 = e([1], [1], [theta2])
-        print(result3)
-
-        # calculate [ <psi1(theta1)|H1|psi1(theta1)>, <psi1(theta3)|H1|psi1(theta3)> ]
-        result4 = e([0, 0], [0, 0], [theta1, theta3])
-        print(result4)
-
-        # calculate [ <psi1(theta1)|H1|psi1(theta1)>,
-        #             <psi2(theta2)|H2|psi2(theta2)>,
-        #             <psi1(theta3)|H3|psi1(theta3)> ]
-        result5 = e([0, 1, 0], [0, 1, 2], [theta1, theta2, theta3])
-        print(result5)
-
-        # Objects can be passed instead of indices.
-        # calculate [ <psi2(theta2)|H2|psi2(theta2)> ]
-        # Note that passing objects has an overhead
-        # since the corresponding indices need to be searched.
-        result6 = e([psi2], [H2], [theta2])
-        print(result6)
+    theta1 = [0, 1, 1, 2, 3, 5]
+    theta2 = [0, 1, 1, 2, 3, 5, 8, 13]
+    theta3 = [1, 2, 3, 4, 5, 6]
+
+    estimator = Estimator()
+
+    # calculate [ <psi1(theta1)|H1|psi1(theta1)> ]
+    result = estimator.run([psi1], [H1], [theta1]).result()
+    print(result)
+
+    # calculate [ <psi2(theta2)|H2|psi2(theta2)> ]
+    result2 = estimator.run([psi2], [H1], [theta2]).result()
+    print(result2)
+
+    # calculate [ <psi1(theta1)|H2|psi1(theta1)>, <psi1(theta1)|H3|psi1(theta1)> ]
+    result3 = estimator.run([psi1, psi1], [H2, H3], [theta1]*2).result()
+    print(result3)
+
+    # calculate [ <psi1(theta1)|H1|psi1(theta1)>,
+    #             <psi2(theta2)|H2|psi2(theta2)>,
+    #             <psi1(theta3)|H3|psi1(theta3)> ]
+    result4 = estimator.run([psi1, psi2, psi1], [H1, H2, H3], [theta1, theta2, theta3]).result()
+    print(result4)
 """
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
 from collections.abc import Iterable, Sequence
 from copy import copy
-from typing import Any
+from typing import cast, Any
+from warnings import warn
 
 import numpy as np
 
 from qiskit.circuit import Parameter, QuantumCircuit
 from qiskit.circuit.parametertable import ParameterView
 from qiskit.exceptions import QiskitError
+from qiskit.opflow import PauliSumOp
+from qiskit.providers import JobV1 as Job
 from qiskit.quantum_info.operators import SparsePauliOp
-from qiskit.utils.deprecation import deprecate_arguments
+from qiskit.quantum_info.operators.base_operator import BaseOperator
+from qiskit.utils.deprecation import deprecate_arguments, deprecate_function
 
 from .estimator_result import EstimatorResult
-from .utils import _finditer
 
 
 class BaseEstimator(ABC):
@@ -133,10 +126,12 @@ class BaseEstimator(ABC):
     Base class for Estimator that estimates expectation values of quantum circuits and observables.
     """
 
+    __hash__ = None  # type: ignore
+
     def __init__(
         self,
-        circuits: Iterable[QuantumCircuit] | QuantumCircuit | None,
-        observables: Iterable[SparsePauliOp] | SparsePauliOp | None,
+        circuits: Iterable[QuantumCircuit] | QuantumCircuit | None = None,
+        observables: Iterable[SparsePauliOp] | SparsePauliOp | None = None,
         parameters: Iterable[Iterable[Parameter]] | None = None,
     ):
         """
@@ -154,32 +149,35 @@ class BaseEstimator(ABC):
         Raises:
             QiskitError: For mismatch of circuits and parameters list.
         """
-        if circuits is None and observables is None:
-            self._circuits: tuple = tuple()
-            self._observables: tuple = tuple()
-            return
-
+        if circuits is not None or observables is not None or parameters is not None:
+            warn(
+                "The BaseEstimator 'circuits', `observables`, `parameters` kwarg are deprecated "
+                "as of 0.22.0 and will be removed no earlier than 3 months after the "
+                "release date. You can use 'run' method to append objects.",
+                DeprecationWarning,
+                2,
+            )
         if isinstance(circuits, QuantumCircuit):
             circuits = (circuits,)
-        self._circuits = tuple(circuits)
+        self._circuits = [] if circuits is None else list(circuits)
 
         if isinstance(observables, SparsePauliOp):
             observables = (observables,)
-        self._observables = tuple(observables)
+        self._observables = [] if observables is None else list(observables)
 
         # To guarantee that they exist as instance variable.
         # With only dynamic set, the python will not know if the attribute exists or not.
-        self._circuit_ids = self._circuit_ids  # type: ignore
-        self._observable_ids = self._observable_ids  # type: ignore
+        self._circuit_ids: dict[int, int] = self._circuit_ids
+        self._observable_ids: dict[int, int] = self._observable_ids
 
         if parameters is None:
-            self._parameters = tuple(circ.parameters for circ in self._circuits)
+            self._parameters = [circ.parameters for circ in self._circuits]
         else:
-            self._parameters = tuple(ParameterView(par) for par in parameters)
-            if len(self._parameters) != len(self._circuits):  # type: ignore
+            self._parameters = [ParameterView(par) for par in parameters]
+            if len(self._parameters) != len(self._circuits):
                 raise QiskitError(
                     f"Different number of parameters ({len(self._parameters)}) and "
-                    f"circuits ({len(self._circuits)})"  # type: ignore
+                    f"circuits ({len(self._circuits)})"
                 )
             for i, (circ, params) in enumerate(zip(self._circuits, self._parameters)):
                 if circ.num_parameters != len(params):
@@ -190,7 +188,6 @@ class BaseEstimator(ABC):
 
     def __new__(  # pylint: disable=unused-argument
         cls,
-        *args: Any,
         circuits: Iterable[QuantumCircuit] | QuantumCircuit | None = None,
         observables: Iterable[SparsePauliOp] | SparsePauliOp | None = None,
         parameters: Iterable[Iterable[Parameter]] | None = None,
@@ -198,25 +195,40 @@ class BaseEstimator(ABC):
     ) -> BaseEstimator:
 
         self = super().__new__(cls)
-        if isinstance(circuits, Iterable):
+        if circuits is None:
+            self._circuit_ids = {}
+        elif isinstance(circuits, Iterable):
             circuits = copy(circuits)
-            self._circuit_ids = [id(circuit) for circuit in circuits]
+            self._circuit_ids = {id(circuit): i for i, circuit in enumerate(circuits)}
         else:
-            self._circuit_ids = [id(circuits)]
-        if isinstance(observables, Iterable):
+            self._circuit_ids = {id(circuits): 0}
+        if observables is None:
+            self._observable_ids = {}
+        elif isinstance(observables, Iterable):
             observables = copy(observables)
-            self._observable_ids = [id(observable) for observable in observables]
+            self._observable_ids = {
+                id(observable): i for i, observable in enumerate(observables)
+            }
         else:
-            self._observable_ids = [id(observables)]
+            self._observable_ids = {id(observables): 0}
         return self
 
+    @deprecate_function(
+        "The BaseEstimator.__enter__ method is deprecated as of Qiskit Terra 0.22.0 "
+        "and will be removed no sooner than 3 months after the releasedate. "
+        "BaseEstimator should be initialized directly.",
+    )
     def __enter__(self) -> BaseEstimator:
         return self
 
+    @deprecate_function(
+        "The BaseEstimator.__exit__ method is deprecated as of Qiskit Terra 0.22.0 "
+        "and will be removed no sooner than 3 months after the releasedate. "
+        "BaseEstimator should be initialized directly.",
+    )
     def __exit__(self, *exc_info: Any) -> None:
         self.close()
 
-    @abstractmethod
     def close(self) -> None:
         """Close the session and free resources"""
         ...
@@ -228,7 +240,7 @@ class BaseEstimator(ABC):
         Returns:
             The quantum circuits.
         """
-        return self._circuits  # type: ignore
+        return tuple(self._circuits)
 
     @property
     def observables(self) -> tuple[SparsePauliOp, ...]:
@@ -237,7 +249,7 @@ class BaseEstimator(ABC):
         Returns:
             The observables.
         """
-        return self._observables  # type: ignore
+        return tuple(self._observables)
 
     @property
     def parameters(self) -> tuple[ParameterView, ...]:
@@ -246,8 +258,13 @@ class BaseEstimator(ABC):
         Returns:
             Parameters, where ``parameters[i][j]`` is the j-th parameter of the i-th circuit.
         """
-        return self._parameters
+        return tuple(self._parameters)
 
+    @deprecate_function(
+        "The BaseSampler.__call__ method is deprecated as of Qiskit Terra 0.22.0 "
+        "and will be removed no sooner than 3 months after the releasedate. "
+        "Use run method instead.",
+    )
     @deprecate_arguments(
         {"circuit_indices": "circuits", "observable_indices": "observables"}
     )
@@ -298,35 +315,36 @@ class BaseEstimator(ABC):
             parameter_values = parameter_values.tolist()
 
         # Allow objects
-        try:
-            circuits = [
-                next(_finditer(id(circuit), self._circuit_ids))  # type: ignore
-                if not isinstance(circuit, (int, np.integer))
-                else circuit
-                for circuit in circuits
-            ]
-        except StopIteration as err:
+        circuits = [
+            self._circuit_ids.get(id(circuit))  # type: ignore
+            if not isinstance(circuit, (int, np.integer))
+            else circuit
+            for circuit in circuits
+        ]
+        if any(circuit is None for circuit in circuits):
             raise QiskitError(
                 "The circuits passed when calling estimator is not one of the circuits used to "
                 "initialize the session."
-            ) from err
-        try:
-            observables = [
-                next(_finditer(id(observable), self._observable_ids))
-                if not isinstance(observable, (int, np.integer))
-                else observable
-                for observable in observables
-            ]
-        except StopIteration as err:
+            )
+        observables = [
+            self._observable_ids.get(id(observable))  # type: ignore
+            if not isinstance(observable, (int, np.integer))
+            else observable
+            for observable in observables
+        ]
+        if any(observable is None for observable in observables):
             raise QiskitError(
                 "The observables passed when calling estimator is not one of the observables used to "
                 "initialize the session."
-            ) from err
+            )
+
+        circuits = cast("list[int]", circuits)
+        observables = cast("list[int]", observables)
 
         # Allow optional
         if parameter_values is None:
             for i in circuits:
-                if len(self._circuits[i].parameters) != 0:  # type: ignore
+                if len(self._circuits[i].parameters) != 0:
                     raise QiskitError(
                         f"The {i}-th circuit is parameterised,"
                         "but parameter values are not given."
@@ -380,6 +398,112 @@ class BaseEstimator(ABC):
             **run_options,
         )
 
+    def run(
+        self,
+        circuits: Sequence[QuantumCircuit],
+        observables: Sequence[BaseOperator | PauliSumOp],
+        parameter_values: Sequence[Sequence[float]] | None = None,
+        parameters: Sequence[Sequence[Parameter]] | None = None,
+        **run_options: Any,
+    ) -> Job:
+        """Run the job of the estimation of expectation value(s).
+
+        ``circuits``, ``observables``, and ``parameter_values`` should have the same
+        length. The i-th element of the result is the expectation of observable
+
+        .. code-block:: python
+
+            obs = observables[i]
+
+        for the state prepared by
+
+        .. code-block:: python
+
+            circ = circuits[i]
+
+        with bound parameters
+
+        .. code-block:: python
+
+            values = parameter_values[i].
+
+        Args:
+            circuits: the list of circuit objects.
+            observables: the list of observable objects.
+            parameter_values: concrete parameters to be bound.
+            parameters: Parameters of quantum circuits, specifying the order in which values
+                will be bound. Defaults to ``[circ.parameters for circ in circuits]``
+                The indexing is such that ``parameters[i, j]`` is the j-th formal parameter of
+                ``circuits[i]``.
+            **run_options: runtime options used for circuit execution.
+
+        Returns:
+            The job object of EstimatorResult.
+
+        Raises:
+            QiskitError: Invalid arguments are given.
+        """
+        # Support ndarray
+        if isinstance(parameter_values, np.ndarray):
+            parameter_values = parameter_values.tolist()
+
+        # Allow optional
+        if parameter_values is None:
+            for i, circuit in enumerate(circuits):
+                if circuit.num_parameters != 0:
+                    raise QiskitError(
+                        f"The {i}-th circuit is parameterised,"
+                        "but parameter values are not given."
+                    )
+            parameter_values = [[]] * len(circuits)
+
+        if parameters is None:
+            parameter_views = [circ.parameters for circ in circuits]
+        else:
+            parameter_views = [ParameterView(par) for par in parameters]
+            if len(self._parameters) != len(self._circuits):
+                raise QiskitError(
+                    f"Different number of parameters ({len(self._parameters)}) and "
+                    f"circuits ({len(self._circuits)})"
+                )
+            for i, (circ, params) in enumerate(zip(self._circuits, self._parameters)):
+                if circ.num_parameters != len(params):
+                    raise QiskitError(
+                        f"Different numbers of parameters of {i}-th circuit: "
+                        f"expected {circ.num_parameters}, actual {len(params)}."
+                    )
+
+        # Validation
+        if len(circuits) != len(observables):
+            raise QiskitError(
+                f"The number of circuits ({len(circuits)}) does not match "
+                f"the number of observables ({len(observables)})."
+            )
+        if len(circuits) != len(parameter_values):
+            raise QiskitError(
+                f"The number of circuits ({len(circuits)}) does not match "
+                f"the number of parameter value sets ({len(parameter_values)})."
+            )
+
+        for i, (circuit, parameter_value) in enumerate(zip(circuits, parameter_values)):
+            if len(parameter_value) != circuit.num_parameters:
+                raise QiskitError(
+                    f"The number of values ({len(parameter_value)}) does not match "
+                    f"the number of parameters ({circuit.num_parameters}) for the {i}-th circuit."
+                )
+
+        for i, (circuit, observable) in enumerate(zip(circuits, observables)):
+            if circuit.num_qubits != observable.num_qubits:
+                raise QiskitError(
+                    f"The number of qubits of the {i}-th circuit ({circuit.num_qubits}) does "
+                    f"not match the number of qubits of the {i}-th observable "
+                    f"({observable.num_qubits})."
+                )
+
+        return self._run(
+            circuits, observables, parameter_values, parameter_views, **run_options
+        )
+
     @abstractmethod
     def _call(
         self,
@@ -389,3 +513,17 @@ class BaseEstimator(ABC):
         **run_options: Any,
     ) -> EstimatorResult:
         ...
+
+    # This will be comment out after 0.22. (This is necessary for the compatibility.)
+    # @abstractmethod
+    def _run(
+        self,
+        circuits: Sequence[QuantumCircuit],
+        observables: Sequence[BaseOperator | PauliSumOp],
+        parameter_values: Sequence[Sequence[float]],
+        parameters: list[ParameterView],
+        **run_options: Any,
+    ) -> Job:
+        raise NotImplementedError(
+            "_run method is not implemented. This method will be @abstractmethod after 0.22."
+        )
diff --git a/qiskit_ibm_runtime/qiskit/primitives/base_sampler.py b/qiskit_ibm_runtime/qiskit/primitives/base_sampler.py
index 47b03a76..bb09e240 100644
--- a/qiskit_ibm_runtime/qiskit/primitives/base_sampler.py
+++ b/qiskit_ibm_runtime/qiskit/primitives/base_sampler.py
@@ -27,7 +27,7 @@ A sampler is initialized with the following elements.
 
 The sampler is run with the following inputs.
 
-* circuit indexes: a list of indices of the circuits to evaluate.
+* circuits: a list of QuantumCircuit objects to evaluate.
 
 * parameter values (:math:`\theta_k`): list of sets of parameter values
   to be bound to the parameters of the quantum circuits.
@@ -55,22 +55,15 @@ Here is an example of how sampler is used.
     bell.measure_all()
 
     # executes a Bell circuit
-    with Sampler(circuits=[bell], parameters=[[]]) as sampler:
-        result = sampler(parameters=[[]], circuits=[0])
-        print([q.binary_probabilities() for q in result.quasi_dists])
+    sampler = Sampler()
+    result = sampler.run(circuits=[bell]).result()
+    print([q.binary_probabilities() for q in result.quasi_dists])
 
     # executes three Bell circuits
-    with Sampler([bell]*3, [[]] * 3) as sampler:
-        result = sampler([0, 1, 2], [[]]*3)
-        print([q.binary_probabilities() for q in result.quasi_dists])
-
-    # executes three Bell circuits with objects.
-    # Objects can be passed instead of indices.
-    # Note that passing objects has an overhead
-    # since the corresponding indices need to be searched.
-    with Sampler([bell]) as sampler:
-        result = sampler([bell, bell, bell])
-        print([q.binary_probabilities() for q in result.quasi_dists])
+    # Argument `parameters` is optional.
+    sampler = Sampler()
+    result = sampler.run([bell, bell, bell]).result()
+    print([q.binary_probabilities() for q in result.quasi_dists])
 
     # parameterized circuit
     pqc = RealAmplitudes(num_qubits=2, reps=2)
@@ -82,35 +75,35 @@ Here is an example of how sampler is used.
     theta2 = [1, 2, 3, 4, 5, 6]
     theta3 = [0, 1, 2, 3, 4, 5, 6, 7]
 
-    with Sampler(circuits=[pqc, pqc2], parameters=[pqc.parameters, pqc2.parameters]) as sampler:
-        result = sampler([0, 0, 1], [theta1, theta2, theta3])
+    sampler = Sampler()
+    result = sampler.run([pqc, pqc, pqc2], [theta1, theta2, theta3]).result()
 
-        # result of pqc(theta1)
-        print(result.quasi_dists[0].binary_probabilities())
+    # result of pqc(theta1)
+    print(result.quasi_dists[0].binary_probabilities())
 
-        # result of pqc(theta2)
-        print(result.quasi_dists[1].binary_probabilities())
-
-        # result of pqc2(theta3)
-        print(result.quasi_dists[2].binary_probabilities())
+    # result of pqc(theta2)
+    print(result.quasi_dists[1].binary_probabilities())
 
+    # result of pqc2(theta3)
+    print(result.quasi_dists[2].binary_probabilities())
 """
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
 from collections.abc import Iterable, Sequence
 from copy import copy
-from typing import Any
+from typing import cast, Any
+from warnings import warn
 
 import numpy as np
 
 from qiskit.circuit import Parameter, QuantumCircuit
 from qiskit.circuit.parametertable import ParameterView
 from qiskit.exceptions import QiskitError
-from qiskit.utils.deprecation import deprecate_arguments
+from qiskit.providers import JobV1 as Job
+from qiskit.utils.deprecation import deprecate_arguments, deprecate_function
 
 from .sampler_result import SamplerResult
-from .utils import _finditer
 
 
 class BaseSampler(ABC):
@@ -119,11 +112,13 @@ class BaseSampler(ABC):
     Base class of Sampler that calculates quasi-probabilities of bitstrings from quantum circuits.
     """
 
+    __hash__ = None  # type: ignore
+
     def __init__(
         self,
-        circuits: Iterable[QuantumCircuit] | QuantumCircuit | None,
+        circuits: Iterable[QuantumCircuit] | QuantumCircuit | None = None,
         parameters: Iterable[Iterable[Parameter]] | None = None,
-    ):
+    ) -> None:
         """
         Args:
             circuits: Quantum circuits to be executed.
@@ -133,22 +128,26 @@ class BaseSampler(ABC):
         Raises:
             QiskitError: For mismatch of circuits and parameters list.
         """
-        if circuits is None:
-            self._circuits: tuple = tuple()
-            return
-
+        if circuits is not None or parameters is not None:
+            warn(
+                "The BaseSampler 'circuits', and `parameters` kwarg are deprecated "
+                "as of 0.22.0 and will be removed no earlier than 3 months after the "
+                "release date. You can use 'run' method to append objects.",
+                DeprecationWarning,
+                2,
+            )
         if isinstance(circuits, QuantumCircuit):
             circuits = (circuits,)
-        self._circuits = tuple(circuits)
+        self._circuits = [] if circuits is None else list(circuits)
 
         # To guarantee that they exist as instance variable.
         # With only dynamic set, the python will not know if the attribute exists or not.
-        self._circuit_ids = self._circuit_ids  # type: ignore
+        self._circuit_ids: dict[int, int] = self._circuit_ids
 
         if parameters is None:
-            self._parameters = tuple(circ.parameters for circ in self._circuits)
+            self._parameters = [circ.parameters for circ in self._circuits]
         else:
-            self._parameters = tuple(ParameterView(par) for par in parameters)
+            self._parameters = [ParameterView(par) for par in parameters]
             if len(self._parameters) != len(self._circuits):
                 raise QiskitError(
                     f"Different number of parameters ({len(self._parameters)}) "
@@ -157,7 +156,6 @@ class BaseSampler(ABC):
 
     def __new__(  # pylint: disable=unused-argument
         cls,
-        *args: Any,
         circuits: Iterable[QuantumCircuit] | QuantumCircuit | None = None,
         parameters: Iterable[Iterable[Parameter]] | None = None,
         **kwargs: Any,
@@ -165,23 +163,30 @@ class BaseSampler(ABC):
 
         self = super().__new__(cls)
         if circuits is None:
-            self._circuit_ids = []
-            return self
-
-        if isinstance(circuits, Iterable):
+            self._circuit_ids = {}
+        elif isinstance(circuits, Iterable):
             circuits = copy(circuits)
-            self._circuit_ids = [id(circuit) for circuit in circuits]
+            self._circuit_ids = {id(circuit): i for i, circuit in enumerate(circuits)}
         else:
-            self._circuit_ids = [id(circuits)]
+            self._circuit_ids = {id(circuits): 0}
         return self
 
+    @deprecate_function(
+        "The BaseSampler.__enter__ method is deprecated as of Qiskit Terra 0.22.0 "
+        "and will be removed no sooner than 3 months after the releasedate. "
+        "BaseSampler should be initialized directly.",
+    )
     def __enter__(self) -> BaseSampler:
         return self
 
+    @deprecate_function(
+        "The BaseSampler.__exit__ method is deprecated as of Qiskit Terra 0.22.0 "
+        "and will be removed no sooner than 3 months after the releasedate. "
+        "BaseSampler should be initialized directly.",
+    )
     def __exit__(self, *exc_info: Any) -> None:
         self.close()
 
-    @abstractmethod
     def close(self) -> None:
         """Close the session and free resources"""
         ...
@@ -193,7 +198,7 @@ class BaseSampler(ABC):
         Returns:
             The quantum circuits to be sampled.
         """
-        return self._circuits
+        return tuple(self._circuits)
 
     @property
     def parameters(self) -> tuple[ParameterView, ...]:
@@ -202,8 +207,13 @@ class BaseSampler(ABC):
         Returns:
             List of the parameters in each quantum circuit.
         """
-        return self._parameters
+        return tuple(self._parameters)
 
+    @deprecate_function(
+        "The BaseSampler.__call__ method is deprecated as of Qiskit Terra 0.22.0 "
+        "and will be removed no sooner than 3 months after the releasedate. "
+        "Use run method instead.",
+    )
     @deprecate_arguments({"circuit_indices": "circuits"})
     def __call__(
         self,
@@ -232,25 +242,26 @@ class BaseSampler(ABC):
             parameter_values = parameter_values.tolist()
 
         # Allow objects
-        try:
-            circuits = [
-                next(_finditer(id(circuit), self._circuit_ids))  # type: ignore
-                if not isinstance(circuit, (int, np.integer))
-                else circuit
-                for circuit in circuits
-            ]
-        except StopIteration as err:
+        circuits = [
+            self._circuit_ids.get(id(circuit))  # type: ignore
+            if not isinstance(circuit, (int, np.integer))
+            else circuit
+            for circuit in circuits
+        ]
+        if any(circuit is None for circuit in circuits):
             raise QiskitError(
                 "The circuits passed when calling sampler is not one of the circuits used to "
                 "initialize the session."
-            ) from err
+            )
+
+        circuits = cast("list[int]", circuits)
 
         # Allow optional
         if parameter_values is None:
             for i in circuits:
                 if len(self._circuits[i].parameters) != 0:
                     raise QiskitError(
-                        f"The {i}-th circuit is parameterised,"
+                        f"The {i}-th circuit ({len(circuits)}) is parameterised,"
                         "but parameter values are not given."
                     )
             parameter_values = [[]] * len(circuits)
@@ -281,6 +292,75 @@ class BaseSampler(ABC):
             **run_options,
         )
 
+    def run(
+        self,
+        circuits: Sequence[QuantumCircuit],
+        parameter_values: Sequence[Sequence[float]] | None = None,
+        parameters: Sequence[Sequence[Parameter]] | None = None,
+        **run_options: Any,
+    ) -> Job:
+        """Run the job of the sampling of bitstrings.
+
+        Args:
+            circuits: the list of circuit objects.
+            parameter_values: Parameters to be bound to the circuit.
+            parameters: Parameters of each of the quantum circuits.
+                Defaults to ``[circ.parameters for circ in circuits]``.
+            **run_options: Backend runtime options used for circuit execution.
+
+        Returns:
+            The job object of the result of the sampler. The i-th result corresponds to
+            ``circuits[i]`` evaluated with parameters bound as ``parameter_values[i]``.
+
+        Raises:
+            QiskitError: Invalid arguments are given.
+        """
+        # Support ndarray
+        if isinstance(parameter_values, np.ndarray):
+            parameter_values = parameter_values.tolist()
+
+        # Allow optional
+        if parameter_values is None:
+            for i, circuit in enumerate(circuits):
+                if circuit.num_parameters != 0:
+                    raise QiskitError(
+                        f"The {i}-th circuit ({len(circuits)}) is parameterised,"
+                        "but parameter values are not given."
+                    )
+            parameter_values = [[]] * len(circuits)
+
+        if parameters is None:
+            parameter_views = [circ.parameters for circ in circuits]
+        else:
+            parameter_views = [ParameterView(par) for par in parameters]
+            if len(self._parameters) != len(self._circuits):
+                raise QiskitError(
+                    f"Different number of parameters ({len(self._parameters)}) and "
+                    f"circuits ({len(self._circuits)})"
+                )
+            for i, (circ, params) in enumerate(zip(self._circuits, self._parameters)):
+                if circ.num_parameters != len(params):
+                    raise QiskitError(
+                        f"Different numbers of parameters of {i}-th circuit: "
+                        f"expected {circ.num_parameters}, actual {len(params)}."
+                    )
+
+        # Validation
+        if len(circuits) != len(parameter_values):
+            raise QiskitError(
+                f"The number of circuits ({len(circuits)}) does not match "
+                f"the number of parameter value sets ({len(parameter_values)})."
+            )
+
+        for i, (circuit, parameter_value) in enumerate(zip(circuits, parameter_values)):
+            if len(parameter_value) != circuit.num_parameters:
+                raise QiskitError(
+                    f"The number of values ({len(parameter_value)}) does not match "
+                    f"the number of parameters ({circuit.num_parameters}) for the {i}-th circuit."
+                )
+
+        return self._run(circuits, parameter_values, parameter_views, **run_options)
+
     @abstractmethod
     def _call(
         self,
@@ -289,3 +369,16 @@ class BaseSampler(ABC):
         **run_options: Any,
     ) -> SamplerResult:
         ...
+
+    # This will be comment out after 0.22. (This is necessary for the compatibility.)
+    # @abstractmethod
+    def _run(
+        self,
+        circuits: Sequence[QuantumCircuit],
+        parameter_values: Sequence[Sequence[float]],
+        parameters: Sequence[ParameterView],
+        **run_options: Any,
+    ) -> Job:
+        raise NotImplementedError(
+            "_run method is not implemented. This method will be @abstractmethod after 0.22."
+        )
diff --git a/qiskit_ibm_runtime/qiskit_runtime_service.py b/qiskit_ibm_runtime/qiskit_runtime_service.py
index e2ee8e2c..89f3752d 100644
--- a/qiskit_ibm_runtime/qiskit_runtime_service.py
+++ b/qiskit_ibm_runtime/qiskit_runtime_service.py
@@ -51,7 +51,7 @@ from .utils.hgp import to_instance_format, from_instance_format
 from .utils.utils import validate_job_tags, validate_runtime_options
 from .api.client_parameters import ClientParameters
 from .runtime_options import RuntimeOptions
-from .utils.deprecation import deprecate_function
+from .utils.deprecation import deprecate_function, issue_deprecation_msg
 
 logger = logging.getLogger(__name__)
 
@@ -64,6 +64,8 @@ DEPRECATED_PROGRAMS = [
     "quantum_kernal_alignment",
 ]
 
+_JOB_DEPRECATION_ISSUED = False
+
 
 class QiskitRuntimeService(Provider):
     """Class for interacting with the Qiskit Runtime service.
@@ -951,6 +953,18 @@ class QiskitRuntimeService(Provider):
         if not backend:
             backend = self.backend(name=response["backend"])
 
+        global _JOB_DEPRECATION_ISSUED  # pylint: disable=global-statement
+        if not (start_session or session_id) and not _JOB_DEPRECATION_ISSUED:
+            # No need to issue warning if session is used since the old style session
+            # didn't return a job, and new style returns new job.
+            _JOB_DEPRECATION_ISSUED = True
+            issue_deprecation_msg(
+                msg="Note that the 'job_id' and 'backend' attributes of "
+                "a runtime job have been deprecated",
+                version="0.7",
+                remedy="Please use the job_id() and backend() methods instead.",
+            )
+
         job = RuntimeJob(
             backend=backend,
             api_client=self._api_client,
diff --git a/qiskit_ibm_runtime/runtime_job.py b/qiskit_ibm_runtime/runtime_job.py
index 7db7c112..e73a1727 100644
--- a/qiskit_ibm_runtime/runtime_job.py
+++ b/qiskit_ibm_runtime/runtime_job.py
@@ -23,6 +23,7 @@ from datetime import datetime
 
 from qiskit.providers.backend import Backend
 from qiskit.providers.jobstatus import JobStatus, JOB_FINAL_STATES
+from qiskit.providers.job import JobV1 as Job
 
 from .constants import API_TO_JOB_ERROR_MESSAGE, API_TO_JOB_STATUS
 from .exceptions import (
@@ -38,11 +39,12 @@ from .exceptions import IBMError
 from .api.exceptions import RequestsApiError
 from .utils.converters import utc_to_local
 from .api.client_parameters import ClientParameters
+from .utils.utils import CallableStr
 
 logger = logging.getLogger(__name__)
 
 
-class RuntimeJob:
+class RuntimeJob(Job):
     """Representation of a runtime program execution.
 
     A new ``RuntimeJob`` instance is returned when you call
@@ -107,8 +109,7 @@ class RuntimeJob:
             result_decoder: A :class:`ResultDecoder` subclass used to decode job results.
             image: Runtime image used for this job: image_name:tag.
         """
-        self._job_id = job_id
-        self._backend = backend
+        super().__init__(backend=backend, job_id=job_id)
         self._api_client = api_client
         self._results: Optional[Any] = None
         self._interim_results: Optional[Any] = None
@@ -137,6 +138,11 @@ class RuntimeJob:
         if user_callback is not None:
             self.stream_results(user_callback)
 
+        # For backward compatibility. These can be removed once 'job_id' and 'backend'
+        # as attributes are no longer supported.
+        self.job_id = CallableStr(job_id)
+        self.backend = self._backend
+
     def interim_results(self, decoder: Optional[Type[ResultDecoder]] = None) -> Any:
         """Return the interim results of the job.
 
@@ -152,14 +158,14 @@ class RuntimeJob:
         if not self._final_interim_results:
             _decoder = decoder or self._result_decoder
             interim_results_raw = self._api_client.job_interim_results(
-                job_id=self.job_id
+                job_id=self.job_id()
             )
             self._interim_results = _decoder.decode(interim_results_raw)
             if self.status() in JOB_FINAL_STATES:
                 self._final_interim_results = True
         return self._interim_results
 
-    def result(
+    def result(  # pylint: disable=arguments-differ
         self,
         timeout: Optional[float] = None,
         decoder: Optional[Type[ResultDecoder]] = None,
@@ -187,7 +193,7 @@ class RuntimeJob:
                 raise RuntimeJobFailureError(
                     f"Unable to retrieve job result. " f"{error_message}"
                 )
-            result_raw = self._api_client.job_results(job_id=self.job_id)
+            result_raw = self._api_client.job_results(job_id=self.job_id())
             self._results = _decoder.decode(result_raw) if result_raw else None
         return self._results
 
@@ -199,7 +205,7 @@ class RuntimeJob:
             IBMRuntimeError: If unable to cancel job.
         """
         try:
-            self._api_client.job_cancel(self.job_id)
+            self._api_client.job_cancel(self.job_id())
         except RequestsApiError as ex:
             if ex.status_code == 409:
                 raise RuntimeInvalidStateError(
@@ -227,7 +233,10 @@ class RuntimeJob:
         self._set_status_and_error_message()
         return self._error_message
 
-    def wait_for_final_state(self, timeout: Optional[float] = None) -> None:
+    def wait_for_final_state(  # pylint: disable=arguments-differ
+        self,
+        timeout: Optional[float] = None,
+    ) -> None:
         """Use the websocket server to wait for the final the state of a job. The server
             will remain open if the job is still running and the connection will be terminated
             once the job completes. Then update and return the status of the job.
@@ -315,31 +324,45 @@ class RuntimeJob:
         if self.status() not in JOB_FINAL_STATES:
             logger.warning("Job logs are only available after the job finishes.")
         try:
-            return self._api_client.job_logs(self.job_id)
+            return self._api_client.job_logs(self.job_id())
         except RequestsApiError as err:
             if err.status_code == 404:
                 return ""
             raise IBMRuntimeError(f"Failed to get job logs: {err}") from None
 
-    def metadata(self) -> Dict[str, Any]:
-        """Return job metadata.
+    def metrics(self) -> Dict[str, Any]:
+        """Return job metrics.
 
         Returns:
-            Job metadata, which includes timestamp information.
+            Job metrics, which includes timestamp information.
 
         Raises:
             IBMRuntimeError: If a network error occurred.
         """
         try:
-            metadata_str = self._api_client.job_metadata(self.job_id)
+            metadata_str = self._api_client.job_metadata(self.job_id())
             return json.loads(metadata_str)
         except RequestsApiError as err:
             raise IBMRuntimeError(f"Failed to get job metadata: {err}") from None
 
+    def submit(self) -> None:
+        """Unsupported method.
+        Note:
+            This method is not supported, please use
+            :meth:`~qiskit_ibm_runtime.QiskitRuntimeService.run`
+            to submit a job.
+        Raises:
+            NotImplementedError: Upon invocation.
+        """
+        raise NotImplementedError(
+            "job.submit() is not supported. Please use "
+            "QiskitRuntimeService.run() to submit a job."
+        )
+
     def _set_status_and_error_message(self) -> None:
         """Fetch and set status and error message."""
         if self._status not in JOB_FINAL_STATES:
-            response = self._api_client.job_get(job_id=self.job_id)
+            response = self._api_client.job_get(job_id=self.job_id())
             self._set_status(response)
             self._set_error_message(response)
 
@@ -381,11 +404,11 @@ class RuntimeJob:
             Error message.
         """
         status = response["state"]["status"].upper()
-        job_result_raw = self._api_client.job_results(job_id=self.job_id)
+        job_result_raw = self._api_client.job_results(job_id=self.job_id())
         error_msg = API_TO_JOB_ERROR_MESSAGE["FAILED"]
         if status == "CANCELLED" and self._reason == "RAN TOO LONG":
             error_msg = API_TO_JOB_ERROR_MESSAGE["CANCELLED - RAN TOO LONG"]
-        return error_msg.format(self.job_id, job_result_raw)
+        return error_msg.format(self.job_id(), job_result_raw)
 
     def _status_from_job_response(self, response: Dict) -> str:
         """Returns the job status from an API response.
@@ -418,13 +441,13 @@ class RuntimeJob:
     def _start_websocket_client(self) -> None:
         """Start websocket client to stream results."""
         try:
-            logger.debug("Start websocket client for job %s", self.job_id)
+            logger.debug("Start websocket client for job %s", self.job_id())
             self._ws_client.job_results()
         except Exception:  # pylint: disable=broad-except
             logger.warning(
                 "An error occurred while streaming results "
                 "from the server for job %s:\n%s",
-                self.job_id,
+                self.job_id(),
                 traceback.format_exc(),
             )
         finally:
@@ -443,7 +466,7 @@ class RuntimeJob:
             user_callback: User callback function.
             decoder: A :class:`ResultDecoder` (sub)class used to decode job results.
         """
-        logger.debug("Start result streaming for job %s", self.job_id)
+        logger.debug("Start result streaming for job %s", self.job_id())
         _decoder = decoder or self._result_decoder
         while True:
             try:
@@ -451,11 +474,11 @@ class RuntimeJob:
                 if response == self._POISON_PILL:
                     self._empty_result_queue(result_queue)
                     return
-                user_callback(self.job_id, _decoder.decode(response))
+                user_callback(self.job_id(), _decoder.decode(response))
             except Exception:  # pylint: disable=broad-except
                 logger.warning(
                     "An error occurred while streaming results " "for job %s:\n%s",
-                    self.job_id,
+                    self.job_id(),
                     traceback.format_exc(),
                 )
 
@@ -474,24 +497,6 @@ class RuntimeJob:
     def __repr__(self) -> str:
         return f"<{self.__class__.__name__}('{self._job_id}', '{self._program_id}')>"
 
-    @property
-    def job_id(self) -> str:
-        """Return a unique ID identifying the job.
-
-        Returns:
-            Job ID.
-        """
-        return self._job_id
-
-    @property
-    def backend(self) -> Backend:
-        """Return the backend where this job was executed.
-
-        Returns:
-            Backend used for the job.
-        """
-        return self._backend
-
     @property
     def image(self) -> str:
         """Return the runtime image used for the job.
@@ -529,7 +534,7 @@ class RuntimeJob:
             ``None`` if creation date is not available.
         """
         if not self._creation_date:
-            response = self._api_client.job_get(job_id=self.job_id)
+            response = self._api_client.job_get(job_id=self.job_id())
             self._creation_date = response.get("created", None)
 
         if not self._creation_date:
diff --git a/qiskit_ibm_runtime/runtime_session.py b/qiskit_ibm_runtime/runtime_session.py
index b85ef31f..62368a34 100644
--- a/qiskit_ibm_runtime/runtime_session.py
+++ b/qiskit_ibm_runtime/runtime_session.py
@@ -77,7 +77,7 @@ class RuntimeSession:
             self._start_session = True
             self._initial_job = self._run(inputs=inputs)
             self._job = self._initial_job
-            self._session_id = self._job.job_id
+            self._session_id = self._job.job_id()
         else:
             self._start_session = False
             self._max_time = None
@@ -118,7 +118,7 @@ class RuntimeSession:
         if self._options:
             out["backend"] = self._options["backend_name"] or "unknown"  # type: ignore
         if self._job:
-            out["job_id"] = self._job.job_id
+            out["job_id"] = self._job.job_id()
             out["job_status"] = self._job.status()
             out["backend"] = self._job.backend
         return out
diff --git a/qiskit_ibm_runtime/sampler.py b/qiskit_ibm_runtime/sampler.py
index 5d180770..46c51866 100644
--- a/qiskit_ibm_runtime/sampler.py
+++ b/qiskit_ibm_runtime/sampler.py
@@ -18,6 +18,7 @@ from typing import Dict, Iterable, Optional, Sequence, Any, Union
 import copy
 
 from qiskit.circuit import QuantumCircuit, Parameter
+from qiskit.circuit.parametertable import ParameterView
 from qiskit.result import QuasiDistribution
 
 # TODO import BaseSampler and SamplerResult from terra once released
@@ -59,7 +60,7 @@ class Sampler(BaseSampler):
             sampler.options.resilience_level = 1
 
             job = sampler.run(bell)
-            print(f"Job ID: {job.job_id}")
+            print(f"Job ID: {job.job_id()}")
             print(f"Job result: {job.result()}")
     """
 
@@ -101,19 +102,11 @@ class Sampler(BaseSampler):
             skip_transpilation (DEPRECATED): Transpilation is skipped if set to True. False by default.
                 Ignored ``skip_transpilation`` is also specified in ``options``.
         """
-        # TODO: Fix base classes once done
         super().__init__(
             circuits=circuits,
             parameters=parameters,
         )
 
-        # TODO: Remove deprecation warnings if done in base class
-        if circuits or parameters:
-            deprecate_arguments(
-                "circuits and parameters",
-                "0.7",
-                f"You can instead specify these inputs using the {self.__class__.__name__}.run method.",
-            )
         if skip_transpilation:
             deprecate_arguments(
                 "skip_transpilation",
@@ -156,9 +149,54 @@ class Sampler(BaseSampler):
 
     def run(
         self,
-        circuits: Union[QuantumCircuit, Sequence[QuantumCircuit]],
-        parameter_values: Sequence[Sequence[float]] | None = None,
-        parameters: Sequence[Sequence[Parameter]] | None = None,
+        circuits: QuantumCircuit | Sequence[QuantumCircuit],
+        parameter_values: Sequence[float] | Sequence[Sequence[float]] | None = None,
+        parameters: Sequence[Parameter] | Sequence[Sequence[Parameter]] | None = None,
+        **kwargs: Any,
+    ) -> RuntimeJob:
+        """Submit a request to the sampler primitive program.
+
+        Args:
+            circuits: A (parameterized) :class:`~qiskit.circuit.QuantumCircuit` or
+                a list of (parameterized) :class:`~qiskit.circuit.QuantumCircuit`.
+            parameter_values: Concrete parameters to be bound.
+            parameters: Parameters of each of the quantum circuits.
+                Defaults to ``[circ.parameters for circ in circuits]``.
+            **kwargs: Individual options to overwrite the default primitive options.
+
+        Returns:
+            Submitted job.
+
+        Raises:
+            QiskitError: Invalid arguments are given.
+        """
+        if not isinstance(circuits, Sequence):
+            circuits = [circuits]
+        if (
+            parameter_values is not None
+            and len(parameter_values) > 1
+            and not isinstance(parameter_values[0], Sequence)
+        ):
+            parameter_values = [parameter_values]  # type: ignore[assignment]
+        if (
+            parameters is not None
+            and len(parameters) > 1
+            and not isinstance(parameters[0], Sequence)
+        ):
+            parameters = [parameters]
+
+        return super().run(
+            circuits=circuits,
+            parameter_values=parameter_values,
+            parameters=parameters,
+            **kwargs,
+        )
+
+    def _run(
+        self,
+        circuits: Sequence[QuantumCircuit],
+        parameter_values: Sequence[Sequence[float]],
+        parameters: Sequence[ParameterView],
         **kwargs: Any,
     ) -> RuntimeJob:
         """Submit a request to the sampler primitive program.
@@ -178,23 +216,11 @@ class Sampler(BaseSampler):
 
         Returns:
             Submitted job.
-
-        Raises:
-            ValueError: If the input values are invalid.
         """
-        if isinstance(circuits, Iterable) and not all(
-            isinstance(inst, QuantumCircuit) for inst in circuits
-        ):
-            raise ValueError(
-                "The circuits parameter has to be instances of QuantumCircuit."
-            )
-
-        circ_count = 1 if isinstance(circuits, QuantumCircuit) else len(circuits)
-
         inputs = {
             "circuits": circuits,
             "parameters": parameters,
-            "circuit_indices": list(range(circ_count)),
+            "circuit_indices": list(range(len(circuits))),
             "parameter_values": parameter_values,
         }
         combined = self.options._merge_options(kwargs)
diff --git a/qiskit_ibm_runtime/session.py b/qiskit_ibm_runtime/session.py
index b622334f..9c562146 100644
--- a/qiskit_ibm_runtime/session.py
+++ b/qiskit_ibm_runtime/session.py
@@ -53,7 +53,7 @@ class Session:
         with Session(backend="ibmq_qasm_simulator") as session:
             sampler = Sampler(session=session, options=options)
             job = sampler.run(circ)
-            print(f"Sampler job ID: {job.job_id}")
+            print(f"Sampler job ID: {job.job_id()}")
             print(f"Sampler job result:" {job.result()})
     """
 
@@ -142,10 +142,10 @@ class Session:
         )
 
         if self._session_id is None:
-            self._session_id = job.job_id
+            self._session_id = job.job_id()
 
         if self._backend is None:
-            self._backend = job.backend.name
+            self._backend = job.backend().name
 
         return job
 
@@ -155,6 +155,14 @@ class Session:
         if self._session_id:
             self._service._api_client.close_session(self._session_id)
 
+    def backend(self) -> Optional[str]:
+        """Return backend for this session.
+
+        Returns:
+            Backend for this session. None if unknown.
+        """
+        return self._backend
+
     @property
     def session_id(self) -> str:
         """Return the session ID.
@@ -173,15 +181,6 @@ class Session:
         """
         return self._service
 
-    @property
-    def backend(self) -> Optional[str]:
-        """Return backend for this session.
-
-        Returns:
-            Backend for this session. None if unknown.
-        """
-        return self._backend
-
     def __enter__(self) -> "Session":
         return self
 
diff --git a/qiskit_ibm_runtime/utils/json.py b/qiskit_ibm_runtime/utils/json.py
index 4d4bedb0..f79b3f89 100644
--- a/qiskit_ibm_runtime/utils/json.py
+++ b/qiskit_ibm_runtime/utils/json.py
@@ -46,6 +46,7 @@ from qiskit.circuit import (
     QuantumRegister,
 )
 from qiskit.circuit.library import BlueprintCircuit
+from qiskit.circuit.parametertable import ParameterView
 from qiskit.result import Result
 from qiskit.version import __version__ as _terra_version_string
 
@@ -233,6 +234,8 @@ class RuntimeEncoder(json.JSONEncoder):
                 compress=False,
             )
             return {"__type__": "ParameterExpression", "__value__": value}
+        if isinstance(obj, ParameterView):
+            return obj.data
         if isinstance(obj, Instruction):
             # Append instruction to empty circuit
             quantum_register = QuantumRegister(obj.num_qubits)
diff --git a/qiskit_ibm_runtime/utils/utils.py b/qiskit_ibm_runtime/utils/utils.py
index df67350f..fff45933 100644
--- a/qiskit_ibm_runtime/utils/utils.py
+++ b/qiskit_ibm_runtime/utils/utils.py
@@ -349,3 +349,10 @@ class RefreshQueue(Queue):
         """Wake up all threads waiting for items on the queued."""
         with self.condition:
             self.condition.notifyAll()
+
+
+class CallableStr(str):
+    """A callable string."""
+
+    def __call__(self) -> str:
+        return self
diff --git a/releasenotes/notes/0.7/job-metrics-endpoint-82d61cf269c27933.yaml b/releasenotes/notes/0.7/job-metrics-endpoint-82d61cf269c27933.yaml
index 49acbab3..8dbe4464 100644
--- a/releasenotes/notes/0.7/job-metrics-endpoint-82d61cf269c27933.yaml
+++ b/releasenotes/notes/0.7/job-metrics-endpoint-82d61cf269c27933.yaml
@@ -1,6 +1,6 @@
 ---
 upgrade:
   - |
-    :class:`qiskit_ibm_runtime.RuntimeJob` has a new method 
-    :meth:`~qiskit_ibm_runtime.RuntimeJob.metadata`. This method returns 
-    the metadata of a job, which includes timestamp information.
+    :class:`qiskit_ibm_runtime.RuntimeJob` has a new method
+    :meth:`~qiskit_ibm_runtime.RuntimeJob.metrics`. This method returns
+    the metrics of a job, which includes timestamp information.
diff --git a/releasenotes/notes/job_id_backend_attr-20dffa6335e508f4.yaml b/releasenotes/notes/job_id_backend_attr-20dffa6335e508f4.yaml
new file mode 100644
index 00000000..9ec61a14
--- /dev/null
+++ b/releasenotes/notes/job_id_backend_attr-20dffa6335e508f4.yaml
@@ -0,0 +1,10 @@
+---
+deprecations:
+  - |
+    ``job_id`` and ``backend`` attributes of :class:`qiskit_ibm_runtime.RuntimeJob`
+    have been deprecated. Please use :meth:`qiskit_ibm_runtime.RuntimeJob.job_id`
+    and :meth:`qiskit_ibm_runtime.RuntimeJob.backend` instead.
+upgrade:
+  - |
+    The :class:`~qiskit_ibm_runtime.RuntimeJob` class is now a subclass of
+    :class:`qiskit.providers.Job`.
\ No newline at end of file


Test patch:
diff --git a/test/ibm_test_case.py b/test/ibm_test_case.py
index 0a299bef..a0526e8c 100644
--- a/test/ibm_test_case.py
+++ b/test/ibm_test_case.py
@@ -102,7 +102,7 @@ class IBMIntegrationTestCase(IBMTestCase):
             with suppress(Exception):
                 job.cancel()
             with suppress(Exception):
-                service.delete_job(job.job_id)
+                service.delete_job(job.job_id())
 
     def _upload_program(
         self,
@@ -214,6 +214,6 @@ class IBMIntegrationJobTestCase(IBMIntegrationTestCase):
             callback=callback,
             start_session=start_session,
         )
-        self.log.info("Runtime job %s submitted.", job.job_id)
+        self.log.info("Runtime job %s submitted.", job.job_id())
         self.to_cancel[service.channel].append(job)
         return job
diff --git a/test/integration/test_job.py b/test/integration/test_job.py
index 68e40169..a2d8ed09 100644
--- a/test/integration/test_job.py
+++ b/test/integration/test_job.py
@@ -65,7 +65,7 @@ class TestIntegrationJob(IBMIntegrationJobTestCase):
             self.skipTest("Not supported on ibm_quantum")
 
         job = self._run_program(service, backend="")
-        self.assertTrue(job.backend, f"Job {job.job_id} has no backend.")
+        self.assertTrue(job.backend(), f"Job {job.job_id()} has no backend.")
 
     @run_integration_test
     def test_run_program_log_level(self, service):
@@ -87,10 +87,10 @@ class TestIntegrationJob(IBMIntegrationJobTestCase):
         """Test a failed program execution."""
         job = self._run_program(service, inputs={})
         job.wait_for_final_state()
-        job_result_raw = service._api_client.job_results(job.job_id)
+        job_result_raw = service._api_client.job_results(job.job_id())
         self.assertEqual(JobStatus.ERROR, job.status())
         self.assertIn(
-            API_TO_JOB_ERROR_MESSAGE["FAILED"].format(job.job_id, job_result_raw),
+            API_TO_JOB_ERROR_MESSAGE["FAILED"].format(job.job_id(), job_result_raw),
             job.error_message(),
         )
         with self.assertRaises(RuntimeJobFailureError) as err_cm:
@@ -108,11 +108,11 @@ class TestIntegrationJob(IBMIntegrationJobTestCase):
         job = self._run_program(service, program_id=program_id, inputs=inputs)
 
         job.wait_for_final_state()
-        job_result_raw = service._api_client.job_results(job.job_id)
+        job_result_raw = service._api_client.job_results(job.job_id())
         self.assertEqual(JobStatus.ERROR, job.status())
         self.assertIn(
             API_TO_JOB_ERROR_MESSAGE["CANCELLED - RAN TOO LONG"].format(
-                job.job_id, job_result_raw
+                job.job_id(), job_result_raw
             ),
             job.error_message(),
         )
@@ -132,7 +132,7 @@ class TestIntegrationJob(IBMIntegrationJobTestCase):
         )
         job.wait_for_final_state()
         self.assertEqual(
-            job._api_client.job_get(job.job_id)["cost"], job_max_execution_time
+            job._api_client.job_get(job.job_id())["cost"], job_max_execution_time
         )
 
     @run_integration_test
@@ -151,7 +151,7 @@ class TestIntegrationJob(IBMIntegrationJobTestCase):
         if not cancel_job_safe(job, self.log):
             return
         time.sleep(10)  # Wait a bit for DB to update.
-        rjob = service.job(job.job_id)
+        rjob = service.job(job.job_id())
         self.assertEqual(rjob.status(), JobStatus.CANCELLED)
 
     @run_integration_test
@@ -162,7 +162,7 @@ class TestIntegrationJob(IBMIntegrationJobTestCase):
         if not cancel_job_safe(job, self.log):
             return
         time.sleep(10)  # Wait a bit for DB to update.
-        rjob = service.job(job.job_id)
+        rjob = service.job(job.job_id())
         self.assertEqual(rjob.status(), JobStatus.CANCELLED)
 
     @run_integration_test
@@ -181,9 +181,9 @@ class TestIntegrationJob(IBMIntegrationJobTestCase):
             with self.subTest(status=status):
                 job = self._run_program(service, iterations=2)
                 wait_for_status(job, status)
-                service.delete_job(job.job_id)
+                service.delete_job(job.job_id())
                 with self.assertRaises(RuntimeJobNotFound):
-                    service.job(job.job_id)
+                    service.job(job.job_id())
 
     @run_integration_test
     def test_delete_job_queued(self, service):
@@ -192,9 +192,9 @@ class TestIntegrationJob(IBMIntegrationJobTestCase):
         _ = self._run_program(service, iterations=10, backend=real_device)
         job = self._run_program(service, iterations=2, backend=real_device)
         wait_for_status(job, JobStatus.QUEUED)
-        service.delete_job(job.job_id)
+        service.delete_job(job.job_id())
         with self.assertRaises(RuntimeJobNotFound):
-            service.job(job.job_id)
+            service.job(job.job_id())
 
     @run_integration_test
     def test_final_result(self, service):
@@ -204,7 +204,7 @@ class TestIntegrationJob(IBMIntegrationJobTestCase):
         result = job.result(decoder=SerializableClassDecoder)
         self.assertEqual(final_result, result)
 
-        rresults = service.job(job.job_id).result(decoder=SerializableClassDecoder)
+        rresults = service.job(job.job_id()).result(decoder=SerializableClassDecoder)
         self.assertEqual(final_result, rresults)
 
     @run_integration_test
@@ -221,7 +221,7 @@ class TestIntegrationJob(IBMIntegrationJobTestCase):
         inputs = {"iterations": 1, "interim_results": interim_results}
         job = self._run_program(service, inputs=inputs)
         self.assertEqual(inputs, job.inputs)
-        rjob = service.job(job.job_id)
+        rjob = service.job(job.job_id())
         rinterim_results = rjob.inputs["interim_results"]
         self._assert_complex_types_equal(interim_results, rinterim_results)
 
@@ -229,7 +229,7 @@ class TestIntegrationJob(IBMIntegrationJobTestCase):
     def test_job_backend(self, service):
         """Test job backend."""
         job = self._run_program(service)
-        self.assertEqual(self.sim_backends[service.channel], job.backend.name)
+        self.assertEqual(self.sim_backends[service.channel], job.backend().name)
 
     @run_integration_test
     def test_job_program_id(self, service):
@@ -259,7 +259,7 @@ class TestIntegrationJob(IBMIntegrationJobTestCase):
         """Test job creation date."""
         job = self._run_program(service, iterations=1)
         self.assertTrue(job.creation_date)
-        rjob = service.job(job.job_id)
+        rjob = service.job(job.job_id())
         self.assertTrue(rjob.creation_date)
         rjobs = service.jobs(limit=2)
         for rjob in rjobs:
@@ -277,14 +277,14 @@ class TestIntegrationJob(IBMIntegrationJobTestCase):
         self.assertIn("this is a stderr message", job_logs)
 
     @run_integration_test
-    def test_job_metadata(self, service):
-        """Test job metadata."""
+    def test_job_metrics(self, service):
+        """Test job metrics."""
         job = self._run_program(service)
         job.wait_for_final_state()
-        metadata = job.metadata()
-        self.assertTrue(metadata)
-        self.assertIn("timestamps", metadata)
-        self.assertIn("qiskit_version", metadata)
+        metrics = job.metrics()
+        self.assertTrue(metrics)
+        self.assertIn("timestamps", metrics)
+        self.assertIn("qiskit_version", metrics)
 
     def _assert_complex_types_equal(self, expected, received):
         """Verify the received data in complex types is expected."""
diff --git a/test/integration/test_results.py b/test/integration/test_results.py
index cde107e1..a4c49a37 100644
--- a/test/integration/test_results.py
+++ b/test/integration/test_results.py
@@ -36,7 +36,7 @@ class TestIntegrationResults(IBMIntegrationJobTestCase):
             if "iteration" in result:
                 final_it = result["iteration"]
             nonlocal callback_err
-            if job_id != job.job_id:
+            if job_id != job.job_id():
                 callback_err.append(f"Unexpected job ID: {job_id}")
             if "interim_results" in result and result["interim_results"] != int_res:
                 callback_err.append(f"Unexpected interim result: {result}")
@@ -67,7 +67,7 @@ class TestIntegrationResults(IBMIntegrationJobTestCase):
             if "iteration" in result:
                 final_it = result["iteration"]
             nonlocal callback_err
-            if job_id != job.job_id:
+            if job_id != job.job_id():
                 callback_err.append(f"Unexpected job ID: {job_id}")
             if "interim_results" in result and result["interim_results"] != int_res:
                 callback_err.append(f"Unexpected interim result: {result}")
@@ -98,7 +98,7 @@ class TestIntegrationResults(IBMIntegrationJobTestCase):
             if "iteration" in result:
                 final_it = result["iteration"]
             nonlocal callback_err
-            if job_id != job.job_id:
+            if job_id != job.job_id():
                 callback_err.append(f"Unexpected job ID: {job_id}")
             if "interim_results" in result and result["interim_results"] != int_res:
                 callback_err.append(f"Unexpected interim result: {result}")
diff --git a/test/integration/test_retrieve_job.py b/test/integration/test_retrieve_job.py
index c656159e..322a658d 100644
--- a/test/integration/test_retrieve_job.py
+++ b/test/integration/test_retrieve_job.py
@@ -31,8 +31,8 @@ class TestIntegrationRetrieveJob(IBMIntegrationJobTestCase):
         _ = self._run_program(service, iterations=10, backend=real_device)
         job = self._run_program(service, iterations=2, backend=real_device)
         wait_for_status(job, JobStatus.QUEUED)
-        rjob = service.job(job.job_id)
-        self.assertEqual(job.job_id, rjob.job_id)
+        rjob = service.job(job.job_id())
+        self.assertEqual(job.job_id(), rjob.job_id())
         self.assertEqual(self.program_ids[service.channel], rjob.program_id)
 
     @run_integration_test
@@ -40,8 +40,8 @@ class TestIntegrationRetrieveJob(IBMIntegrationJobTestCase):
         """Test retrieving a running job."""
         job = self._run_program(service, iterations=10)
         wait_for_status(job, JobStatus.RUNNING)
-        rjob = service.job(job.job_id)
-        self.assertEqual(job.job_id, rjob.job_id)
+        rjob = service.job(job.job_id())
+        self.assertEqual(job.job_id(), rjob.job_id())
         self.assertEqual(self.program_ids[service.channel], rjob.program_id)
 
     @run_integration_test
@@ -49,8 +49,8 @@ class TestIntegrationRetrieveJob(IBMIntegrationJobTestCase):
         """Test retrieving a finished job."""
         job = self._run_program(service)
         job.wait_for_final_state()
-        rjob = service.job(job.job_id)
-        self.assertEqual(job.job_id, rjob.job_id)
+        rjob = service.job(job.job_id())
+        self.assertEqual(job.job_id(), rjob.job_id())
         self.assertEqual(self.program_ids[service.channel], rjob.program_id)
 
     @run_integration_test
@@ -60,12 +60,12 @@ class TestIntegrationRetrieveJob(IBMIntegrationJobTestCase):
         rjobs = service.jobs()
         found = False
         for rjob in rjobs:
-            if rjob.job_id == job.job_id:
+            if rjob.job_id() == job.job_id():
                 self.assertEqual(job.program_id, rjob.program_id)
                 self.assertEqual(job.inputs, rjob.inputs)
                 found = True
                 break
-        self.assertTrue(found, f"Job {job.job_id} not returned.")
+        self.assertTrue(found, f"Job {job.job_id()} not returned.")
 
     @run_integration_test
     def test_retrieve_jobs_limit(self, service):
@@ -75,9 +75,11 @@ class TestIntegrationRetrieveJob(IBMIntegrationJobTestCase):
             jobs.append(self._run_program(service))
 
         rjobs = service.jobs(limit=2, program_id=self.program_ids[service.channel])
-        self.assertEqual(len(rjobs), 2, f"Retrieved jobs: {[j.job_id for j in rjobs]}")
-        job_ids = {job.job_id for job in jobs}
-        rjob_ids = {rjob.job_id for rjob in rjobs}
+        self.assertEqual(
+            len(rjobs), 2, f"Retrieved jobs: {[j.job_id() for j in rjobs]}"
+        )
+        job_ids = {job.job_id() for job in jobs}
+        rjob_ids = {rjob.job_id() for rjob in rjobs}
         self.assertTrue(
             rjob_ids.issubset(job_ids), f"Submitted: {job_ids}, Retrieved: {rjob_ids}"
         )
@@ -91,7 +93,7 @@ class TestIntegrationRetrieveJob(IBMIntegrationJobTestCase):
         after_status = job.status()
         found = False
         for rjob in rjobs:
-            if rjob.job_id == job.job_id:
+            if rjob.job_id() == job.job_id():
                 self.assertEqual(job.program_id, rjob.program_id)
                 self.assertEqual(job.inputs, rjob.inputs)
                 found = True
@@ -99,7 +101,7 @@ class TestIntegrationRetrieveJob(IBMIntegrationJobTestCase):
 
         self.assertTrue(
             found or after_status == JobStatus.RUNNING,
-            f"Pending job {job.job_id} not retrieved.",
+            f"Pending job {job.job_id()} not retrieved.",
         )
 
     @run_integration_test
@@ -110,12 +112,12 @@ class TestIntegrationRetrieveJob(IBMIntegrationJobTestCase):
         rjobs = service.jobs(pending=False)
         found = False
         for rjob in rjobs:
-            if rjob.job_id == job.job_id:
+            if rjob.job_id() == job.job_id():
                 self.assertEqual(job.program_id, rjob.program_id)
                 self.assertEqual(job.inputs, rjob.inputs)
                 found = True
                 break
-        self.assertTrue(found, f"Returned job {job.job_id} not retrieved.")
+        self.assertTrue(found, f"Returned job {job.job_id()} not retrieved.")
 
     @run_integration_test
     def test_retrieve_jobs_by_program_id(self, service):
@@ -125,7 +127,9 @@ class TestIntegrationRetrieveJob(IBMIntegrationJobTestCase):
         job.wait_for_final_state()
         rjobs = service.jobs(program_id=program_id)
         self.assertEqual(program_id, rjobs[0].program_id)
-        self.assertEqual(1, len(rjobs), f"Retrieved jobs: {[j.job_id for j in rjobs]}")
+        self.assertEqual(
+            1, len(rjobs), f"Retrieved jobs: {[j.job_id() for j in rjobs]}"
+        )
 
     @run_integration_test
     def test_retrieve_jobs_by_job_tags(self, service):
@@ -134,7 +138,9 @@ class TestIntegrationRetrieveJob(IBMIntegrationJobTestCase):
         job = self._run_program(service, job_tags=job_tags)
         job.wait_for_final_state()
         rjobs = service.jobs(job_tags=job_tags)
-        self.assertEqual(1, len(rjobs), f"Retrieved jobs: {[j.job_id for j in rjobs]}")
+        self.assertEqual(
+            1, len(rjobs), f"Retrieved jobs: {[j.job_id() for j in rjobs]}"
+        )
         rjobs = service.jobs(job_tags=["no_test_tag"])
         self.assertFalse(rjobs)
 
@@ -143,10 +149,12 @@ class TestIntegrationRetrieveJob(IBMIntegrationJobTestCase):
         """Test retrieving jobs by session_id."""
         job = self._run_program(service, start_session=True)
         job.wait_for_final_state()
-        job_2 = self._run_program(service, session_id=job.job_id)
+        job_2 = self._run_program(service, session_id=job.job_id())
         job_2.wait_for_final_state()
-        rjobs = service.jobs(session_id=job.job_id)
-        self.assertEqual(2, len(rjobs), f"Retrieved jobs: {[j.job_id for j in rjobs]}")
+        rjobs = service.jobs(session_id=job.job_id())
+        self.assertEqual(
+            2, len(rjobs), f"Retrieved jobs: {[j.job_id() for j in rjobs]}"
+        )
         rjobs = service.jobs(session_id="test")
         self.assertFalse(rjobs)
 
@@ -158,7 +166,7 @@ class TestIntegrationRetrieveJob(IBMIntegrationJobTestCase):
         job.wait_for_final_state()
         time_after_job = datetime.now(timezone.utc)
         rjobs = service.jobs(created_before=time_after_job, created_after=current_time)
-        self.assertTrue(job.job_id in [j.job_id for j in rjobs])
+        self.assertTrue(job.job_id() in [j.job_id() for j in rjobs])
         for job in rjobs:
             self.assertTrue(job.creation_date <= time_after_job)
             self.assertTrue(job.creation_date >= current_time)
@@ -175,7 +183,9 @@ class TestIntegrationRetrieveJob(IBMIntegrationJobTestCase):
         job.wait_for_final_state()
         rjobs = service.jobs(program_id=program_id, instance=default_hgp)
         self.assertEqual(program_id, rjobs[0].program_id)
-        self.assertEqual(1, len(rjobs), f"Retrieved jobs: {[j.job_id for j in rjobs]}")
+        self.assertEqual(
+            1, len(rjobs), f"Retrieved jobs: {[j.job_id() for j in rjobs]}"
+        )
 
         uuid_ = uuid.uuid4().hex
         fake_hgp = f"{uuid_}/{uuid_}/{uuid_}"
@@ -195,5 +205,5 @@ class TestIntegrationRetrieveJob(IBMIntegrationJobTestCase):
         self.assertTrue(rjobs[0], rjobs_asc[1])
         self.assertTrue(rjobs[1], rjobs_asc[0])
         self.assertEqual(
-            [job.job_id for job in rjobs], [job.job_id for job in rjobs_desc]
+            [job.job_id() for job in rjobs], [job.job_id() for job in rjobs_desc]
         )
diff --git a/test/unit/test_ibm_primitives.py b/test/unit/test_ibm_primitives.py
index 96706a3f..a19c390f 100644
--- a/test/unit/test_ibm_primitives.py
+++ b/test/unit/test_ibm_primitives.py
@@ -16,6 +16,8 @@ import sys
 from unittest.mock import MagicMock, patch, ANY
 from dataclasses import asdict
 
+from qiskit.test.reference_circuits import ReferenceCircuits
+from qiskit.quantum_info import SparsePauliOp
 from qiskit_ibm_runtime import Sampler, Estimator, Options, Session
 from qiskit_ibm_runtime.ibm_backend import IBMBackend
 import qiskit_ibm_runtime.session as session_pkg
@@ -25,6 +27,12 @@ from ..ibm_test_case import IBMTestCase
 class TestPrimitives(IBMTestCase):
     """Class for testing the Sampler class."""
 
+    @classmethod
+    def setUpClass(cls):
+        cls.qx = ReferenceCircuits.bell()
+        cls.obs = SparsePauliOp.from_list([("IZ", 1)])
+        return super().setUpClass()
+
     def tearDown(self) -> None:
         super().tearDown()
         session_pkg._DEFAULT_SESSION = None
@@ -84,7 +92,7 @@ class TestPrimitives(IBMTestCase):
                 with self.subTest(primitive=cls, backend=backend):
                     options = {"backend": backend}
                     inst = cls(service=MagicMock(), options=options)
-                    self.assertEqual(inst.session.backend, backend_name)
+                    self.assertEqual(inst.session.backend(), backend_name)
 
     @patch("qiskit_ibm_runtime.session.Session")
     @patch("qiskit_ibm_runtime.session.QiskitRuntimeService")
@@ -108,7 +116,7 @@ class TestPrimitives(IBMTestCase):
         for cls in primitives:
             with self.subTest(primitive=cls):
                 inst = cls(session=backend)
-                self.assertEqual(inst.session.backend, backend.name)
+                self.assertEqual(inst.session.backend(), backend.name)
 
     @patch("qiskit_ibm_runtime.session.Session")
     @patch("qiskit_ibm_runtime.session.QiskitRuntimeService")
@@ -141,7 +149,7 @@ class TestPrimitives(IBMTestCase):
             for options, expected in options_vars:
                 with self.subTest(primitive=cls, options=options):
                     inst = cls(session=session, options=options)
-                    inst.run(MagicMock(), MagicMock())
+                    inst.run(self.qx, observables=self.obs)
                     if sys.version_info >= (3, 8):
                         inputs = session.run.call_args.kwargs["inputs"]
                     else:
@@ -160,7 +168,7 @@ class TestPrimitives(IBMTestCase):
                 inst.options.resilience_level = 1
                 inst.options.optimization_level = 2
                 inst.options.execution.shots = 3
-                inst.run(MagicMock(), MagicMock())
+                inst.run(self.qx, observables=self.obs)
                 if sys.version_info >= (3, 8):
                     inputs = session.run.call_args.kwargs["inputs"]
                 else:
@@ -202,7 +210,7 @@ class TestPrimitives(IBMTestCase):
             for options, expected in options_vars:
                 with self.subTest(primitive=cls, options=options):
                     inst = cls(session=session)
-                    inst.run(MagicMock(), MagicMock(), **options)
+                    inst.run(self.qx, observables=self.obs, **options)
                     if sys.version_info >= (3, 8):
                         inputs = session.run.call_args.kwargs["inputs"]
                     else:
diff --git a/test/unit/test_job_retrieval.py b/test/unit/test_job_retrieval.py
index 27867be1..b1829a58 100644
--- a/test/unit/test_job_retrieval.py
+++ b/test/unit/test_job_retrieval.py
@@ -37,8 +37,8 @@ class TestRetrieveJobs(IBMTestCase):
         program_id = upload_program(service)
         params = {"param1": "foo"}
         job = run_program(service=service, program_id=program_id, inputs=params)
-        rjob = service.job(job.job_id)
-        self.assertEqual(job.job_id, rjob.job_id)
+        rjob = service.job(job.job_id())
+        self.assertEqual(job.job_id(), rjob.job_id())
         self.assertEqual(program_id, rjob.program_id)
 
     @run_quantum_and_cloud_fake
@@ -233,12 +233,12 @@ class TestRetrieveJobs(IBMTestCase):
 
         job = run_program(service=service, program_id=program_id)
         job_2 = run_program(
-            service=service, program_id=program_id, session_id=job.job_id
+            service=service, program_id=program_id, session_id=job.job_id()
         )
         with mock_wait_for_final_state(service, job):
             job.wait_for_final_state()
             job_2.wait_for_final_state()
-        rjobs = service.jobs(program_id=program_id, session_id=job.job_id)
+        rjobs = service.jobs(program_id=program_id, session_id=job.job_id())
         self.assertTrue(rjobs)
         self.assertEqual(2, len(rjobs))
         rjobs = service.jobs(program_id=program_id, session_id="no_test_session_id")
@@ -256,7 +256,7 @@ class TestRetrieveJobs(IBMTestCase):
             created_before=time_after_job,
             created_after=current_time,
         )
-        self.assertTrue(job.job_id in [j.job_id for j in rjobs])
+        self.assertTrue(job.job_id() in [j.job_id() for j in rjobs])
         self.assertTrue(job._creation_date <= time_after_job)
         self.assertTrue(job._creation_date >= current_time)
 
@@ -276,7 +276,7 @@ class TestRetrieveJobs(IBMTestCase):
         self.assertTrue(rjobs[0], rjobs_asc[1])
         self.assertTrue(rjobs[1], rjobs_asc[0])
         self.assertEqual(
-            [job.job_id for job in rjobs], [job.job_id for job in rjobs_desc]
+            [job.job_id() for job in rjobs], [job.job_id() for job in rjobs_desc]
         )
 
     def test_jobs_bad_instance(self):
@@ -299,8 +299,8 @@ class TestRetrieveJobs(IBMTestCase):
         backend_name = FakeRuntimeService.DEFAULT_UNIQUE_BACKEND_PREFIX + "1"
         job = run_program(service, program_id=program_id, backend_name=backend_name)
 
-        rjob = service.job(job.job_id)
-        self.assertIsNotNone(rjob.backend)
+        rjob = service.job(job.job_id())
+        self.assertIsNotNone(rjob.backend())
 
     def _populate_jobs_with_all_statuses(self, service, program_id):
         """Populate the database with jobs of all statuses."""
diff --git a/test/unit/test_jobs.py b/test/unit/test_jobs.py
index 5a9c45e0..c6cd5ac7 100644
--- a/test/unit/test_jobs.py
+++ b/test/unit/test_jobs.py
@@ -14,11 +14,13 @@
 
 import random
 import time
+from unittest.mock import MagicMock
 
 from qiskit.providers.exceptions import QiskitBackendNotFoundError
 from qiskit.providers.jobstatus import JobStatus
 
 from qiskit_ibm_runtime import RuntimeJob
+from qiskit_ibm_runtime.ibm_backend import IBMBackend
 from qiskit_ibm_runtime.constants import API_TO_JOB_ERROR_MESSAGE
 from qiskit_ibm_runtime.exceptions import (
     RuntimeJobFailureError,
@@ -49,7 +51,7 @@ class TestRuntimeJob(IBMTestCase):
         """Test running program."""
         params = {"param1": "foo"}
         job = run_program(service=service, inputs=params)
-        self.assertTrue(job.job_id)
+        self.assertTrue(job.job_id())
         self.assertIsInstance(job, RuntimeJob)
         self.assertIsInstance(job.status(), JobStatus)
         self.assertEqual(job.inputs, params)
@@ -84,7 +86,7 @@ class TestRuntimeJob(IBMTestCase):
             instance="crn:v1:bluemix:public:quantum-computing:my-region:a/...:...::",
         )
         job = run_program(service=service, backend_name="")
-        self.assertTrue(job.backend)
+        self.assertTrue(job.backend())
 
     def test_run_program_default_hgp_backend(self):
         """Test running a program with a backend in default hgp."""
@@ -93,9 +95,9 @@ class TestRuntimeJob(IBMTestCase):
         default_hgp = list(service._hgps.values())[0]
         self.assertIn(backend, default_hgp.backends.keys())
         job = run_program(service=service, backend_name=backend)
-        self.assertEqual(job.backend.name, backend)
+        self.assertEqual(job.backend().name, backend)
         self.assertEqual(
-            job.backend._api_client.hgp, FakeRuntimeService.DEFAULT_HGPS[0]
+            job.backend()._api_client.hgp, FakeRuntimeService.DEFAULT_HGPS[0]
         )
 
     def test_run_program_non_default_hgp_backend(self):
@@ -105,7 +107,7 @@ class TestRuntimeJob(IBMTestCase):
         default_hgp = list(service._hgps.values())[0]
         self.assertNotIn(backend, default_hgp.backends.keys())
         job = run_program(service=service, backend_name=backend)
-        self.assertEqual(job.backend.name, backend)
+        self.assertEqual(job.backend().name, backend)
 
     def test_run_program_by_hgp_backend(self):
         """Test running a program with both backend and hgp."""
@@ -115,8 +117,8 @@ class TestRuntimeJob(IBMTestCase):
         job = run_program(
             service=service, backend_name=backend, instance=non_default_hgp
         )
-        self.assertEqual(job.backend.name, backend)
-        self.assertEqual(job.backend._api_client.hgp, non_default_hgp)
+        self.assertEqual(job.backend().name, backend)
+        self.assertEqual(job.backend()._api_client.hgp, non_default_hgp)
 
     def test_run_program_by_hgp_bad_backend(self):
         """Test running a program with backend not in hgp."""
@@ -147,7 +149,7 @@ class TestRuntimeJob(IBMTestCase):
         params = {"param1": "foo"}
         image = "name:tag"
         job = run_program(service=service, inputs=params, image=image)
-        self.assertTrue(job.job_id)
+        self.assertTrue(job.job_id())
         self.assertIsInstance(job, RuntimeJob)
         self.assertIsInstance(job.status(), JobStatus)
         self.assertEqual(job.inputs, params)
@@ -161,7 +163,7 @@ class TestRuntimeJob(IBMTestCase):
     def test_run_program_with_custom_log_level(self, service):
         """Test running program with a custom image."""
         job = run_program(service=service, log_level="DEBUG")
-        job_raw = service._api_client._get_job(job.job_id)
+        job_raw = service._api_client._get_job(job.job_id())
         self.assertEqual(job_raw.log_level, "DEBUG")
 
     @run_quantum_and_cloud_fake
@@ -170,10 +172,10 @@ class TestRuntimeJob(IBMTestCase):
         job = run_program(service=service, job_classes=FailedRuntimeJob)
         with mock_wait_for_final_state(service, job):
             job.wait_for_final_state()
-            job_result_raw = service._api_client.job_results(job.job_id)
+            job_result_raw = service._api_client.job_results(job.job_id())
             self.assertEqual(JobStatus.ERROR, job.status())
             self.assertEqual(
-                API_TO_JOB_ERROR_MESSAGE["FAILED"].format(job.job_id, job_result_raw),
+                API_TO_JOB_ERROR_MESSAGE["FAILED"].format(job.job_id(), job_result_raw),
                 job.error_message(),
             )
             with self.assertRaises(RuntimeJobFailureError):
@@ -185,11 +187,11 @@ class TestRuntimeJob(IBMTestCase):
         job = run_program(service=service, job_classes=FailedRanTooLongRuntimeJob)
         with mock_wait_for_final_state(service, job):
             job.wait_for_final_state()
-            job_result_raw = service._api_client.job_results(job.job_id)
+            job_result_raw = service._api_client.job_results(job.job_id())
             self.assertEqual(JobStatus.ERROR, job.status())
             self.assertEqual(
                 API_TO_JOB_ERROR_MESSAGE["CANCELLED - RAN TOO LONG"].format(
-                    job.job_id, job_result_raw
+                    job.job_id(), job_result_raw
                 ),
                 job.error_message(),
             )
@@ -211,7 +213,7 @@ class TestRuntimeJob(IBMTestCase):
         time.sleep(1)
         job.cancel()
         self.assertEqual(job.status(), JobStatus.CANCELLED)
-        rjob = service.job(job.job_id)
+        rjob = service.job(job.job_id())
         self.assertEqual(rjob.status(), JobStatus.CANCELLED)
 
     @run_quantum_and_cloud_fake
@@ -276,7 +278,32 @@ class TestRuntimeJob(IBMTestCase):
         """Test deleting a job."""
         params = {"param1": "foo"}
         job = run_program(service=service, inputs=params)
-        self.assertTrue(job.job_id)
-        service.delete_job(job.job_id)
+        self.assertTrue(job.job_id())
+        service.delete_job(job.job_id())
         with self.assertRaises(RuntimeJobNotFound):
-            service.job(job.job_id)
+            service.job(job.job_id())
+
+    def test_job_id_attribute(self):
+        """Test using job_id as an attribute still works."""
+        job = RuntimeJob(
+            backend=MagicMock(),
+            api_client=MagicMock(),
+            client_params=MagicMock(),
+            job_id="12345",
+            program_id="foo",
+        )
+        self.assertIsInstance(job.job_id, str)
+        self.assertEqual(job.job_id, "12345")
+
+    def test_job_backend_attribute(self):
+        """Test using backend as an attribute still works."""
+        backend = MagicMock(spec=IBMBackend)
+        job = RuntimeJob(
+            backend=backend,
+            api_client=MagicMock(),
+            client_params=MagicMock(),
+            job_id="12345",
+            program_id="foo",
+        )
+        self.assertIsInstance(job.backend, IBMBackend)
+        self.assertEqual(job.backend, backend)
diff --git a/test/unit/test_session.py b/test/unit/test_session.py
index 614e01bf..534dd5b7 100644
--- a/test/unit/test_session.py
+++ b/test/unit/test_session.py
@@ -47,7 +47,7 @@ class TestSession(IBMTestCase):
         backend = MagicMock(spec=IBMBackend)
         backend.name = "ibm_gotham"
         session = Session(service=MagicMock(), backend=backend)
-        self.assertEqual(session.backend, "ibm_gotham")
+        self.assertEqual(session.backend(), "ibm_gotham")
 
     def test_max_time(self):
         """Test max time."""
@@ -74,7 +74,7 @@ class TestSession(IBMTestCase):
     def test_run(self):
         """Test the run method."""
         job = MagicMock()
-        job.job_id = "12345"
+        job.job_id.return_value = "12345"
         service = MagicMock()
         service.run.return_value = job
         backend = "ibm_gotham"
@@ -84,7 +84,7 @@ class TestSession(IBMTestCase):
         decoder = MagicMock()
         max_time = 42
         session = Session(service=service, backend=backend, max_time=max_time)
-        session_ids = [None, job.job_id]
+        session_ids = [None, job.job_id()]
         start_sessions = [True, False]
 
         for idx in range(2):
@@ -101,8 +101,8 @@ class TestSession(IBMTestCase):
             self.assertEqual(kwargs["session_id"], session_ids[idx])
             self.assertEqual(kwargs["start_session"], start_sessions[idx])
             self.assertEqual(kwargs["result_decoder"], decoder)
-            self.assertEqual(session.session_id, job.job_id)
-            self.assertEqual(session.backend, backend)
+            self.assertEqual(session.session_id, job.job_id())
+            self.assertEqual(session.backend(), backend)
 
     def test_close_without_run(self):
         """Test closing without run."""
@@ -122,14 +122,14 @@ class TestSession(IBMTestCase):
     def test_default_backend(self):
         """Test default backend set."""
         job = MagicMock()
-        job.backend.name = "ibm_gotham"
+        job.backend().name = "ibm_gotham"
         service = MagicMock()
         service.run.return_value = job
         service.channel = "ibm_cloud"
 
         session = Session(service=service)
         session.run(program_id="foo", inputs={})
-        self.assertEqual(session.backend, "ibm_gotham")
+        self.assertEqual(session.backend(), "ibm_gotham")
 
     def test_opening_default_session(self):
         """Test opening default session."""
@@ -138,7 +138,7 @@ class TestSession(IBMTestCase):
         session = get_default_session(service=service, backend=backend)
         self.assertIsInstance(session, Session)
         self.assertEqual(session.service, service)
-        self.assertEqual(session.backend, backend)
+        self.assertEqual(session.backend(), backend)
 
         session2 = get_default_session(service=service, backend=backend)
         self.assertEqual(session, session2)
@@ -153,7 +153,7 @@ class TestSession(IBMTestCase):
         session2 = get_default_session(service=service, backend=backend)
         self.assertNotEqual(session, session2)
         self.assertEqual(session2.service, service)
-        self.assertEqual(session2.backend, backend)
+        self.assertEqual(session2.backend(), backend)
 
     def test_default_session_different_backend(self):
         """Test default session backend change."""
@@ -162,7 +162,7 @@ class TestSession(IBMTestCase):
         session2 = get_default_session(service=service, backend="ibm_metropolis")
         self.assertNotEqual(session, session2)
         self.assertFalse(session._active)
-        self.assertEqual(session2.backend, "ibm_metropolis")
+        self.assertEqual(session2.backend(), "ibm_metropolis")
         self.assertTrue(session2._active)
 
     def test_default_session_different_service(self):
@@ -181,7 +181,7 @@ class TestSession(IBMTestCase):
         backend = "ibm_gotham"
         session = get_default_session(backend=backend)
         self.assertIsInstance(session, Session)
-        self.assertEqual(session.backend, backend)
+        self.assertEqual(session.backend(), backend)
         mock_service.assert_called_once()
 
     def test_default_session_backend_service(self):
diff --git a/test/utils.py b/test/utils.py
index 293b994a..1ec2832f 100644
--- a/test/utils.py
+++ b/test/utils.py
@@ -109,7 +109,7 @@ def cancel_job_safe(job: RuntimeJob, logger: logging.Logger) -> bool:
         assert (
             status is JobStatus.CANCELLED
         ), "cancel() was successful for job {} but its " "status is {}.".format(
-            job.job_id, status
+            job.job_id(), status
         )
         return True
     except RuntimeInvalidStateError:
@@ -126,7 +126,7 @@ def wait_for_status(job, status, poll_time=1, time_out=20):
         time.sleep(wait_time)
         time_out -= wait_time
     if job.status() != status:
-        raise unittest.SkipTest(f"Job {job.job_id} unable to reach status {status}.")
+        raise unittest.SkipTest(f"Job {job.job_id()} unable to reach status {status}.")
 
 
 def get_real_device(service):
@@ -145,5 +145,5 @@ def mock_wait_for_final_state(service, job):
     return mock.patch.object(
         RuntimeJob,
         "wait_for_final_state",
-        side_effect=service._api_client.wait_for_final_state(job.job_id),
+        side_effect=service._api_client.wait_for_final_state(job.job_id()),
     )