ionq_core.session

  1# SPDX-FileCopyrightText: 2026 IonQ, Inc.
  2# SPDX-License-Identifier: Apache-2.0
  3
  4"""Session lifecycle manager for IonQ QPU sessions.
  5
  6Sessions allow you to reserve priority access to a QPU backend. The
  7`SessionManager` class wraps the session create / end / status APIs
  8and supports both sync and async context managers for automatic cleanup.
  9
 10Example:
 11    ```python
 12    from ionq_core import IonQClient, SessionManager
 13
 14    client = IonQClient()
 15
 16    # Context manager creates and automatically ends the session
 17    with SessionManager(client, "qpu.aria-1", max_jobs=10) as session:
 18        print(session.session_id)
 19        print(session.status())  # "started"
 20        # ... submit jobs using session.session_id ...
 21
 22    # Or reconnect to an existing session
 23    session = SessionManager.from_id(client, "existing-session-id")
 24    print(session.status())
 25    ```
 26"""
 27
 28from __future__ import annotations
 29
 30__all__ = ["SessionManager"]
 31
 32import logging
 33from typing import TYPE_CHECKING
 34
 35from .api.default import create_session, end_session, get_session
 36from .exceptions import IonQError
 37from .models.create_session_request import CreateSessionRequest
 38from .models.session_cost_limit import SessionCostLimit
 39from .models.session_settings_request import SessionSettingsRequest
 40from .types import UNSET, Unset
 41
 42if TYPE_CHECKING:
 43    from .client import AuthenticatedClient
 44
 45logger = logging.getLogger("ionq_core")
 46
 47
 48class SessionManager:
 49    """Convenience wrapper around session create / end / status APIs.
 50
 51    Can be used as both a sync and async context manager. On exit the
 52    session is automatically ended. Exceptions during close are logged
 53    and suppressed so that cleanup does not mask the original error.
 54
 55    Args:
 56        client: An authenticated API client.
 57        backend: The backend to create a session on (e.g. ``"qpu.aria-1"``).
 58        max_jobs: Optional maximum number of jobs for this session.
 59        max_time: Optional maximum session duration in minutes.
 60        max_cost: Optional maximum cost in USD for the session.
 61
 62    Examples:
 63        Sync context manager:
 64
 65        ```python
 66        with SessionManager(client, "qpu.aria-1", max_jobs=10) as session:
 67            print(session.session_id)
 68        ```
 69
 70        Async context manager:
 71
 72        ```python
 73        async with SessionManager(client, "qpu.aria-1") as session:
 74            print(session.session_id)
 75        ```
 76    """
 77
 78    def __init__(
 79        self,
 80        client: AuthenticatedClient,
 81        backend: str,
 82        *,
 83        max_jobs: int | None = None,
 84        max_time: int | None = None,
 85        max_cost: float | None = None,
 86    ) -> None:
 87        self._client = client
 88        self._backend = backend
 89        self._max_jobs = max_jobs
 90        self._max_time = max_time
 91        self._max_cost = max_cost
 92        self._session_id: str | None = None
 93
 94    @classmethod
 95    def from_id(cls, client: AuthenticatedClient, session_id: str) -> SessionManager:
 96        """Reconnect to an existing session without creating a new one.
 97
 98        This is useful for resuming work with a session that was created
 99        in a previous process or by another client.
100
101        Args:
102            client: An authenticated API client.
103            session_id: The ID of the existing session.
104
105        Returns:
106            A `SessionManager` bound to the given session ID. The ``backend``
107            field will be empty since it is not needed for status checks
108            or ending the session.
109        """
110        mgr = cls(client, backend="")
111        mgr._session_id = session_id
112        return mgr
113
114    @property
115    def session_id(self) -> str | None:
116        """The session ID, or ``None`` if `open` has not been called."""
117        return self._session_id
118
119    def _build_settings(self) -> SessionSettingsRequest | Unset:
120        kw: dict = {}
121        if self._max_jobs is not None:
122            kw["job_count_limit"] = self._max_jobs
123        if self._max_time is not None:
124            kw["duration_limit_min"] = self._max_time
125        if self._max_cost is not None:
126            kw["cost_limit"] = SessionCostLimit(unit="usd", value=self._max_cost)
127        return SessionSettingsRequest(**kw) if kw else UNSET
128
129    def open(self) -> None:
130        """Create a new session on the configured backend.
131
132        Raises:
133            IonQError: If a session is already open or creation fails.
134        """
135        if self._session_id is not None:
136            raise IonQError("Session already open")
137        body = CreateSessionRequest(backend=self._backend, settings=self._build_settings())
138        session = create_session.sync(client=self._client, body=body)
139        if session is None:
140            raise IonQError("Failed to create session")
141        self._session_id = session.id
142        logger.info("Opened session %s", self._session_id)
143
144    def close(self) -> None:
145        """End the session. Suppresses exceptions so cleanup is safe."""
146        if self._session_id is None:
147            return
148        try:
149            end_session.sync(session_id=self._session_id, client=self._client)
150            logger.info("Closed session %s", self._session_id)
151        except Exception:
152            logger.warning("Failed to end session %s", self._session_id, exc_info=True)
153
154    def status(self) -> str:
155        """Get the current session status (e.g. ``"created"``, ``"started"``, ``"ended"``).
156
157        Raises:
158            IonQError: If no session is open or the status fetch fails.
159        """
160        if self._session_id is None:
161            raise IonQError("No session ID; call open() first")
162        session = get_session.sync(session_id=self._session_id, client=self._client)
163        if session is None:
164            raise IonQError(f"Failed to fetch session {self._session_id}")
165        return session.status
166
167    async def async_open(self) -> None:
168        """Async version of `open`."""
169        if self._session_id is not None:
170            raise IonQError("Session already open")
171        body = CreateSessionRequest(backend=self._backend, settings=self._build_settings())
172        session = await create_session.asyncio(client=self._client, body=body)
173        if session is None:
174            raise IonQError("Failed to create session")
175        self._session_id = session.id
176        logger.info("Opened session %s", self._session_id)
177
178    async def async_close(self) -> None:
179        """End the session (async). Suppresses exceptions so cleanup is safe."""
180        if self._session_id is None:
181            return
182        try:
183            await end_session.asyncio(session_id=self._session_id, client=self._client)
184            logger.info("Closed session %s", self._session_id)
185        except Exception:
186            logger.warning("Failed to end session %s", self._session_id, exc_info=True)
187
188    async def async_status(self) -> str:
189        """Async version of `status`."""
190        if self._session_id is None:
191            raise IonQError("No session ID; call open() first")
192        session = await get_session.asyncio(session_id=self._session_id, client=self._client)
193        if session is None:
194            raise IonQError(f"Failed to fetch session {self._session_id}")
195        return session.status
196
197    def __enter__(self) -> SessionManager:
198        self.open()
199        return self
200
201    def __exit__(self, *exc: object) -> None:
202        self.close()
203
204    async def __aenter__(self) -> SessionManager:
205        await self.async_open()
206        return self
207
208    async def __aexit__(self, *exc: object) -> None:
209        await self.async_close()