Examples for Downstream SDK Integration

CHANGELOG.md

@@ -9,7 +9,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 ### Added
 
 - `QctrlQaoaJobCreationPayload` and `QctrlQaoaJobInput` for submitting Q-CTRL QAOA maxcut combinatorial-optimization jobs via `create_job`. The `create_job` body union now also accepts `QctrlQaoaJobCreationPayload`.
-- `cost_model` optional field on `BaseJob`, `GetCircuitJobResponse`, and `GetJobResponse`, typed as `CostModel` (`"quantum_compute_time"` or `"execution_time"`).
+- `cost_model` optional field on `BaseJob`, `GetCircuitJobResponse`, and `GetJobResponse`, typed as `CostModel` (`"quantum_compute_time"`, `"execution_time"`, or `"2QGE_operations"`).
+- `examples/` directory with runnable sync and async downstream-SDK integration scripts demonstrating the extension API (`user_agent_token`, `default_headers`, event hooks, and `error_mapper`), runnable against the free `simulator`.
 
 ### Changed
 

README.md

@@ -65,6 +65,15 @@ Each generated endpoint module exposes four callables: `sync`, `sync_detailed`,
 
 For options (`api_key`, `base_url`, `max_retries`, `timeout`, `extension`), error classes, retry behavior, pagination, polling, sessions, and downstream-SDK extension hooks, see the [API reference](https://ionq.github.io/ionq-core-python/).
 
+## Examples
+
+Runnable, copy-pasteable scripts live in [`examples/`](examples/) and execute end-to-end against the free `simulator`:
+
+- [`downstream_integration.py`](examples/downstream_integration.py): downstream-SDK integration via the extension API (sync).
+- [`downstream_integration_async.py`](examples/downstream_integration_async.py): the same flow on the async path.
+
+See [`examples/README.md`](examples/README.md) for setup.
+
 ## Versioning
 
 This package follows [SemVer 2.0](https://semver.org/spec/v2.0.0.html), independent of the upstream REST API version - pass an explicit `base_url` to `IonQClient` to pin against a different API. Print the installed version with:

examples/README.md

@@ -0,0 +1,29 @@
+# Examples
+
+Runnable, copy-pasteable scripts that exercise `ionq-core` end to end against
+the free `simulator` backend.
+
+## Setup
+
+```sh
+source .venv/bin/activate # Run in project root directory
+pip install ionq-core     # To ensure latest version
+export IONQ_API_KEY=...   # Create a key at https://identity.ionq.com/create-account
+```
+
+## Scripts
+
+| Script | What it shows |
+| --- | --- |
+| [`downstream_integration.py`](downstream_integration.py) | Synchronous downstream-SDK integration via the [extension API](https://ionq.github.io/ionq-core-python/ionq_core/extensions.html): a `ClientExtension` with a `user_agent_token`, `default_headers`, an `EventHook`, and an `error_mapper`, then a Bell-state job submitted, polled, and read back. |
+| [`downstream_integration_async.py`](downstream_integration_async.py) | The same flow on the async path with an `AsyncEventHook` and the `asyncio` endpoint variants. |
+
+Run either with:
+
+```sh
+python examples/downstream_integration.py
+python examples/downstream_integration_async.py
+```
+
+Each prints the per-request hook log lines and the measured Bell-state
+probabilities.

examples/downstream_integration.py

@@ -0,0 +1,132 @@
+# SPDX-FileCopyrightText: 2026 IonQ, Inc.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Downstream-SDK integration example (synchronous).
+
+Shows how a higher-level SDK can wrap ``ionq-core`` through the extension API
+without modifying the library: a ``ClientExtension`` bundle carrying a
+user-agent token, default headers, a request/response ``EventHook``, and an
+``error_mapper`` that translates ``ionq-core`` exceptions into SDK-defined
+error types.
+
+The script submits a Bell-state circuit to the free ``simulator`` backend,
+waits for it to finish, and prints the measured probabilities. See
+``examples/README.md`` for setup (install, ``IONQ_API_KEY``) and how to run it.
+"""
+
+import logging
+
+import httpx
+
+from ionq_core import (
+    APIError,
+    AuthenticatedClient,
+    ClientExtension,
+    EventHook,
+    IonQClient,
+    RateLimitError,
+    wait_for_job,
+)
+from ionq_core.api.default import create_job, get_job, get_job_probabilities
+from ionq_core.models.circuit_job_creation_payload import CircuitJobCreationPayload
+
+logger = logging.getLogger("downstream_sdk")
+
+# --- Errors the downstream SDK exposes to its own users --------------------
+
+
+class DownstreamSDKError(Exception):
+    """Base error type raised by this example SDK."""
+
+
+class DownstreamRateLimitError(DownstreamSDKError):
+    """Raised when IonQ rate-limits a request."""
+
+    def __init__(self, message: str, *, retry_after: float | None = None) -> None:
+        super().__init__(message)
+        self.retry_after = retry_after
+
+
+def map_error(exc: Exception) -> Exception:
+    """Translate ``ionq-core`` exceptions into SDK-defined error types.
+
+    Passed to ``ionq-core`` via ``ClientExtension.error_mapper``. Returning the
+    original exception leaves it unchanged; returning a new one makes
+    ``ionq-core`` raise that instead (chained from the original).
+    """
+    if isinstance(exc, RateLimitError):
+        return DownstreamRateLimitError(f"IonQ rate limit hit: {exc.message}", retry_after=exc.retry_after)
+    if isinstance(exc, APIError):
+        return DownstreamSDKError(f"IonQ API error {exc.status_code}: {exc.message}")
+    return exc
+
+
+# --- Observability: log every request and response -------------------------
+
+
+class LoggingHook(EventHook):
+    """Logs each outgoing request and incoming response."""
+
+    def on_request(self, request: httpx.Request) -> None:
+        logger.info("--> %s %s", request.method, request.url)
+
+    def on_response(self, request: httpx.Request, response: httpx.Response) -> None:
+        logger.info("<-- %s %s", response.status_code, request.url)
+
+
+BELL_STATE = CircuitJobCreationPayload.from_dict(
+    {
+        "type": "ionq.circuit.v1",
+        "backend": "simulator",
+        "shots": 100,
+        "input": {
+            "gateset": "qis",
+            "qubits": 2,
+            "circuit": [
+                {"gate": "h", "target": 0},
+                {"gate": "cnot", "control": 0, "target": 1},
+            ],
+        },
+    }
+)
+
+
+def build_client() -> AuthenticatedClient:
+    """Construct an IonQ client configured the way a downstream SDK would."""
+    extension = ClientExtension(
+        user_agent_token="downstream-sdk/1.0.0",
+        default_headers={"X-Downstream-SDK": "example"},
+        event_hooks=(LoggingHook(),),
+        error_mapper=map_error,
+    )
+    return IonQClient(extension=extension)  # reads IONQ_API_KEY from the environment
+
+
+def main() -> None:
+    logging.basicConfig(level=logging.INFO, format="%(message)s")
+    with build_client() as client:
+        try:
+            job = create_job.sync(client=client, body=BELL_STATE)
+            if job is None:
+                raise DownstreamSDKError("create_job returned no response body")
+            completed = wait_for_job(client, job.id)
+            probabilities = get_job_probabilities.sync(uuid=job.id, client=client)
+            if probabilities is None:
+                raise DownstreamSDKError("get_job_probabilities returned no response body")
+            logger.info("job %s finished with status %r", job.id, completed.status)
+            print("Bell-state probabilities:", probabilities.additional_properties)
+        except DownstreamSDKError:
+            logger.exception("downstream SDK call failed")
+            raise
+
+        # The error_mapper also covers failures: requesting a job that does not
+        # exist returns 404, which ionq-core raises as NotFoundError (an APIError
+        # subclass) and map_error converts into a DownstreamSDKError.
+        try:
+            get_job.sync(uuid="00000000-0000-0000-0000-000000000000", client=client)
+        except DownstreamSDKError as exc:
+            logger.info("error_mapper converted a failure into: %s", exc)
+
+
+if __name__ == "__main__":
+    main()

examples/downstream_integration_async.py

@@ -0,0 +1,127 @@
+# SPDX-FileCopyrightText: 2026 IonQ, Inc.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Downstream-SDK integration example (asynchronous).
+
+The async counterpart of ``downstream_integration.py``: the same
+``ClientExtension`` integration (user-agent token, default headers, and a
+shared ``error_mapper``) but using an ``AsyncEventHook`` and the ``asyncio``
+endpoint variants on the async client path.
+
+The script submits a Bell-state circuit to the free ``simulator`` backend,
+waits for it to finish, and prints the measured probabilities. See
+``examples/README.md`` for setup (install, ``IONQ_API_KEY``) and how to run it.
+"""
+
+import asyncio
+import logging
+
+import httpx
+
+from ionq_core import (
+    APIError,
+    AsyncEventHook,
+    ClientExtension,
+    IonQClient,
+    RateLimitError,
+    async_wait_for_job,
+)
+from ionq_core.api.default import create_job, get_job, get_job_probabilities
+from ionq_core.models.circuit_job_creation_payload import CircuitJobCreationPayload
+
+logger = logging.getLogger("downstream_sdk")
+
+
+# --- Errors the downstream SDK exposes to its own users --------------------
+
+
+class DownstreamSDKError(Exception):
+    """Base error type raised by this example SDK."""
+
+
+class DownstreamRateLimitError(DownstreamSDKError):
+    """Raised when IonQ rate-limits a request."""
+
+    def __init__(self, message: str, *, retry_after: float | None = None) -> None:
+        super().__init__(message)
+        self.retry_after = retry_after
+
+
+def map_error(exc: Exception) -> Exception:
+    """Translate ``ionq-core`` exceptions into SDK-defined error types.
+
+    The mapper is synchronous on both client paths: ``ionq-core`` invokes it
+    synchronously even inside the async transport, so the same function serves
+    the sync and async examples.
+    """
+    if isinstance(exc, RateLimitError):
+        return DownstreamRateLimitError(f"IonQ rate limit hit: {exc.message}", retry_after=exc.retry_after)
+    if isinstance(exc, APIError):
+        return DownstreamSDKError(f"IonQ API error {exc.status_code}: {exc.message}")
+    return exc
+
+
+# --- Observability: log every request and response -------------------------
+
+
+class AsyncLoggingHook(AsyncEventHook):
+    """Logs each outgoing request and incoming response (async)."""
+
+    async def on_request(self, request: httpx.Request) -> None:
+        logger.info("--> %s %s", request.method, request.url)
+
+    async def on_response(self, request: httpx.Request, response: httpx.Response) -> None:
+        logger.info("<-- %s %s", response.status_code, request.url)
+
+
+BELL_STATE = CircuitJobCreationPayload.from_dict(
+    {
+        "type": "ionq.circuit.v1",
+        "backend": "simulator",
+        "shots": 100,
+        "input": {
+            "gateset": "qis",
+            "qubits": 2,
+            "circuit": [
+                {"gate": "h", "target": 0},
+                {"gate": "cnot", "control": 0, "target": 1},
+            ],
+        },
+    }
+)
+
+
+async def main() -> None:
+    logging.basicConfig(level=logging.INFO, format="%(message)s")
+    extension = ClientExtension(
+        user_agent_token="downstream-sdk/1.0.0",
+        default_headers={"X-Downstream-SDK": "example"},
+        async_event_hooks=(AsyncLoggingHook(),),
+        error_mapper=map_error,
+    )
+    async with IonQClient(extension=extension) as client:  # reads IONQ_API_KEY from the environment
+        try:
+            job = await create_job.asyncio(client=client, body=BELL_STATE)
+            if job is None:
+                raise DownstreamSDKError("create_job returned no response body")
+            completed = await async_wait_for_job(client, job.id)
+            probabilities = await get_job_probabilities.asyncio(uuid=job.id, client=client)
+            if probabilities is None:
+                raise DownstreamSDKError("get_job_probabilities returned no response body")
+            logger.info("job %s finished with status %r", job.id, completed.status)
+            print("Bell-state probabilities:", probabilities.additional_properties)
+        except DownstreamSDKError:
+            logger.exception("downstream SDK call failed")
+            raise
+
+        # The error_mapper also covers failures: requesting a job that does not
+        # exist returns 404, which ionq-core raises as NotFoundError (an APIError
+        # subclass) and map_error converts into a DownstreamSDKError.
+        try:
+            await get_job.asyncio(uuid="00000000-0000-0000-0000-000000000000", client=client)
+        except DownstreamSDKError as exc:
+            logger.info("error_mapper converted a failure into: %s", exc)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

openapi-overlay.yaml

@@ -1,7 +1,7 @@
 overlay: 1.0.0
 info:
   title: ionq-core-python local OpenAPI fixes
-  version: 0.2.0
+  version: 0.3.0
   description: |
     Patches applied to openapi.json before client generation. The upstream
     spec marks QisCircuitInput.qubits as optional, but the simulator
@@ -13,6 +13,12 @@ info:
     (and extended it to NativeCircuitInput and JsonMultiCircuitInput), so
     that action was removed.
 
+    The API also returns a CostModel value ("2QGE_operations") that the
+    upstream spec's enum does not list (it documents only
+    quantum_compute_time and execution_time), so deserializing a completed
+    job's cost_model raises a TypeError. We extend the enum locally until
+    upstream documents the value; remove this action once it ships upstream.
+
 actions:
   - target: $.components.schemas.QisCircuitInput.required
     remove: true
@@ -22,3 +28,11 @@ actions:
         - circuit
         - gateset
         - qubits
+  - target: $.components.schemas.CostModel.enum
+    remove: true
+  - target: $.components.schemas.CostModel
+    update:
+      enum:
+        - quantum_compute_time
+        - execution_time
+        - "2QGE_operations"