Config parsing rough initial idea

.env.qnx

@@ -0,0 +1,2 @@
+project_name=my cool project
+url=https://nexus.quantinuum.com

.gitignore

@@ -0,0 +1,11 @@
+
+__pycache__/
+
+# Mac Stuff
+**/.DS_Store
+
+# vs code
+*.code-workspace
+.vscode
+
+.env*.local

README.md

@@ -1,2 +1,19 @@
 # qnexus
 Quantinuum Nexus python client.
+
+```bash
+qnx jobs list                                                                                                                            
+Fetching jobs...
+Total jobs: 28 / Page 0 of 4 / Page Size: 15
+
+                                                Name Job Type     Status                            Project Id Progress
+0        compile-job-veiled-tawny-favorable-gigantea  Compile      Error  17fdb53c-d0d9-47fd-8204-5d9cce224fdb  Unknown
+1          compile-job-brittle-late-tender-marginata  Compile      Error  17fdb53c-d0d9-47fd-8204-5d9cce224fdb  Unknown
+2     compile-job-conical-booted-awestruck-variegata  Compile      Error  17fdb53c-d0d9-47fd-8204-5d9cce224fdb  Unknown
+3     compile-job-indistinct-american-linear-felleus  Compile      Error  3534740d-6eff-4b09-9ee3-bdabcb142d41  Unknown
+4  compile-job-subdecurrent-golden-nebulous-steph...  Compile  Completed  7b1144fc-2cc5-4404-a1d2-336c2b465bec  Unknown
+5         process-job-veiled-pink-mystical-rufescens  Execute  Completed  7b1144fc-2cc5-4404-a1d2-336c2b465bec  Unknown
+6     compile-job-pliable-variegated-forgiving-lutea  Compile  Completed  7b1144fc-2cc5-4404-a1d2-336c2b465bec  Unknown
+7       process-job-offset-club-nebulous-platyphylla  Execute  Completed  7b1144fc-2cc5-4404-a1d2-336c2b465bec  Unknown
+
+```

examples/README.md

@@ -0,0 +1,17 @@
+Some example usage of QNexus
+
+Run example script
+```bash
+poetry run python demo.py
+```
+
+Use QNX CLI globally
+```bash
+pipx install . --force 
+qnx
+```
+
+Use QNX CLI locally
+```bash
+poetry run qnx
+```

examples/demo.py

@@ -0,0 +1 @@
+import qnexus as qnx

pyproject.toml

@@ -0,0 +1,27 @@
+[tool.poetry]
+name = "qnexus"
+version = "0.1.0"
+description = "Quantinuum Nexus python client."
+authors = ["Your Name <[email protected]>"]
+readme = "README.md"
+
+[tool.poetry.dependencies]
+python = ">=3.10,<3.13"
+pydantic = "^2.4.2"
+uuid = "^1.30"
+colorama = "^0.4.6"
+click = "^8.1.7"
+httpx = "^0.25.0"
+pandas = "^2.1.1"
+pendulum = "^2.1.2"
+jinja2 = "^3.1.2"
+halo = "^0.0.31"
+rich = "^13.6.0"
+pytket = "^1.21.0"
+
+[tool.poetry.scripts]
+qnx = "qnexus.cli:entrypoint"
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"

qnexus/__init__.py

@@ -0,0 +1 @@
+from .client import *

qnexus/cli/__init__.py

@@ -0,0 +1,21 @@
+import click
+
+from .projects import projects
+from .jobs import jobs
+from .utils import status, init
+from ..config import get_config
+from .auth import login, logout
+
+
+@click.group()
+def entrypoint():
+    """Quantinuum Nexus API client."""
+    get_config()  # Will raise an exception if the config is invalid.
+
+
+entrypoint.add_command(init)
+entrypoint.add_command(status)
+entrypoint.add_command(projects)
+entrypoint.add_command(jobs)
+entrypoint.add_command(login)
+entrypoint.add_command(logout)

qnexus/cli/auth.py

@@ -0,0 +1,14 @@
+import click
+from ..client import auth as _auth
+
+
+@click.command()
+def login():
+    """Log in to quantinuum nexus using your web browser."""
+    click.echo(_auth.browser_login())
+
+
+@click.command()
+def logout():
+    """Log out of quantinuum nexus."""
+    click.echo(_auth.logout())

qnexus/cli/jobs.py

@@ -0,0 +1,22 @@
+import click
+from .utils import add_options_to_command
+from ..client import jobs as _jobs
+from typing_extensions import Unpack
+
+
+@click.command()
+def list_jobs(**kwargs: Unpack[_jobs.ParamsDict]):
+    """List all jobs"""
+    click.echo(_jobs.list_jobs(**kwargs))
+
+
+add_options_to_command(list_jobs, _jobs.Params)
+
+
+@click.group()
+def jobs():
+    """List jobs."""
+    pass
+
+
+jobs.add_command(list_jobs, name="list")

qnexus/cli/projects.py

@@ -0,0 +1,22 @@
+import click
+from .utils import add_options_to_command
+from ..client import projects as _projects
+from typing_extensions import Unpack
+
+
+@click.command()
+def list_projects(**kwargs: Unpack[_projects.ParamsDict]):
+    """List all projects"""
+    click.echo(_projects.list_projects(**kwargs))
+
+
+add_options_to_command(list_projects, _projects.Params)
+
+
+@click.group()
+def projects():
+    """List, create & delete circuits."""
+    pass
+
+
+projects.add_command(list_projects, name="list")

qnexus/cli/utils.py

@@ -0,0 +1,52 @@
+import click
+import os
+from colorama import Fore
+from ..consts import CONFIG_FILE_NAME
+from ..config import get_config
+from ..client import health as _health
+from click import Option, Command
+from typing import Any
+
+current_path = os.getcwd()
+current_dir = current_path.split(os.sep)[-1]
+
+
+# QNX utils interface
+@click.command()
+def status():
+    """Print a short summary of the current project."""
+    # click.echo(client.status())
+    click.echo(_health.status())
+
+
+@click.command()
+def init():
+    """Initialize a new qnexus project."""
+    # A project with that name already exists, use that one?
+    config = get_config()
+    if config:
+        raise click.ClickException(
+            Fore.GREEN
+            + f"Project already initialized: {Fore.YELLOW + config.project_name}"
+        )
+    if config is None:
+        name: str = click.prompt(
+            "Enter a project name:",
+            default=current_dir,
+        )
+        click.echo(Fore.GREEN + f"Intialized qnexus project: {name}")
+
+
+def add_options_to_command(command: Command, model: Any):
+    """Add click options using fields of a pydantic model."""
+    # Annotate command with options from dict
+    for field, value in model.model_fields.items():
+        command.params.append(
+            Option(
+                [f"--{field}"],
+                help=value.description,
+                show_default=True,
+                default=value.default,
+                multiple=isinstance(value.default, list),
+            )
+        )

qnexus/client/__init__.py

@@ -0,0 +1,5 @@
+from .health import *
+from .projects import *
+from .auth import *
+from .jobs import *
+from .utils import *

qnexus/client/auth.py

@@ -0,0 +1,108 @@
+import httpx
+from ..config import get_config
+from .utils import write_token_file, consolidate_error
+import webbrowser
+from http import HTTPStatus
+import time
+from rich.console import Console
+from rich.style import Style
+from rich.panel import Panel
+from rich.text import Text
+
+console = Console()
+
+config = get_config()
+
+
+def browser_login() -> None:
+    """
+    Log in to Quantinuum Nexus using the web browser.
+    ...
+    """
+
+    res = httpx.Client(base_url=f"{config.url}/auth").post(
+        "/device/device_authorization",
+        headers={"Content-Type": "application/x-www-form-urlencoded"},
+        data={"client_id": "scales", "scope": "myqos"},
+    )
+
+    user_code = res.json()["user_code"]
+    device_code = res.json()["device_code"]
+    verification_uri_complete = res.json()["verification_uri_complete"]
+    expires_in = res.json()["expires_in"]
+    poll_interval = res.json()["interval"]
+
+    webbrowser.open(verification_uri_complete, new=2)
+
+    token_request_body = {
+        "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
+        "device_code": device_code,
+        "client_id": "scales",
+    }
+
+    #  print(
+    #                 "Browser login initiated and will involve the following steps:\n"
+    #                 f"1. Visit this URL in a browser (using any device): {verification_uri_complete}\n"
+    #                 f"2. Confirm that the browser shows the following code: {user_code}\n"
+    #                 "3. Click 'Allow' and log in (with third-party such as Microsoft if required).\n"
+    #                 "4. Wait for this program to confirm successful login.\n"
+    #             )
+
+    console.print("\n")
+    console.print(
+        Panel(
+            f"""
+Confirm that the browser shows the following code:
+{Text(user_code, style=Style(bold=True, color="cyan"))}
+            """,
+            width=100,
+            style=Style(color="white", bold=True),
+        ),
+        justify="center",
+    )
+    console.print("\n")
+    console.print(
+        f"Browser didn't open automatically? Click this link: {verification_uri_complete}"
+    )
+    polling_for_seconds = 0
+    with console.status(
+        "[bold cyan]Waiting for user to confirm code via web browser...",
+        spinner_style="cyan",
+    ) as status:
+        while polling_for_seconds < expires_in:
+            time.sleep(poll_interval)
+            polling_for_seconds += poll_interval
+            resp = httpx.Client(base_url=f"{config.url}/auth").post(
+                "/device/token",
+                headers={"Content-Type": "application/x-www-form-urlencoded"},
+                data=token_request_body,
+            )
+            if (
+                resp.status_code == HTTPStatus.BAD_REQUEST
+                and resp.json().get("error") == "AUTHORIZATION_PENDING"
+            ):
+                continue
+            if resp.status_code == HTTPStatus.TOO_MANY_REQUESTS:
+                continue
+            if resp.status_code == HTTPStatus.OK:
+                resp_json = resp.json()
+                write_token_file("refresh_token", resp_json["refresh_token"])
+                write_token_file(
+                    "access_token",
+                    resp_json["access_token"],
+                )
+                print(
+                    f"Successfully logged in as {resp_json['email']} using the browser."
+                )
+                return
+            # Fail for all other statuses
+            consolidate_error(res=resp, description="Browser Login")
+            return
+    raise Exception("Browser Login Failed, code has expired.")
+
+
+def logout() -> None:
+    """Clear tokens from file system"""
+    write_token_file("refresh_token", "")
+    write_token_file("access_token", "")
+    print("Successfully logged out.")

qnexus/client/client.py

@@ -0,0 +1,55 @@
+import httpx
+from ..config import get_config
+from .utils import read_token_file, write_token_file
+from ..errors import NotAuthenticatedException
+import os
+
+
+config = get_config()
+
+
+def refresh_cookies(cookies: httpx.Cookies) -> None:
+    """Mutate cookie object with fresh access token and save updated cookies to disk."""
+    refresh_response = httpx.Client(base_url=f"{config.url}/auth").post(
+        "/tokens/refresh",
+        cookies=cookies,
+    )
+    cookies.extract_cookies(response=refresh_response)
+    write_token_file(
+        "access_token",
+        cookies.get("myqos_id", domain="nexus.quantinuum.com") or "",
+    )
+
+
+class AuthHandler(httpx.Auth):
+    """Custom nexus auth handler"""
+
+    requires_response_body = True
+
+    def auth_flow(self, request):
+
+        try: 
+            if access_token := read_token_file("access_token"):
+                httpx.Cookies({"myqos_id": access_token}).set_cookie_header(request)
+                response: httpx.Response = yield request
+                if response.status_code != 401:
+                    return response
+        except FileNotFoundError:
+            pass
+
+        try:
+            refresh_token = read_token_file("refresh_token")
+            cookies = httpx.Cookies({"myqos_oat": refresh_token})
+            refresh_cookies(cookies)
+            cookies.set_cookie_header(request)
+            response: httpx.Response = yield request
+            return response
+        except FileNotFoundError:
+            raise NotAuthenticatedException("Please login")
+
+
+config = get_config()
+nexus_client = httpx.Client(
+    base_url=config.url,
+    auth=AuthHandler(),
+)