Source code for qiskit_ibm_runtime.session

# 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.

"""Qiskit Runtime flexible session."""

from typing import Dict, Optional, Type, Union, Callable
from types import TracebackType
from functools import wraps
from contextvars import ContextVar

from qiskit_ibm_provider.utils.converters import hms_to_seconds

from qiskit_ibm_runtime import QiskitRuntimeService
from .runtime_job import RuntimeJob
from .runtime_program import ParameterNamespace
from .program.result_decoder import ResultDecoder
from .ibm_backend import IBMBackend

def _active_session(func):  # type: ignore
    """Decorator used to ensure the session is active."""

    def _wrapper(self, *args, **kwargs):  # type: ignore
        if not self._active:
            raise RuntimeError("The session is closed.")
        return func(self, *args, **kwargs)

    return _wrapper

[docs] class Session: """Class for creating a flexible Qiskit Runtime session. A Qiskit Runtime ``session`` allows you to group a collection of iterative calls to the quantum computer. A session is started when the first job within the session is started. Subsequent jobs within the session are prioritized by the scheduler. Data used within a session, such as transpiled circuits, is also cached to avoid unnecessary overhead. You can open a Qiskit Runtime session using this ``Session`` class and submit jobs to one or more primitives. For example:: from qiskit.test.reference_circuits import ReferenceCircuits from qiskit_ibm_runtime import Sampler, Session, Options options = Options(optimization_level=3) with Session(backend="ibmq_qasm_simulator") as session: sampler = Sampler(session=session, options=options) job = sampler.run(ReferenceCircuits.bell()) print(f"Sampler job ID: {job.job_id()}") print(f"Sampler job result: {job.result()}") # Close the session only if all jobs are finished and # you don't need to run more in the session. session.close() """ def __init__( self, service: Optional[QiskitRuntimeService] = None, backend: Optional[Union[str, IBMBackend]] = None, max_time: Optional[Union[int, str]] = None, ): # pylint: disable=line-too-long """Session constructor. Args: service: Optional instance of the ``QiskitRuntimeService`` class. If ``None``, the service associated with the backend, if known, is used. Otherwise ``QiskitRuntimeService()`` is used to initialize your default saved account. backend: Optional instance of :class:`qiskit_ibm_runtime.IBMBackend` class or string name of backend. An instance of :class:`qiskit_ibm_provider.IBMBackend` will not work. If not specified, a backend will be selected automatically (IBM Cloud channel only). max_time: (EXPERIMENTAL setting, can break between releases without warning) Maximum amount of time, a runtime session can be open before being forcibly closed. Can be specified as seconds (int) or a string like "2h 30m 40s". This value must be less than the `system imposed maximum <https://qiskit.org/documentation/partners/qiskit_ibm_runtime/faqs/max_execution_time.html>`_. Raises: ValueError: If an input value is invalid. """ if service is None: if isinstance(backend, IBMBackend): self._service = backend.service else: self._service = ( QiskitRuntimeService() 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._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: ") )
[docs] @_active_session def run( self, program_id: str, inputs: Union[Dict, ParameterNamespace], options: Optional[Dict] = None, callback: Optional[Callable] = None, result_decoder: Optional[Type[ResultDecoder]] = None, ) -> RuntimeJob: """Run a program in the session. Args: program_id: Program ID. inputs: Program input parameters. These input values are passed to the runtime program. options: Runtime options that control the execution environment. See :class:`qiskit_ibm_runtime.RuntimeOptions` for all available options. callback: Callback function to be invoked for any interim results and final result. Returns: Submitted job. """ options = options or {} if "instance" not in options: options["instance"] = self._instance options["backend"] = self._backend if not self._session_id: # 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 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() if self._backend is None: self._backend = job.backend().name return job
[docs] def close(self) -> None: """Close the session.""" self._active = False if self._session_id: self._service._api_client.close_session(self._session_id)
[docs] 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. Returns: Session ID. None until a job runs in the session. """ return self._session_id @property def service(self) -> QiskitRuntimeService: """Return service associated with this session. Returns: :class:`qiskit_ibm_runtime.QiskitRuntimeService` associated with this session. """ return self._service
[docs] @classmethod def from_id( cls, session_id: str, service: Optional[QiskitRuntimeService] = None, backend: Optional[Union[str, IBMBackend]] = None, max_time: Optional[Union[int, str]] = None, ) -> "Session": """Construct a Session object with a given session_id""" session = cls(service, backend, max_time) session._session_id = session_id return session
def __enter__(self) -> "Session": set_cm_session(self) return self def __exit__( self, exc_type: Optional[Type[BaseException]], exc_val: Optional[BaseException], exc_tb: Optional[TracebackType], ) -> None: set_cm_session(None)
# Default session _DEFAULT_SESSION: ContextVar[Optional[Session]] = ContextVar("_DEFAULT_SESSION", default=None) _IN_SESSION_CM: ContextVar[bool] = ContextVar("_IN_SESSION_CM", default=False) def set_cm_session(session: Optional[Session]) -> None: """Set the context manager session.""" _DEFAULT_SESSION.set(session) _IN_SESSION_CM.set(session is not None) def get_cm_session() -> Session: """Return the context managed session.""" return _DEFAULT_SESSION.get()