Expose new session details

qiskit_ibm_runtime/api/clients/runtime.py

@@ -345,6 +345,17 @@ def close_session(self, session_id: str) -> None:
         """
         self._api.runtime_session(session_id=session_id).close()
 
+    def session_details(self, session_id: str) -> Dict[str, Any]:
+        """Get session details.
+
+        Args:
+            session_id: Session ID.
+
+        Returns:
+            Session details.
+        """
+        return self._api.runtime_session(session_id=session_id).details()
+
     def list_backends(
         self, hgp: Optional[str] = None, channel_strategy: Optional[str] = None
     ) -> List[str]:

qiskit_ibm_runtime/api/rest/base.py

@@ -0,0 +1,57 @@
+# This code is part of Qiskit.
+#
+# (C) Copyright IBM 2021.
+#
+# This code is licensed under the Apache License, Version 2.0. You may
+# obtain a copy of this license in the LICENSE.txt file in the root directory
+# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
+#
+# Any modifications or derivative works of this code must retain this
+# copyright notice, and modified files need to carry a notice indicating
+# that they have been altered from the originals.
+
+"""Base REST adapter."""
+
+from ..session import RetrySession
+
+
+class RestAdapterBase:
+    """Base class for REST adapters."""
+
+    URL_MAP = {}  # type: ignore[var-annotated]
+    """Mapping between the internal name of an endpoint and the actual URL."""
+
+    _HEADER_JSON_CONTENT = {"Content-Type": "application/json"}
+
+    def __init__(self, session: RetrySession, prefix_url: str = "") -> None:
+        """RestAdapterBase constructor.
+
+        Args:
+            session: Session to be used in the adapter.
+            prefix_url: String to be prepend to all URLs.
+        """
+        self.session = session
+        self.prefix_url = prefix_url
+
+    def get_url(self, identifier: str) -> str:
+        """Return the resolved URL for the specified identifier.
+
+        Args:
+            identifier: Internal identifier of the endpoint.
+
+        Returns:
+            The resolved URL of the endpoint (relative to the session base URL).
+        """
+        return "{}{}".format(self.prefix_url, self.URL_MAP[identifier])
+
+    def get_prefixed_url(self, prefix: str, identifier: str) -> str:
+        """Return an adjusted URL for the specified identifier.
+
+        Args:
+            prefix: string to be prepended to the URL.
+            identifier: Internal identifier of the endpoint.
+
+        Returns:
+            The resolved facade URL of the endpoint.
+        """
+        return "{}{}{}".format(prefix, self.prefix_url, self.URL_MAP[identifier])

qiskit_ibm_runtime/api/rest/runtime.py

@@ -19,9 +19,9 @@
 
 from qiskit_ibm_provider.api.rest.base import RestAdapterBase
 from qiskit_ibm_provider.api.rest.program_job import ProgramJob
-from qiskit_ibm_provider.api.rest.runtime_session import RuntimeSession
 from qiskit_ibm_provider.utils import local_to_utc
 
+from .runtime_session import RuntimeSession
 from .program import Program
 from ...utils import RuntimeEncoder
 from .cloud_backend import CloudBackend

qiskit_ibm_runtime/api/rest/runtime_session.py

@@ -0,0 +1,49 @@
+# This code is part of Qiskit.
+#
+# (C) Copyright IBM 2022.
+#
+# This code is licensed under the Apache License, Version 2.0. You may
+# obtain a copy of this license in the LICENSE.txt file in the root directory
+# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
+#
+# Any modifications or derivative works of this code must retain this
+# copyright notice, and modified files need to carry a notice indicating
+# that they have been altered from the originals.
+
+"""Runtime Session REST adapter."""
+
+from typing import Dict, Any
+from .base import RestAdapterBase
+from ..session import RetrySession
+
+
+class RuntimeSession(RestAdapterBase):
+    """Rest adapter for session related endpoints."""
+
+    URL_MAP = {
+        "self": "",
+        "close": "/close",
+    }
+
+    def __init__(self, session: RetrySession, session_id: str, url_prefix: str = "") -> None:
+        """Job constructor.
+
+        Args:
+            session: RetrySession to be used in the adapter.
+            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))
+
+    def close(self) -> None:
+        """Close this session."""
+        url = self.get_url("close")
+        self.session.delete(url)
+
+    def details(self) -> Dict[str, Any]:
+        """Return the details of this session."""
+        try:
+            return self.session.get(self.get_url("self")).json()
+        # return None if API is not supported
+        except:  # pylint: disable=bare-except
+            return None

qiskit_ibm_runtime/session.py

@@ -12,7 +12,7 @@
 
 """Qiskit Runtime flexible session."""
 
-from typing import Dict, Optional, Type, Union, Callable
+from typing import Dict, Optional, Type, Union, Callable, Any
 from types import TracebackType
 from functools import wraps
 from contextvars import ContextVar
@@ -193,6 +193,70 @@ def backend(self) -> Optional[str]:
         """
         return self._backend
 
+    def status(self) -> Optional[str]:
+        """Return current session status.
+
+        Returns:
+            The current status of the session, including:
+                Pending: Session is created but no jobs have been dequeued, or
+                    the ``interactive_timeout`` expired before more jobs were available
+                In progress, accepting new jobs: session is active and accepting new jobs
+                In progress, not accepting new jobs: session is active and not accepting new jobs
+                Closed: ``max_time`` expired or session was explicitly closed
+                None: status details are not available
+        """
+        details = self.details()
+        if details:
+            state = details["state"]
+            accepting_jobs = details["accepting_jobs"]
+            if state in ["open", "inactive"]:
+                return "Pending"
+            if state == "active" and accepting_jobs:
+                return "In progress, accepting new jobs"
+            if state == "active" and not accepting_jobs:
+                return "In progress, not accepting new jobs"
+            return state.capitalize()
+
+        return None
+
+    def details(self) -> Optional[Dict[str, Any]]:
+        """Return session details.
+
+        Returns:
+            A dictionary with session details, including:
+                id: id of the session
+                backned_name: backend used for the session
+                interactive_timeout: The max time (in seconds) between jobs that
+                    is allowed to occur to keep the session active
+                max_time: Max allowed time for the session to run in seconds, subject to plan limits
+                active_timeout: The remaining time (in seconds) for the session to
+                    be in the active state whilst jobs are running
+                state: State of the session - open, active, inactive, or closed
+                accepting_jobs: Whether or not the session is accepting jobs
+                last_job_started: Timestamp of when the last job in the session started
+                last_job_completed: Timestamp of when the last job in the session completed
+                started_at: Timestamp of when the session was started
+                closed_at: Timestamp of when the session was closed
+
+        """
+        if self._session_id:
+            response = self._service._api_client.session_details(self._session_id)
+            if response:
+                return {
+                    "id": response.get("id"),
+                    "backend_name": response.get("backend_name"),
+                    "interactive_timeout": response.get("interactive_ttl"),
+                    "max_time": response.get("max_ttl"),
+                    "active_timeout": response.get("active_ttl"),
+                    "state": response.get("state"),
+                    "accepting_jobs": response.get("accepting_jobs"),
+                    "last_job_started": response.get("last_job_started"),
+                    "last_job_completed": response.get("last_job_completed"),
+                    "started_at": response.get("started_at"),
+                    "closed_at": response.get("closed_at"),
+                }
+        return None
+
     @property
     def session_id(self) -> str:
         """Return the session ID.

releasenotes/notes/expose-session-details-c4a44316d30dad33.yaml

@@ -0,0 +1,10 @@
+---
+features:
+  - |
+    Added a new method, :meth:`~qiskit_ibm_runtime.Session.details` that returns information
+    about a session, including: maximum session time, active time remaining, the current state, 
+    and whether or not the session is accepting jobs.
+
+    Also added :meth:`~qiskit_ibm_runtime.Session.status`, which returns the current status of 
+    the session.
+