Use POST /sessions endpoint

qiskit_ibm_runtime/api/clients/runtime.py

@@ -230,6 +230,23 @@ def job_metadata(self, job_id: str) -> Dict[str, Any]:
         """
         return self._api.program_job(job_id).metadata()
 
+    def create_session(
+        self,
+        backend: Optional[str] = None,
+        instance: Optional[str] = None,
+        max_time: Optional[int] = None,
+        channel: Optional[str] = None,
+        mode: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        """Create a session.
+
+        Args:
+            mode: Execution mode.
+        """
+        return self._api.runtime_session(session_id=None).create(
+            backend, instance, max_time, channel, mode
+        )
+
     def cancel_session(self, session_id: str) -> None:
         """Close all jobs in the runtime session.
 

qiskit_ibm_runtime/api/rest/runtime.py

@@ -49,7 +49,7 @@ def program_job(self, job_id: str) -> "ProgramJob":
         """
         return ProgramJob(self.session, job_id)
 
-    def runtime_session(self, session_id: str) -> "RuntimeSession":
+    def runtime_session(self, session_id: str = None) -> "RuntimeSession":
         """Return an adapter for the session.
 
         Args:

qiskit_ibm_runtime/api/rest/runtime_session.py

@@ -12,7 +12,7 @@
 
 """Runtime Session REST adapter."""
 
-from typing import Dict, Any
+from typing import Dict, Any, Optional
 from .base import RestAdapterBase
 from ..session import RetrySession
 from ..exceptions import RequestsApiError
@@ -35,7 +35,34 @@ def __init__(self, session: RetrySession, session_id: str, url_prefix: str = "")
             session_id: Job ID of the first job in a runtime session.
             url_prefix: Prefix to use in the URL.
         """
-        super().__init__(session, "{}/sessions/{}".format(url_prefix, session_id))
+        if not session_id:
+            super().__init__(session, "{}/sessions".format(url_prefix))
+        else:
+            super().__init__(session, "{}/sessions/{}".format(url_prefix, session_id))
+
+    def create(
+        self,
+        backend: Optional[str] = None,
+        instance: Optional[str] = None,
+        max_time: Optional[int] = None,
+        channel: Optional[str] = None,
+        mode: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        """Create a session"""
+        url = self.get_url("self")
+        payload = {}
+        if mode:
+            payload["mode"] = mode
+        if backend:
+            payload["backend"] = backend
+        if instance:
+            payload["instance"] = instance
+        if max_time:
+            if channel == "ibm_quantum":
+                payload["max_session_ttl"] = max_time  # type: ignore[assignment]
+            else:
+                payload["max_ttl"] = max_time  # type: ignore[assignment]
+        return self.session.post(url, json=payload).json()
 
     def cancel(self) -> None:
         """Cancel all jobs in the session."""

qiskit_ibm_runtime/batch.py

@@ -12,10 +12,27 @@
 
 """Qiskit Runtime batch mode."""
 
+from typing import Optional, Union
+from qiskit_ibm_runtime import QiskitRuntimeService
+from .ibm_backend import IBMBackend
+
 from .session import Session
 
 
 class Batch(Session):
     """Class for creating a batch mode in Qiskit Runtime."""
 
-    pass
+    def __init__(
+        self,
+        service: Optional[QiskitRuntimeService] = None,
+        backend: Optional[Union[str, IBMBackend]] = None,
+        max_time: Optional[Union[int, str]] = None,
+    ):
+        super().__init__(service=service, backend=backend, max_time=max_time)
+
+    def _create_session(self) -> str:
+        """Create a session."""
+        session = self._service._api_client.create_session(
+            self._backend, self._instance, self._max_time, self._service.channel, "batch"
+        )
+        return session.get("id")

qiskit_ibm_runtime/runtime_job.py

@@ -657,7 +657,7 @@ def session_id(self) -> str:
         """Session ID.
 
         Returns:
-            Job ID of the first job in a runtime session.
+            Session ID. None if the backend is a simulator.
         """
         if not self._session_id:
             response = self._api_client.job_get(job_id=self.job_id())

qiskit_ibm_runtime/session.py

@@ -15,7 +15,6 @@
 from typing import Dict, Optional, Type, Union, Callable, Any
 from types import TracebackType
 from functools import wraps
-from threading import Lock
 
 from qiskit_ibm_runtime import QiskitRuntimeService
 from .runtime_job import RuntimeJob
@@ -110,28 +109,43 @@ def __init__(
                     if QiskitRuntimeService.global_service is None
                     else QiskitRuntimeService.global_service
                 )
-
         else:
             self._service = service
 
         if self._service.channel == "ibm_quantum" and not backend:
             raise ValueError('"backend" is required for ``ibm_quantum`` channel.')
 
         self._instance = None
-        if isinstance(backend, IBMBackend):
-            self._instance = backend._instance
-            backend = backend.name
-        self._backend = backend
 
-        self._setup_lock = Lock()
-        self._session_id: Optional[str] = None
         self._active = True
         self._max_time = (
             max_time
             if max_time is None or isinstance(max_time, int)
             else hms_to_seconds(max_time, "Invalid max_time value: ")
         )
 
+        if isinstance(backend, IBMBackend):
+            self._instance = backend._instance
+            sim_backend = backend.configuration().simulator
+            backend = backend.name
+        else:
+            backend_obj = self._service.backend(backend)
+            self._instance = backend_obj._instance
+            sim_backend = backend_obj.configuration().simulator
+        self._backend = backend
+
+        if not sim_backend:
+            self._session_id = self._create_session()
+        else:
+            self._session_id = None
+
+    def _create_session(self) -> str:
+        """Create a session."""
+        session = self._service._api_client.create_session(
+            self._backend, self._instance, self._max_time, self._service.channel
+        )
+        return session.get("id")
+
     @_active_session
     def run(
         self,
@@ -162,29 +176,15 @@ def run(
 
         options["backend"] = self._backend
 
-        if not self._session_id:
-            # Make sure only one thread can send the session starter job.
-            self._setup_lock.acquire()
-            # TODO: What happens if session max time != first job max time?
-            # Use session max time if this is first job.
-            options["session_time"] = self._max_time
-
-        try:
-            job = self._service.run(
-                program_id=program_id,
-                options=options,
-                inputs=inputs,
-                session_id=self._session_id,
-                start_session=self._session_id is None,
-                callback=callback,
-                result_decoder=result_decoder,
-            )
-
-            if self._session_id is None:
-                self._session_id = job.job_id()
-        finally:
-            if self._setup_lock.locked():
-                self._setup_lock.release()
+        job = self._service.run(
+            program_id=program_id,
+            options=options,
+            inputs=inputs,
+            session_id=self._session_id,
+            start_session=False,
+            callback=callback,
+            result_decoder=result_decoder,
+        )
 
         if self._backend is None:
             self._backend = job.backend().name
@@ -278,11 +278,11 @@ def details(self) -> Optional[Dict[str, Any]]:
         return None
 
     @property
-    def session_id(self) -> str:
+    def session_id(self) -> Optional[str]:
         """Return the session ID.
 
         Returns:
-            Session ID. None until a job is submitted.
+            Session ID. None if the backend is a simulator.
         """
         return self._session_id
 

qiskit_ibm_runtime/utils/json.py

@@ -215,7 +215,7 @@ def default(self, obj: Any) -> Any:  # pylint: disable=arguments-differ
         if hasattr(obj, "to_json"):
             return {"__type__": "to_json", "__value__": obj.to_json()}
         if isinstance(obj, QuantumCircuit):
-            kwargs: dict[str, object] = {"use_symengine": bool(optionals.HAS_SYMENGINE)}
+            kwargs: Dict[str, object] = {"use_symengine": bool(optionals.HAS_SYMENGINE)}
             if _TERRA_VERSION[0] >= 1:
                 # NOTE: This can be updated only after the server side has
                 # updated to a newer qiskit version.

releasenotes/notes/session-modes-5c22b68620f8d690.yaml

@@ -0,0 +1,10 @@
+---
+features:
+  - |
+    Sessions will now be started with a new ``/sessions`` endpoint that allows for different 
+    execution modes. Batch mode is now supported through :class:`~qiskit_ibm_runtime.Batch`, and 
+    :class:`~qiskit_ibm_runtime.Session` will work the same as way as before. 
+    Please see https://docs.quantum.ibm.com/run/sessions for more information.
+
+    Note that ``Session`` and ``Batch`` created from ``qiskit-ibm-runtime`` prior to this release will no longer be 
+    supported after March 31, 2024. Please update your ``qiskit-ibm-runtime`` version as soon as possible before this date. 

test/unit/mock/fake_runtime_client.py

@@ -285,6 +285,17 @@ def set_job_classes(self, classes):
             classes = [classes]
         self._job_classes = classes
 
+    # pylint: disable=unused-argument
+    def create_session(
+        self, backend: str = None, instance: str = None, mode: str = None
+    ) -> Dict[str, Any]:
+        """Create a session."""
+        return {"id": uuid.uuid4().hex}
+
+    def close_session(self, session_id: str) -> None:
+        """Close a session."""
+        pass
+
     def is_qctrl_enabled(self):
         """Return whether or not channel_strategy q-ctrl is enabled."""
         return False

test/unit/test_batch.py

@@ -12,25 +12,48 @@
 
 """Tests for Batch class."""
 
-from unittest.mock import patch
+from unittest.mock import MagicMock
 
 from qiskit_ibm_runtime import Batch
+from qiskit_ibm_runtime.ibm_backend import IBMBackend
 from qiskit_ibm_runtime.utils.default_session import _DEFAULT_SESSION
 
 from ..ibm_test_case import IBMTestCase
 
 
 class TestBatch(IBMTestCase):
-    """Class for testing the Session class."""
+    """Class for testing the Batch class."""
 
     def tearDown(self) -> None:
         super().tearDown()
         _DEFAULT_SESSION.set(None)
 
-    @patch("qiskit_ibm_runtime.session.QiskitRuntimeService", autospec=True)
-    def test_default_batch(self, mock_service):
-        """Test using default batch mode."""
-        mock_service.global_service = None
-        batch = Batch(backend="ibm_gotham")
-        self.assertIsNotNone(batch.service)
-        mock_service.assert_called_once()
+    def test_passing_ibm_backend(self):
+        """Test passing in IBMBackend instance."""
+        backend = MagicMock(spec=IBMBackend)
+        backend._instance = None
+        backend.name = "ibm_gotham"
+        session = Batch(service=MagicMock(), backend=backend)
+        self.assertEqual(session.backend(), "ibm_gotham")
+
+    def test_using_ibm_backend_service(self):
+        """Test using service from an IBMBackend instance."""
+        backend = MagicMock(spec=IBMBackend)
+        backend._instance = None
+        backend.name = "ibm_gotham"
+        session = Batch(backend=backend)
+        self.assertEqual(session.service, backend.service)
+
+    def test_run_after_close(self):
+        """Test running after session is closed."""
+        session = Batch(service=MagicMock(), backend="ibm_gotham")
+        session.cancel()
+        with self.assertRaises(RuntimeError):
+            session.run(program_id="program_id", inputs={})
+
+    def test_context_manager(self):
+        """Test session as a context manager."""
+        with Batch(service=MagicMock(), backend="ibm_gotham") as session:
+            session.run(program_id="foo", inputs={})
+            session.cancel()
+        self.assertFalse(session._active)

test/unit/test_runtime_ws.py

@@ -91,6 +91,7 @@ def _patched_run(callback, *args, **kwargs):  # pylint: disable=unused-argument
         service = MagicMock(spec=QiskitRuntimeService)
         service.run = _patched_run
         service._channel_strategy = None
+        service._api_client = MagicMock()
 
         circ = bell()
         obs = SparsePauliOp.from_list([("IZ", 1)])