Adding ty type checking (experimental)

.github/workflows/ci.yml

@@ -73,7 +73,7 @@ jobs:
           checks/lint_.py
 
   mypy:
-    name: Type check
+    name: Type check (mypy)
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v6
@@ -93,6 +93,27 @@ jobs:
         run: |
           checks/mypy_.py
 
+  ty:
+    name: Type check (ty)
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v6
+      - name: Set up Python 3.14
+        uses: actions/setup-python@v6
+        with:
+          python-version: 3.14
+      - uses: actions/cache@v5
+        with:
+          path: ${{ env.pythonLocation }}
+          key: ${{ env.pythonLocation }}-${{ hashFiles('**/pyproject.toml', '**/*requirements.txt', '.github/workflows/ci.yml') }}
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -e ./checks-superstaq -e ./general-superstaq[dev] -e ./qiskit-superstaq[dev] -e ./cirq-superstaq[dev] -e ./supermarq-benchmarks[dev]
+      - name: Type check
+        run: |
+          checks/ty_.py general-superstaq
+
   coverage:
     name: Pytest and Coverage check
     strategy:

checks-superstaq/checks_superstaq/__init__.py

@@ -26,6 +26,7 @@
     mypy_,
     pytest_,
     requirements,
+    ty_,
 )
 
 __all__ = [
@@ -41,4 +42,5 @@
     "mypy_",
     "pytest_",
     "requirements",
+    "ty_",
 ]

checks-superstaq/checks_superstaq/check_utils.py

@@ -297,15 +297,7 @@ def get_test_files(
 ####################################################################################################
 # file parsing, incremental checks, and decorator to exit instead of returning a failing exit code
 
-CHECK_LIST = [
-    "configs",
-    "format",
-    "mypy",
-    "pytest",
-    "coverage",
-    "requirements",
-    "build_docs",
-]
+CHECK_LIST = ["configs", "format", "mypy", "pytest", "coverage", "requirements", "build_docs", "ty"]
 
 
 def get_check_parser(no_files: bool = False) -> argparse.ArgumentParser:

checks-superstaq/checks_superstaq/ty_.py

@@ -0,0 +1,64 @@
+#!/usr/bin/env python3
+# Copyright 2026 Infleqtion
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import annotations
+
+import subprocess
+import sys
+import textwrap
+from collections.abc import Iterable
+
+from checks_superstaq import check_utils
+
+
+@check_utils.enable_exit_on_failure
+def run(
+    *args: str,
+    include: str | Iterable[str] = "*.py",
+    exclude: str | Iterable[str] = (),
+    silent: bool = False,
+) -> int:
+    """Runs ty on the repository (typing check).
+
+    Args:
+        *args: Command line arguments.
+        include: Glob(s) indicating which tracked files to consider (e.g. "*.py").
+        exclude: Glob(s) indicating which tracked files to skip (e.g. "*integration_test.py").
+        silent: If True, restrict printing to warning and error messages.
+
+    Returns:
+        Terminal exit code. 0 indicates success, while any other integer indicates a test failure.
+    """
+    parser = check_utils.get_check_parser()
+    parser.description = textwrap.dedent(
+        """
+        Runs ty on the repository (typing check).
+        Ignores files in the [repo_root]/examples directory.
+        """
+    )
+
+    parsed_args, args_to_pass = parser.parse_known_intermixed_args(args)
+    if "ty" in parsed_args.skip:
+        return 0
+
+    files = check_utils.extract_files(parsed_args, include, exclude, silent)
+
+    return subprocess.call(
+        [sys.executable, "-m", "ty", "check", *files, *args_to_pass], cwd=check_utils.root_dir
+    )
+
+
+if __name__ == "__main__":
+    sys.exit(run(*sys.argv[1:]))

checks-superstaq/requirements.txt

@@ -15,3 +15,7 @@ ruff>=0.13.0
 setuptools>=67.0.0
 sphinx-autoapi>=3.2.1
 sphinx-rtd-theme>=1.0.0
+ty>=0.0.29
+types-decorator>=5.2.0
+types-PyYAML>=6.0.12
+types-requests>=2.32.0

general-superstaq/general_superstaq/check/__init__.py → checks/ty_.py

@@ -1,3 +1,4 @@
+#!/usr/bin/env python3
 # Copyright 2026 Infleqtion
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -11,30 +12,12 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-from checks_superstaq import (  # To be removed in version 0.5.0
-    __version__,
-    all_,
-    build_docs,
-    check_utils,
-    configs,
-    coverage_,
-    format_,
-    lint_,
-    mypy_,
-    pytest_,
-    requirements,
-)
 
-__all__ = [
-    "__version__",
-    "all_",
-    "build_docs",
-    "check_utils",
-    "configs",
-    "coverage_",
-    "format_",
-    "lint_",
-    "mypy_",
-    "pytest_",
-    "requirements",
-]
+from __future__ import annotations
+
+import sys
+
+import checks_superstaq as checks
+
+if __name__ == "__main__":
+    sys.exit(checks.ty_.run(*sys.argv[1:]))

general-superstaq/general_superstaq/compiler_output.py

@@ -163,6 +163,21 @@ def __init__(
         )
 
 
+def _deserialize_qubit_mappings(json_text: str) -> list[dict[int, int]]:
+    """Deserializes qubit mappings encoded as JSON lists of key-value pairs.
+
+    Args:
+        json_text: A JSON string representing qubit mappings.
+
+    Returns:
+        A list of dictionaries mapping logical qubits to physical qubits.
+    """
+    # The service serializes mappings as key-value pairs so JSON round-tripping preserves integer
+    # qubit indices instead of coercing dictionary keys to strings.
+    serialized_mappings: list[list[tuple[int, int]]] = json.loads(json_text)
+    return [dict(qubit_mapping) for qubit_mapping in serialized_mappings]
+
+
 def _jaqal_programs_to_subcircuits(jaqal_programs: Sequence[str]) -> str:
     separator = "prepare_all"
     subcircuits = [jaqal_programs[0]]
@@ -185,15 +200,15 @@ def read_json_jaqal(
     """
     compiled_circuits = json.loads(json_dict["jaqal_strs"])
 
-    initial_logical_to_physicals_list: list[dict[int, int]] = list(
-        map(dict, json.loads(json_dict["initial_logical_to_physicals"]))
+    initial_logical_to_physicals_list = _deserialize_qubit_mappings(
+        json_dict["initial_logical_to_physicals"]
     )
     initial_logical_to_physicals: list[dict[int, int]] | list[list[dict[int, int]]] = (
         initial_logical_to_physicals_list
     )
 
-    final_logical_to_physicals_list: list[dict[int, int]] = list(
-        map(dict, json.loads(json_dict["final_logical_to_physicals"]))
+    final_logical_to_physicals_list = _deserialize_qubit_mappings(
+        json_dict["final_logical_to_physicals"]
     )
     final_logical_to_physicals: list[dict[int, int]] | list[list[dict[int, int]]] = (
         final_logical_to_physicals_list

general-superstaq/general_superstaq/compiler_output_test.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 from __future__ import annotations
 
+import json
 import textwrap
 
 import general_superstaq as gss
@@ -78,3 +79,24 @@ def test_compiler_output_eq() -> None:
     ) != gss.compiler_output.CompilerOutput(
         [jaqal_program, jaqal_program_alt], [{0: 0}, {}], [{0: 1}, {}]
     )
+
+
+def test_read_json_jaqal_deserializes_qubit_mappings() -> None:
+    jaqal_program = textwrap.dedent(
+        """\
+        register allqubits[1]
+
+        prepare_all
+        measure_all
+        """
+    )
+    compiler_output = gss.compiler_output.read_json_jaqal(
+        {
+            "jaqal_strs": json.dumps([jaqal_program]),
+            "initial_logical_to_physicals": json.dumps([[(0, 1)], [(2, 3)]]),
+            "final_logical_to_physicals": json.dumps([[(0, 4)], [(2, 5)]]),
+        }
+    )
+
+    assert compiler_output.initial_logical_to_physicals == [{0: 1}, {2: 3}]
+    assert compiler_output.final_logical_to_physicals == [{0: 4}, {2: 5}]

general-superstaq/general_superstaq/job.py

@@ -78,8 +78,8 @@ def _refresh_job(self) -> None:
         if self._job_data is not None:
             if all(s in gss.models.TERMINAL_CIRCUIT_STATES for s in self._job_data.statuses):
                 return
-        self._job_data = gss.models.JobData(
-            **self._client.fetch_jobs([self._job_id])[str(self._job_id)]
+        self._job_data = gss.models.JobData.model_validate(
+            self._client.fetch_jobs([self._job_id])[str(self._job_id)]
         )
         self._update_status_queue_info()
 

general-superstaq/general_superstaq/job_test.py

@@ -120,7 +120,7 @@ def test_to_dict(
     mock_get.return_value = _mocked_response({str(uuid.UUID(int=123)): job_dict})
 
     job = gss.job.Job(mock_client, uuid.UUID(int=123))
-    assert job.to_dict() == gss.models.JobData(**job_dict).model_dump()
+    assert job.to_dict() == gss.models.JobData.model_validate(job_dict).model_dump()
 
 
 def test_equality(mock_client: gss.superstaq_client._SuperstaqClientV3) -> None:
@@ -286,23 +286,23 @@ def test_update_status_queue_info(mock_client: gss.superstaq_client._SuperstaqCl
     job_dict = _job_dict()
     job_dict["num_circuits"] = 3
     job_dict["statuses"] = ["completed"] * 3
-    job._job_data = gss.models.JobData(**job_dict)
+    job._job_data = gss.models.JobData.model_validate(job_dict)
 
     job._update_status_queue_info()
     assert job._overall_status == "completed"
 
     job_dict["statuses"] = ["awaiting_submission", "cancelled", "cancelled"]
-    job._job_data = gss.models.JobData(**job_dict)
+    job._job_data = gss.models.JobData.model_validate(job_dict)
     job._update_status_queue_info()
     assert job._overall_status == "awaiting_submission"
 
     job_dict["statuses"] = ["cancelled", "cancelled", "awaiting_submission"]
-    job._job_data = gss.models.JobData(**job_dict)
+    job._job_data = gss.models.JobData.model_validate(job_dict)
     job._update_status_queue_info()
     assert job._overall_status == "awaiting_submission"
 
     job_dict["statuses"] = ["completed", "completed", "failed"]
-    job._job_data = gss.models.JobData(**job_dict)
+    job._job_data = gss.models.JobData.model_validate(job_dict)
     job._update_status_queue_info()
     assert job._overall_status == "failed"
 

general-superstaq/general_superstaq/machine_api.py

@@ -56,12 +56,12 @@ def post_result(
         the task result including checking if the number of shots in the bitstring is equal to the
         number requested and if the bitstrings match the set of qubit_readout operations requested.
         """
-        compressed_bitstrings: dict[str, list[int]] | None = None
+        compressed_bitstrings: dict[str, set[int]] | None = None
         if bitstrings is not None:
             compressed_bitstrings = {}
             for idx, bs in enumerate(bitstrings):
-                bs_index_list = compressed_bitstrings.setdefault(bs, [])
-                bs_index_list.append(idx)
+                bs_index_list = compressed_bitstrings.setdefault(bs, set())
+                bs_index_list.add(idx)
 
         results = gss.models.WorkerTaskResults(
             circuit_ref=task_id,

general-superstaq/general_superstaq/models.py

@@ -21,7 +21,7 @@
 import uuid
 from collections.abc import Mapping, Sequence
 from enum import Enum
-from typing import Annotated, Any
+from typing import Annotated, Any, cast
 
 import pydantic.functional_validators
 
@@ -153,7 +153,8 @@ def _validate_cq_token(cls, cq_token: object) -> object:
         Previously CQ tokens were specified via a dict, e.g. `cq_token={"access_token": "token"}`.
         """
         if isinstance(cq_token, Mapping):
-            return cq_token.get("access_token")
+            cq_token_mapping = cast("Mapping[object, object]", cq_token)
+            return cq_token_mapping.get("access_token")
         return cq_token
 
 

general-superstaq/general_superstaq/models_test.py

@@ -51,7 +51,7 @@ def test_external_provider_credentials() -> None:
 
     # Old-style CQ credentials
     options_old = {"cq_token": {"access_token": "token"}, "project_id": "123", "org_id": "456"}
-    credentials = gss.models.ExternalProviderCredentials(**options_old)
+    credentials = gss.models.ExternalProviderCredentials.model_validate(options_old)
     assert credentials.cq_token == "token"
     assert credentials.cq_project_id == "123"
     assert credentials.cq_org_id == "456"
@@ -121,7 +121,7 @@ def test_worker_task_results_validation() -> None:
             circuit_ref="f76e84f7-0c65-4f0b-b2d7-14135db3900c",
             status=gss.models.CircuitStatus.COMPLETED,
             successful_shots=10,
-            measurements={"a": [0], "b": [1, 2]},
+            measurements={"a": {0}, "b": {1, 2}},
         )
 
     with pytest.raises(
@@ -132,7 +132,7 @@ def test_worker_task_results_validation() -> None:
             circuit_ref="f76e84f7-0c65-4f0b-b2d7-14135db3900c",
             status=gss.models.CircuitStatus.COMPLETED,
             successful_shots=10,
-            measurements={"101": [0], "01": [1, 2]},
+            measurements={"101": {0}, "01": {1, 2}},
         )
 
     with pytest.raises(
@@ -143,14 +143,14 @@ def test_worker_task_results_validation() -> None:
             circuit_ref="f76e84f7-0c65-4f0b-b2d7-14135db3900c",
             status=gss.models.CircuitStatus.COMPLETED,
             successful_shots=2,
-            measurements={"101": [0], "001": [1, 3]},
+            measurements={"101": {0}, "001": {1, 3}},
         )
 
     _ = gss.models.WorkerTaskResults(
         circuit_ref="f76e84f7-0c65-4f0b-b2d7-14135db3900c",
         status=gss.models.CircuitStatus.COMPLETED,
         successful_shots=4,
-        measurements={"000": [0, 1, 3], "101": [2]},
+        measurements={"000": {0, 1, 3}, "101": {2}},
     )
 
     _ = gss.models.WorkerTaskResults(