add ruff

.github/workflows/docs/conf.py

@@ -1,8 +1,17 @@
-# -*- coding: utf-8 -*-
-
 # Configuration file for the Sphinx documentation builder.
 # See https://www.sphinx-doc.org/en/master/usage/configuration.html
 
+# The following code is for resolving broken hyperlinks in the doc.
+
+import re
+from typing import Any, Optional
+from urllib.parse import urljoin
+
+from docutils import nodes
+from docutils.nodes import Element, TextElement
+from sphinx.application import Sphinx
+from sphinx.environment import BuildEnvironment
+
 copyright = "2024 Quantinuum"
 author = "Quantinuum"
 
@@ -47,16 +56,6 @@
 
 autodoc_member_order = "groupwise"
 
-# The following code is for resolving broken hyperlinks in the doc.
-
-import re
-from typing import Any, Dict, List, Optional
-from urllib.parse import urljoin
-
-from docutils import nodes
-from docutils.nodes import Element, TextElement
-from sphinx.application import Sphinx
-from sphinx.environment import BuildEnvironment
 
 # Mappings for broken hyperlinks that intersphinx cannot resolve
 external_url_mapping = {
@@ -95,7 +94,8 @@
 def add_reference(
     app: Sphinx, env: BuildEnvironment, node: Element, contnode: TextElement
 ) -> Optional[nodes.reference]:
-    # Fix references in docstrings that are inherited from the base pytket.backends.Backend class.
+    # Fix references in docstrings that are inherited from
+    # the base pytket.backends.Backend class.
     mapping = app.config.external_url_mapping
     if node.astext() in mapping:
         newnode = nodes.reference(
@@ -115,7 +115,7 @@ def correct_signature(
     what: str,
     name: str,
     obj: Any,
-    options: Dict,
+    options: dict,
     signature: str,
     return_annotation: str,
 ) -> (str, str):
@@ -126,7 +126,8 @@ def correct_signature(
             new_signature = new_signature.replace(k, v)
         if return_annotation is not None:
             new_return_annotation = new_return_annotation.replace(k, v)
-    # e.g. Replace <CXConfigType.Snake: 0> by CXConfigType.Snake to avoid silent failure in later stages.
+    # e.g. Replace <CXConfigType.Snake: 0> by CXConfigType.Snake to
+    # avoid silent failure in later stages.
     if new_signature is not None:
         enums_signature = re.findall(r"<.+?\: \d+>", new_signature)
         for e in enums_signature:

.github/workflows/lint.yml

@@ -19,10 +19,13 @@ jobs:
     - name: Update pip
       run: pip install --upgrade pip
     - name: Install black and pylint
-      run: pip install "black[jupyter]" pylint
+      run: pip install "black[jupyter]" pylint ruff
     - name: Check files are formatted with black
       run: |
         black --check .
+    - name: Run ruff
+      run: |
+        ruff check .        
     - name: Run pylint
       run: |        
         pylint */

pytket/extensions/quantinuum/__init__.py

@@ -16,13 +16,30 @@
 """
 
 # _metadata.py is copied to the folder after installation.
-from ._metadata import __extension_version__, __extension_name__
+from ._metadata import (
+    __extension_name__ as __extension_name__,
+)
+from ._metadata import (
+    __extension_version__ as __extension_version__,
+)
+from .backends import (
+    Language as Language,
+)
+from .backends import (
+    QuantinuumAPI as QuantinuumAPI,
+)
+from .backends import (
+    QuantinuumAPIOffline as QuantinuumAPIOffline,
+)
+from .backends import (
+    QuantinuumBackend as QuantinuumBackend,
+)
+from .backends import (
+    QuantinuumBackendCompilationConfig as QuantinuumBackendCompilationConfig,
+)
+from .backends import (
+    have_pecos as have_pecos,
+)
 from .backends import (
-    QuantinuumBackend,
-    QuantinuumAPI,
-    QuantinuumAPIOffline,
-    QuantinuumBackendCompilationConfig,
-    Language,
-    prune_shots_detected_as_leaky,
-    have_pecos,
+    prune_shots_detected_as_leaky as prune_shots_detected_as_leaky,
 )

pytket/extensions/quantinuum/backends/__init__.py

@@ -15,11 +15,24 @@
 """Backends for processing pytket circuits with Quantinuum devices
 """
 
+from .api_wrappers import (
+    QuantinuumAPI as QuantinuumAPI,
+)
+from .api_wrappers import (
+    QuantinuumAPIOffline as QuantinuumAPIOffline,
+)
+from .leakage_gadget import (
+    prune_shots_detected_as_leaky as prune_shots_detected_as_leaky,
+)
+from .quantinuum import (
+    Language as Language,
+)
+from .quantinuum import (
+    QuantinuumBackend as QuantinuumBackend,
+)
+from .quantinuum import (
+    QuantinuumBackendCompilationConfig as QuantinuumBackendCompilationConfig,
+)
 from .quantinuum import (
-    QuantinuumBackend,
-    QuantinuumBackendCompilationConfig,
-    Language,
-    have_pecos,
+    have_pecos as have_pecos,
 )
-from .api_wrappers import QuantinuumAPI, QuantinuumAPIOffline
-from .leakage_gadget import prune_shots_detected_as_leaky

pytket/extensions/quantinuum/backends/api_wrappers.py

@@ -16,27 +16,27 @@
 Functions used to submit jobs with Quantinuum API.
 """
 
-import time
-from http import HTTPStatus
-from typing import Optional, Dict, Tuple, List
 import asyncio
-import json
+import contextlib
 import getpass
+import json
+import time
+from http import HTTPStatus
+from typing import Optional
+
+import nest_asyncio  # type: ignore
 from requests import Session
 from requests.models import Response
 from websockets import connect, exceptions
-import nest_asyncio  # type: ignore
 
 from .config import QuantinuumConfig
 from .credential_storage import CredentialStorage, MemoryCredentialStorage
 from .federated_login import microsoft_login
 
 # This is necessary for use in Jupyter notebooks to allow for nested asyncio loops
-try:
+# May fail in some cloud environments: ignore.
+with contextlib.suppress(RuntimeError, ValueError):
     nest_asyncio.apply()
-except (RuntimeError, ValueError):
-    # May fail in some cloud environments: ignore.
-    pass
 
 
 class QuantinuumAPIError(Exception):
@@ -199,7 +199,7 @@ def _request_tokens_federated(self) -> None:
             _, token = microsoft_login()
         else:
             raise RuntimeError(
-                f"Unsupported provider for login", HTTPStatus.UNAUTHORIZED
+                "Unsupported provider for login", HTTPStatus.UNAUTHORIZED
             )
 
         body = {"provider-token": token}
@@ -247,7 +247,7 @@ def _refresh_id_token(self, refresh_token: str) -> None:
             del refresh_token
             del body
 
-    def _get_credentials(self) -> Tuple[str, str]:
+    def _get_credentials(self) -> tuple[str, str]:
         """Method to ask for user's credentials"""
         user_name = self._cred_store.user_name
         pwd = None
@@ -302,7 +302,7 @@ def delete_authentication(self) -> None:
         """Remove stored credentials and tokens"""
         self._cred_store.delete_credential()
 
-    def _submit_job(self, body: Dict) -> Response:
+    def _submit_job(self, body: dict) -> Response:
         id_token = self.login()
         # send job request
         return self.session.post(
@@ -317,10 +317,8 @@ def _response_check(self, res: Response, description: str) -> None:
         if res.status_code == HTTPStatus.UNAUTHORIZED:
             jr = res.json()
             raise QuantinuumAPIError(
-                (
-                    f"Authorization failure attempting: {description}."
-                    f"\n\nServer Response: {jr}"
-                )
+                f"Authorization failure attempting: {description}."
+                f"\n\nServer Response: {jr}"
             )
         elif res.status_code != HTTPStatus.OK:
             jr = res.json()
@@ -330,7 +328,7 @@ def _response_check(self, res: Response, description: str) -> None:
 
     def retrieve_job_status(
         self, job_id: str, use_websocket: Optional[bool] = None
-    ) -> Optional[Dict]:
+    ) -> Optional[dict]:
         """
         Retrieves job status from device.
 
@@ -347,7 +345,7 @@ def retrieve_job_status(
             job_url += "?websocket=true"
         res = self.session.get(job_url, headers={"Authorization": id_token})
 
-        jr: Optional[Dict] = None
+        jr: Optional[dict] = None
         # Check for invalid responses, and raise an exception if so
         self._response_check(res, "job status")
         # if we successfully got status return the decoded details
@@ -357,7 +355,7 @@ def retrieve_job_status(
 
     def retrieve_job(
         self, job_id: str, use_websocket: Optional[bool] = None
-    ) -> Optional[Dict]:
+    ) -> Optional[dict]:
         """
         Retrieves job from device.
 
@@ -387,7 +385,7 @@ def retrieve_job(
             jr = self._poll_results(job_id)
         return jr
 
-    def _poll_results(self, job_id: str) -> Optional[Dict]:
+    def _poll_results(self, job_id: str) -> Optional[dict]:
         jr = None
         start_time = time.time()
         while True:
@@ -407,16 +405,14 @@ def _poll_results(self, job_id: str) -> Optional[Dict]:
                 raise RuntimeError("Keyboard Interrupted")
         return jr
 
-    async def _wait_results(self, job_id: str) -> Optional[Dict]:
+    async def _wait_results(self, job_id: str) -> Optional[dict]:
         start_time = time.time()
         while True:
             if self.timeout is not None and time.time() > (start_time + self.timeout):
                 break
             self.login()
             jr = self.retrieve_job_status(job_id, True)
-            if jr is None:
-                return jr
-            elif "status" in jr and jr["status"] in self.JOB_DONE:
+            if jr is None or "status" in jr and jr["status"] in self.JOB_DONE:
                 return jr
             else:
                 task_token = jr["websocket"]["task_token"]
@@ -436,7 +432,7 @@ async def _wait_results(self, job_id: str) -> Optional[Dict]:
                                 websocket.recv(), timeout=self.ws_timeout
                             )
                             jr = json.loads(res)
-                            if not isinstance(jr, Dict):
+                            if not isinstance(jr, dict):
                                 raise RuntimeError("Unable to decode response.")
                             if "status" in jr and jr["status"] in self.JOB_DONE:
                                 return jr
@@ -495,7 +491,7 @@ def cancel(self, job_id: str) -> dict:
 
         return jr  # type: ignore
 
-    def get_calendar(self, start_date: str, end_date: str) -> List[Dict[str, str]]:
+    def get_calendar(self, start_date: str, end_date: str) -> list[dict[str, str]]:
         """
         Retrieves calendar data using L4 API. All dates and times
         are in the UTC timezone.
@@ -514,7 +510,7 @@ def get_calendar(self, start_date: str, end_date: str) -> List[Dict[str, str]]:
             headers={"Authorization": id_token},
         )
         self._response_check(res, "get calendar events")
-        jr: List[Dict[str, str]] = res.json()
+        jr: list[dict[str, str]] = res.json()
         return jr
 
 
@@ -611,7 +607,7 @@ def __init__(self, machine_list: Optional[list] = None):
             "batching": True,
             }
         """
-        if machine_list == None:
+        if machine_list is None:
             machine_list = OFFLINE_MACHINE_LIST
         self.provider = ""
         self.url = ""
@@ -637,7 +633,7 @@ def login(self) -> str:
         return an empty api token"""
         return ""
 
-    def _submit_job(self, body: Dict) -> None:
+    def _submit_job(self, body: dict) -> None:
         """The function will take the submitted job and store it for later
 
         :param body: submitted job
@@ -659,30 +655,24 @@ def _response_check(self, res: Response, description: str) -> None:
 
         jr = res.json()
         raise QuantinuumAPIError(
-            (
-                f"Reponse can't be checked offline: {description}."
-                f"\n\nServer Response: {jr}"
-            )
+            f"Reponse can't be checked offline: {description}."
+            f"\n\nServer Response: {jr}"
         )
 
     def retrieve_job_status(
         self, job_id: str, use_websocket: Optional[bool] = None
     ) -> None:
         """No retrieve_job_status offline"""
         raise QuantinuumAPIError(
-            (
-                f"Can't retrieve job status offline: job_id {job_id}."
-                f"\n use_websocket {use_websocket}"
-            )
+            f"Can't retrieve job status offline: job_id {job_id}."
+            f"\n use_websocket {use_websocket}"
         )
 
     def retrieve_job(self, job_id: str, use_websocket: Optional[bool] = None) -> None:
         """No retrieve_job_status offline"""
         raise QuantinuumAPIError(
-            (
-                f"Can't retrieve job status offline: job_id {job_id}."
-                f"\n use_websocket {use_websocket}"
-            )
+            f"Can't retrieve job status offline: job_id {job_id}."
+            f"\n use_websocket {use_websocket}"
         )
 
     def status(self, machine: str) -> str:
@@ -692,4 +682,4 @@ def status(self, machine: str) -> str:
 
     def cancel(self, job_id: str) -> dict:
         """No cancel offline"""
-        raise QuantinuumAPIError((f"Can't cancel job offline: job_id {job_id}."))
+        raise QuantinuumAPIError(f"Can't cancel job offline: job_id {job_id}.")