New: support engagement against a specified quantum processor endpoint

CHANGELOG.md

@@ -8,6 +8,9 @@ Changelog
 
 ### Improvements and Changes
 
+- Both `get_qc` and `QPU` now accept an `endpoint_id` argument which is used to engage
+  against a specific QCS [quantum processor endpoint](https://docs.api.qcs.rigetti.com/#tag/endpoints).
+
 ### Bugfixes
 
 - Allow `np.ndarray` when writing QAM memory. Disallow non-integer and non-float types.

docs/source/advanced_usage.rst

@@ -82,6 +82,30 @@ Below is an example that demonstrates how to use pyQuil in a multithreading scen
         print(f"Results for program {i}:\n{result}\n")
 
 
+Alternative QPU Endpoints
+~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Rigetti QCS supports alternative endpoints for access to a QPU architecture, useful for very particular cases.
+Generally, this is useful to call "mock" or test endpoints, which simulate the results of execution for the
+purposes of integration testing without the need for an active reservation or contention with other users.
+See the `QCS API Docs <https://docs.api.qcs.rigetti.com/#tag/endpoints>`_ for more information on QPU Endpoints.
+
+To be able to call these endpoints using pyQuil, enter the ``endpoint_id`` of your desired endpoint in one
+of the sites where ``quantum_processor_id`` is used:
+
+.. code:: python
+
+    # Option 1
+    qc = get_qc("Aspen-9", endpoint_id="my_endpoint")
+
+    # Option 2
+    qam = QPU("Aspen-9", endpoint_id="my_endpoint")
+
+After doing so, for all intents and purposes - compilation, optimization, etc - your program will behave the same
+as when using "default" endpoint for a given quantum processor, except that it will be executed by an
+alternate QCS service, and the results of execution should not be treated as correct or meaningful.
+
+
 Using Qubit Placeholders
 ~~~~~~~~~~~~~~~~~~~~~~~~
 

pyproject.toml

@@ -45,13 +45,13 @@ black = "^20.8b1"
 flake8 = "^3.8.1"
 pytest = "^6.2.2"
 pytest-cov = "^2.11.1"
-pytest-httpx = "^0.9"
 mypy = "0.740"
 pytest-xdist = "^2.2.1"
 pytest-rerunfailures = "^9.1.1"
 pytest-timeout = "^1.4.2"
 pytest-mock = "^3.6.1"
 pytest-freezegun = "^0.4.2"
+respx = "^0.15"
 
 [tool.poetry.extras]
 latex = ["ipython"]

pyquil/api/_engagement_manager.py

@@ -15,20 +15,26 @@
 ##############################################################################
 import threading
 from datetime import datetime
-from typing import Dict, Optional, TYPE_CHECKING
+from typing import Dict, NamedTuple, Optional, TYPE_CHECKING
 
 from dateutil.parser import parse as parsedate
 from dateutil.tz import tzutc
 from qcs_api_client.client import QCSClientConfiguration
 from qcs_api_client.models import EngagementWithCredentials, CreateEngagementRequest
 from qcs_api_client.operations.sync import create_engagement
+from qcs_api_client.types import UNSET
 
 from pyquil.api._qcs_client import qcs_client
 
 if TYPE_CHECKING:
     import httpx
 
 
+class EngagementCacheKey(NamedTuple):
+    quantum_processor_id: str
+    endpoint_id: Optional[str]
+
+
 class EngagementManager:
     """
     Fetches (and caches) engagements for use when accessing a QPU.
@@ -44,28 +50,36 @@ def __init__(self, *, client_configuration: QCSClientConfiguration) -> None:
         :param client_configuration: Client configuration, used for refreshing engagements.
         """
         self._client_configuration = client_configuration
-        self._cached_engagements: Dict[str, EngagementWithCredentials] = {}
+        self._cached_engagements: Dict[EngagementCacheKey, EngagementWithCredentials] = {}
         self._lock = threading.Lock()
 
-    def get_engagement(self, *, quantum_processor_id: str, request_timeout: float = 10.0) -> EngagementWithCredentials:
+    def get_engagement(
+        self, *, quantum_processor_id: str, request_timeout: float = 10.0, endpoint_id: Optional[str] = None
+    ) -> EngagementWithCredentials:
         """
-        Gets an engagement for the given quantum processor. If an engagement was already fetched previously and
-        remains valid, it will be returned instead of creating a new engagement.
+        Gets an engagement for the given quantum processor endpoint.
+
+        If an engagement was already fetched previously and remains valid, it will be returned instead
+        of creating a new engagement.
 
         :param quantum_processor_id: Quantum processor being engaged.
         :param request_timeout: Timeout for request, in seconds.
+        :param endpoint_id: Optional ID of the endpoint to use for engagement. If provided, it must
+            correspond to an endpoint serving the provided Quantum Processor.
         :return: Fetched or cached engagement.
         """
+        key = EngagementCacheKey(quantum_processor_id, endpoint_id)
+
         with self._lock:
-            if not self._engagement_valid(self._cached_engagements.get(quantum_processor_id)):
+            if not self._engagement_valid(self._cached_engagements.get(key)):
                 with qcs_client(
                     client_configuration=self._client_configuration, request_timeout=request_timeout
                 ) as client:  # type: httpx.Client
-                    request = CreateEngagementRequest(quantum_processor_id=quantum_processor_id)
-                    self._cached_engagements[quantum_processor_id] = create_engagement(
-                        client=client, json_body=request
-                    ).parsed
-            return self._cached_engagements[quantum_processor_id]
+                    request = CreateEngagementRequest(
+                        quantum_processor_id=quantum_processor_id, endpoint_id=endpoint_id or UNSET
+                    )
+                    self._cached_engagements[key] = create_engagement(client=client, json_body=request).parsed
+            return self._cached_engagements[key]
 
     @staticmethod
     def _engagement_valid(engagement: Optional[EngagementWithCredentials]) -> bool:

pyquil/api/_qpu.py

@@ -115,6 +115,7 @@ def __init__(
         timeout: float = 10.0,
         client_configuration: Optional[QCSClientConfiguration] = None,
         engagement_manager: Optional[EngagementManager] = None,
+        endpoint_id: Optional[str] = None,
     ) -> None:
         """
         A connection to the QPU.
@@ -123,10 +124,9 @@ def __init__(
         :param priority: The priority with which to insert jobs into the QPU queue. Lower integers
             correspond to higher priority.
         :param timeout: Time limit for requests, in seconds.
-        :param client_configuration: Optional client configuration. If none is provided, a default
-            one will be loaded.
-        :param engagement_manager: Optional engagement manager. If none is provided, a default one
-            will be created.
+        :param client_configuration: Optional client configuration. If none is provided, a default one will be loaded.
+        :param endpoint_id: Optional endpoint ID to be used for engagement.
+        :param engagement_manager: Optional engagement manager. If none is provided, a default one will be created.
         """
         super().__init__()
 
@@ -136,6 +136,7 @@ def __init__(
         engagement_manager = engagement_manager or EngagementManager(client_configuration=client_configuration)
         self._qpu_client = QPUClient(
             quantum_processor_id=quantum_processor_id,
+            endpoint_id=endpoint_id,
             engagement_manager=engagement_manager,
             request_timeout=timeout,
         )

pyquil/api/_qpu_client.py

@@ -15,7 +15,7 @@
 ##############################################################################
 from dataclasses import dataclass
 from datetime import datetime
-from typing import Dict, cast, Tuple, Union, List, Any
+from typing import Dict, Optional, cast, Tuple, Union, List, Any
 
 import rpcq
 from dateutil.parser import parse as parsedate
@@ -104,6 +104,7 @@ def __init__(
         *,
         quantum_processor_id: str,
         engagement_manager: EngagementManager,
+        endpoint_id: Optional[str] = None,
         request_timeout: float = 10.0,
     ) -> None:
         """
@@ -114,6 +115,7 @@ def __init__(
         :param request_timeout: Timeout for requests, in seconds.
         """
         self.quantum_processor_id = quantum_processor_id
+        self._endpoint_id = endpoint_id
         self._engagement_manager = engagement_manager
         self.timeout = request_timeout
 
@@ -157,6 +159,7 @@ def get_buffers(self, request: GetBuffersRequest) -> GetBuffersResponse:
     @retry(exceptions=TimeoutError, tries=2)  # type: ignore
     def _rpcq_request(self, method_name: str, *args: Any, **kwargs: Any) -> Any:
         engagement = self._engagement_manager.get_engagement(
+            endpoint_id=self._endpoint_id,
             quantum_processor_id=self.quantum_processor_id,
             request_timeout=self.timeout,
         )

pyquil/api/_quantum_computer.py

@@ -725,6 +725,7 @@ def get_qc(
     compiler_timeout: float = 10.0,
     execution_timeout: float = 10.0,
     client_configuration: Optional[QCSClientConfiguration] = None,
+    endpoint_id: Optional[str] = None,
     engagement_manager: Optional[EngagementManager] = None,
 ) -> QuantumComputer:
     """
@@ -794,9 +795,12 @@ def get_qc(
     :param compiler_timeout: Time limit for compilation requests, in seconds.
     :param execution_timeout: Time limit for execution requests, in seconds.
     :param client_configuration: Optional client configuration. If none is provided, a default one will be loaded.
+    :param endpoint_id: Optional quantum processor endpoint ID, as used in the `QCS API Docs`_.
     :param engagement_manager: Optional engagement manager. If none is provided, a default one will be created.
 
     :return: A pre-configured QuantumComputer
+
+    .. _QCS API Docs: https://docs.api.qcs.rigetti.com/#tag/endpoints
     """
 
     client_configuration = client_configuration or QCSClientConfiguration.load()
@@ -862,6 +866,7 @@ def get_qc(
             quantum_processor_id=quantum_processor.quantum_processor_id,
             timeout=execution_timeout,
             client_configuration=client_configuration,
+            endpoint_id=endpoint_id,
             engagement_manager=engagement_manager,
         )
         compiler = QPUCompiler(