Use POST /sessions endpoint

qiskit_ibm_provider/api/clients/runtime.py

@@ -315,6 +315,25 @@ def update_tags(self, job_id: str, tags: list) -> Response:
         """
         return self._api.program_job(job_id).update_tags(tags)
 
+    def create_session(
+        self,
+        backend: Optional[str] = None,
+        instance: Optional[str] = None,
+        max_time: Optional[int] = None,
+        mode: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        """Create a new runtime session.
+
+        Args:
+            backend: The name of the backend to use.
+            instance: The instance to use.
+            mode: The mode to use.
+
+        Returns:
+            The created session.
+        """
+        return self._api.runtime_session().create(backend, instance, max_time, mode)
+
     def close_session(self, session_id: str) -> None:
         """Close session
 

qiskit_ibm_provider/api/rest/runtime.py

@@ -57,7 +57,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_provider/api/rest/runtime_session.py

@@ -12,7 +12,7 @@
 
 """Runtime Session REST adapter."""
 
-
+from typing import Dict, Any, Optional
 from qiskit_ibm_provider.api.rest.base import RestAdapterBase
 from qiskit_ibm_provider.exceptions import IBMApiError
 from ..exceptions import RequestsApiError
@@ -37,7 +37,30 @@ def __init__(
             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,
+        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:
+            payload["max_session_ttl"] = max_time  # type: ignore[assignment]
+        return self.session.post(url, json=payload).json()
 
     def close(self) -> None:
         """Set accepting_jobs flag to false, so no more jobs can be submitted."""

qiskit_ibm_provider/ibm_backend.py

@@ -511,18 +511,11 @@ def _runtime_run(
         """Runs the runtime program and returns the corresponding job object"""
         hgp_name = self._instance or self.provider._get_hgp().name
 
-        session = self._session
-
-        if session:
-            if not session.active:
-                raise RuntimeError(f"The session {session.session_id} is closed.")
-            session_id = session.session_id
-            session_time = session._max_time
-            start_session = session_id is None
-        else:
-            session_id = None
-            session_time = None
-            start_session = False
+        session_id = None
+        if self._session:
+            if not self._session.active:
+                raise RuntimeError(f"The session {self._session.session_id} is closed.")
+            session_id = self._session.session_id
 
         try:
             response = self.provider._runtime_client.program_run(
@@ -532,15 +525,11 @@ def _runtime_run(
                 hgp=hgp_name,
                 job_tags=job_tags,
                 session_id=session_id,
-                start_session=start_session,
-                session_time=session_time,
+                start_session=False,
                 image=image,
             )
         except RequestsApiError as ex:
             raise IBMBackendApiError("Error submitting job: {}".format(str(ex))) from ex
-        session_id = response.get("session_id")
-        if self._session:
-            self._session._session_id = session_id
         try:
             job = IBMCircuitJob(
                 backend=self,
@@ -873,7 +862,13 @@ def _check_faulty(self, circuit: QuantumCircuit) -> None:
 
     def open_session(self, max_time: Optional[Union[int, str]] = None) -> Session:
         """Open session"""
-        self._session = Session(max_time)
+        if not self._configuration.simulator:
+            new_session = self.provider._runtime_client.create_session(
+                self.name, self._instance, max_time
+            )
+            self._session = Session(max_time=max_time, session_id=new_session.get("id"))
+        else:
+            self._session = Session()
         return self._session
 
     @property

qiskit_ibm_provider/session.py

@@ -63,6 +63,7 @@ class Session:
     def __init__(
         self,
         max_time: Optional[Union[int, str]] = None,
+        session_id: Optional[str] = None,
     ):
         """Session constructor.
 
@@ -78,7 +79,7 @@ def __init__(
             ValueError: If an input value is invalid.
         """
         self._instance = None
-        self._session_id: Optional[str] = None
+        self._session_id = session_id
         self._active = True
 
         self._max_time = (

test/unit/mock/fake_account_client.py

@@ -29,7 +29,8 @@ class FakeApiBackend:
     def __init__(self, config_update=None, status_update=None):
         fake_backend = Fake5QV1()
         self.properties = fake_backend.properties().to_dict()
-        self.defaults = fake_backend.defaults().to_dict()
+        if hasattr(fake_backend, "defaults"):
+            self.defaults = fake_backend.defaults().to_dict()
 
         self.configuration = fake_backend.configuration().to_dict()
         self.configuration["online_date"] = python_datetime.now().isoformat()

test/unit/transpiler/passes/scheduling/test_scheduler.py

@@ -20,10 +20,7 @@
 from qiskit.transpiler.passmanager import PassManager
 from qiskit.transpiler.exceptions import TranspilerError
 
-try:
-    from qiskit.providers.fake_provider import Fake7QPulseV1
-except ImportError:
-    from qiskit.providers.fake_provider import FakeJakarta as Fake7QPulseV1
+from qiskit_ibm_runtime.fake_provider import FakeJakarta
 
 from qiskit_ibm_provider.transpiler.passes.scheduling.pad_delay import PadDelay
 from qiskit_ibm_provider.transpiler.passes.scheduling.scheduler import (
@@ -850,12 +847,12 @@ def test_c_if_plugin_conversion_with_transpile(self):
         after transpilation with the plugin."""
         # Patch the test backend with the plugin
         with patch.object(
-            Fake7QPulseV1,
+            FakeJakarta,
             "get_translation_stage_plugin",
             return_value="ibm_dynamic_circuits",
             create=True,
         ):
-            backend = Fake7QPulseV1()
+            backend = FakeJakarta()
             # Temporary workaround for mock backends. For real backends this is not required.
             backend.configuration().basis_gates.append("if_else")
 
@@ -1867,7 +1864,7 @@ def test_for_loop(self):
 
     def test_transpile_mock_backend(self):
         """Test scheduling works with transpilation."""
-        backend = Fake7QPulseV1()
+        backend = FakeJakarta()
         # Temporary workaround for mock backends. For real backends this is not required.
         backend.configuration().basis_gates.append("if_else")
         backend.configuration().basis_gates.append("while_loop")
@@ -1915,7 +1912,7 @@ def test_transpile_mock_backend(self):
 
     def test_transpile_both_paths(self):
         """Test scheduling works with both fast- and standard path after transpiling."""
-        backend = Fake7QPulseV1()
+        backend = FakeJakarta()
         # Temporary workaround for mock backends. For real backends this is not required.
         backend.configuration().basis_gates.append("if_else")
 
@@ -1957,12 +1954,12 @@ def test_c_if_plugin_conversion_with_transpile(self):
         transpilation with the plugin."""
         # Patch the test backend with the plugin
         with patch.object(
-            Fake7QPulseV1,
+            FakeJakarta,
             "get_translation_stage_plugin",
             return_value="ibm_dynamic_circuits",
             create=True,
         ):
-            backend = Fake7QPulseV1()
+            backend = FakeJakarta()
             # Temporary workaround for mock backends. For real backends this is not required.
             backend.configuration().basis_gates.append("if_else")