api_client: implement workspaces() query

docs/apidoc/api_client.rst

@@ -0,0 +1,25 @@
+.. _api_client:
+
+==========
+API client
+==========
+
+.. autoclass:: qiskit_aqt_provider.api_client.portal_client.PortalClient
+   :members:
+   :show-inheritance:
+   :inherited-members:
+
+.. autoclass:: qiskit_aqt_provider.api_client.models.Workspaces
+   :members:
+   :show-inheritance:
+   :exclude-members: __init__, __new__, model_fields, model_computed_fields, model_config
+
+.. autoclass:: qiskit_aqt_provider.api_client.models.Workspace
+   :members:
+   :show-inheritance:
+   :exclude-members: __init__, __new__, model_fields, model_computed_fields, model_config
+
+.. autoclass:: qiskit_aqt_provider.api_client.models.Resource
+   :members:
+   :show-inheritance:
+   :exclude-members: __init__, __new__, model_fields, model_computed_fields, model_config

docs/conf.py

@@ -57,6 +57,10 @@
     ("py:class", "Target"),
     ("py:exc", "QiskitBackendNotFoundError"),
     ("py:class", "qiskit_aqt_provider.aqt_resource._OptionsType"),
+    # No inventory available for httpx
+    # https://github.com/encode/httpx/issues/3145
+    ("py:exc", "httpx.NetworkError"),
+    ("py:exc", "httpx.HTTPStatusError"),
 ]
 nitpick_ignore_regex = [
     ("py:class", r"qiskit_aqt_provider\.api_models_generated.*"),
@@ -83,4 +87,5 @@
 intersphinx_mapping = {
     "python": ("https://docs.python.org/3", None),
     "qiskit": ("https://docs.quantum.ibm.com/api/qiskit/", None),
+    "pydantic": ("https://docs.pydantic.dev/latest/", None),
 }

docs/index.rst

@@ -72,6 +72,7 @@ For more details see the :ref:`user guide <user-guide>`, a selection of `example
   Options <apidoc/options>
   Qiskit primitives <apidoc/primitives>
   Transpiler plugin <apidoc/transpiler_plugin>
+  API client <apidoc/api_client>
 
 .. toctree::
   :hidden:

qiskit_aqt_provider/api_client/__init__.py

@@ -7,3 +7,17 @@
 # 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.
+
+from .models import Resource, ResourceType, Workspace, Workspaces
+from .portal_client import DEFAULT_PORTAL_URL, PortalClient
+from .versions import __version__
+
+__all__ = [
+    "DEFAULT_PORTAL_URL",
+    "PortalClient",
+    "Resource",
+    "ResourceType",
+    "Workspace",
+    "Workspaces",
+    "__version__",
+]

qiskit_aqt_provider/api_client/models.py

@@ -15,14 +15,16 @@
 import importlib.metadata
 import platform
 import re
+import typing
+from collections.abc import Collection, Iterator
 from re import Pattern
 from typing import Any, Final, Literal, Optional, Union
 from uuid import UUID
 
 import httpx
 import pydantic as pdt
 from qiskit.providers.exceptions import JobError
-from typing_extensions import Self, TypeAlias
+from typing_extensions import Self, TypeAlias, override
 
 from . import models_generated as api_models
 from .models_generated import (
@@ -32,7 +34,6 @@
     QuantumCircuits,
     SubmitJobRequest,
 )
-from .models_generated import Type as ResourceType
 
 __all__ = [
     "Circuit",
@@ -43,7 +44,6 @@
     "QuantumCircuit",
     "QuantumCircuits",
     "Response",
-    "ResourceType",
 ]
 
 
@@ -78,11 +78,153 @@ def http_client(
     return httpx.Client(headers=headers, base_url=base_url, timeout=10.0, follow_redirects=True)
 
 
-class Workspaces(pdt.RootModel[list[api_models.Workspace]]):
-    """List of available workspaces and devices."""
+ResourceType: TypeAlias = Literal["device", "simulator", "offline_simulator"]
+
+
+class Resource(pdt.BaseModel):
+    """Description of a resource.
+
+    This is the element type in :py:attr:`Workspace.resources`.
+    """
+
+    model_config = pdt.ConfigDict(frozen=True)
+
+    workspace_id: str
+    """Identifier of the workspace this resource belongs to."""
+
+    resource_id: str
+    """Resource identifier."""
+
+    resource_name: str
+    """Resource name."""
+
+    resource_type: ResourceType
+    """Type of resource."""
+
+
+class Workspace(pdt.BaseModel):
+    """Description of a workspace and the resources it contains.
+
+    This is the element type in the :py:class:`Workspaces` container.
+    """
+
+    model_config = pdt.ConfigDict(frozen=True)
+
+    workspace_id: str
+    """Workspace identifier."""
+
+    resources: list[Resource]
+    """Resources in the workspace."""
+
+
+class Workspaces(
+    pdt.RootModel,  # type: ignore[type-arg]
+    Collection[Workspace],
+):
+    """List of available workspaces and devices.
+
+    ..
+        >>> workspaces = Workspaces(
+        ...     root=[
+        ...         api_models.Workspace(
+        ...             id="workspace0",
+        ...             resources=[
+        ...                 api_models.Resource(
+        ...                     id="resource0",
+        ...                     name="resource0",
+        ...                     type=api_models.Type.device,
+        ...                 ),
+        ...             ],
+        ...         ),
+        ...        api_models.Workspace(
+        ...            id="workspace1",
+        ...            resources=[
+        ...                api_models.Resource(
+        ...                    id="resource0",
+        ...                    name="resource0",
+        ...                    type=api_models.Type.device,
+        ...                ),
+        ...                api_models.Resource(
+        ...                    id="resource1",
+        ...                    name="resource1",
+        ...                    type=api_models.Type.simulator,
+        ...                ),
+        ...            ],
+        ...        ),
+        ...    ]
+        ... )
+
+    Examples:
+        Assume a :py:class:`Workspaces` instance retrieved from the API with
+        the following contents:
+
+        .. code-block::
+
+            | Workspace ID | Resource ID | Resource Type |
+            |--------------+-------------+---------------|
+            | workspace0   | resource0   | device        |
+            | workspace1   | resource0   | device        |
+            | workspace1   | resource1   | simulator     |
+
+        Gather basic information:
+
+        >>> # workspaces = PortalClient(...).workspaces()
+        >>> len(workspaces)
+        2
+        >>> [ws.workspace_id for ws in workspaces]
+        ['workspace0', 'workspace1']
+
+        Inclusion tests rely only on the identifier:
+
+        >>> Workspace(workspace_id="workspace0", resources=[]) in workspaces
+        True
+
+        The :py:meth:`Workspaces.filter` method allows for complex filtering. For example
+        by workspace identifier ending in ``0``:
+
+        >>> [ws.workspace_id for ws in workspaces.filter(workspace_pattern=re.compile(".+0$"))]
+        ['workspace0']
+
+        or only the non-simulated devices:
+
+        >>> workspaces_devices = workspaces.filter(backend_type="device")
+        >>> [(ws.workspace_id, resource.resource_id)
+        ...  for ws in workspaces_devices for resource in ws.resources]
+        [('workspace0', 'resource0'), ('workspace1', 'resource0')]
+    """
 
     root: list[api_models.Workspace]
 
+    @override
+    def __len__(self) -> int:
+        """Number of available workspaces."""
+        return len(self.root)
+
+    @override
+    def __iter__(self) -> Iterator[Workspace]:  # type: ignore[override]
+        """Iterator over the workspaces."""
+        for ws in self.root:
+            yield Workspace(
+                workspace_id=ws.id,
+                resources=[
+                    Resource(
+                        workspace_id=ws.id,
+                        resource_id=res.id,
+                        resource_name=res.name,
+                        resource_type=typing.cast(ResourceType, res.type.value),
+                    )
+                    for res in ws.resources
+                ],
+            )
+
+    @override
+    def __contains__(self, obj: object) -> bool:
+        """Whether a given workspace is in this workspaces collection."""
+        if not isinstance(obj, Workspace):  # pragma: no cover
+            return False
+
+        return any(ws.id == obj.workspace_id for ws in self.root)
+
     def filter(
         self,
         *,
@@ -100,7 +242,7 @@ def filter(
             backend_type: backend type to select.
 
         Returns:
-            Workspaces model that only contains matching resources.
+            :py:class:`Workspaces` instance that only contains matching resources.
         """
         filtered_workspaces = []
         for workspace in self.root:
@@ -110,7 +252,7 @@ def filter(
             filtered_resources = []
 
             for resource in workspace.resources:
-                if backend_type is not None and resource.type is not backend_type:
+                if backend_type is not None and resource.type.value != backend_type:
                     continue
 
                 if name_pattern is not None and not re.match(name_pattern, resource.id):
@@ -125,27 +267,6 @@ def filter(
         return self.__class__(root=filtered_workspaces)
 
 
-GeneralResourceType: TypeAlias = Literal["device", "simulator", "offline_simulator"]
-
-
-class ResourceId(pdt.BaseModel):
-    """Resource identification and metadata."""
-
-    model_config = pdt.ConfigDict(frozen=True)
-
-    workspace_id: str
-    """Workspace containing the resource."""
-
-    resource_id: str
-    """Unique identifier of the resource in the containing workspace."""
-
-    resource_name: str
-    """Pretty display name for the resource."""
-
-    resource_type: GeneralResourceType
-    """Resource type, also includes offline simulators."""
-
-
 class Operation:
     """Factories for API payloads of circuit operations."""
 

qiskit_aqt_provider/api_client/portal_client.py

@@ -0,0 +1,67 @@
+# (C) Copyright Alpine Quantum Technologies GmbH 2024
+#
+# 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.
+
+import os
+from typing import Final, Optional
+
+import httpx
+
+from . import models
+from .versions import make_user_agent
+
+DEFAULT_PORTAL_URL: Final = httpx.URL("https://arnica.aqt.eu")
+"""Default URL for the remote portal."""
+
+
+class PortalClient:
+    """Client for the AQT portal API."""
+
+    USER_AGENT_NAME: Final = "aqt-portal-client"
+
+    def __init__(
+        self, *, token: str, user_agent_extra: Optional[str] = None, timeout: Optional[float] = 10.0
+    ) -> None:
+        """Initialize a new client for the AQT remote computing portal API.
+
+        The portal base URL can be overridden using the ``AQT_PORTAL_URL``
+        environment variable.
+
+        Args:
+            token: authentication token.
+            user_agent_extra: data appended to the default user-agent string.
+            timeout: HTTP timeout, in seconds.
+        """
+        self.portal_url = httpx.URL(os.environ.get("AQT_PORTAL_URL", DEFAULT_PORTAL_URL))
+
+        user_agent = make_user_agent(self.USER_AGENT_NAME, extra=user_agent_extra)
+        headers = {"User-Agent": user_agent}
+
+        if token:
+            headers["Authorization"] = f"Bearer {token}"
+
+        self._http_client = httpx.Client(
+            base_url=self.portal_url.join("/api/v1"),
+            headers=headers,
+            timeout=timeout,
+            follow_redirects=True,
+        )
+
+    def workspaces(self) -> models.Workspaces:
+        """List the workspaces visible to the used token.
+
+        Raises:
+            httpx.NetworkError: connection to the remote portal failed.
+            httpx.HTTPStatusError: something went wrong with the request to the remote portal.
+        """
+        with self._http_client as client:
+            response = client.get("/workspaces")
+
+        response.raise_for_status()
+        return models.Workspaces.model_validate(response.json())