diff --git a/pyproject.toml b/pyproject.toml index 02d7d64ea..273d24847 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -97,6 +97,9 @@ ibm_backend = "qiskit_ibm_runtime.transpiler.plugin:IBMTranslationPlugin" ibm_dynamic_circuits = "qiskit_ibm_runtime.transpiler.plugin:IBMDynamicTranslationPlugin" ibm_fractional = "qiskit_ibm_runtime.transpiler.plugin:IBMFractionalTranslationPlugin" +[project.entry-points."console_scripts"] +qiskit-ibm-runtime = "qiskit_ibm_runtime._cli:entry_point" + [project.urls] documentation = "https://docs.quantum.ibm.com/" repository = "https://github.com/Qiskit/qiskit-ibm-runtime" diff --git a/qiskit_ibm_runtime/_cli.py b/qiskit_ibm_runtime/_cli.py new file mode 100644 index 000000000..71ba7c75a --- /dev/null +++ b/qiskit_ibm_runtime/_cli.py @@ -0,0 +1,247 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 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. + +""" +The `save-account` command-line interface. + +These classes and functions are not public. +""" + +import argparse +import sys +from getpass import getpass +from typing import List, Literal, Callable, TypeVar + +from ibm_cloud_sdk_core.api_exception import ApiException + +from .qiskit_runtime_service import QiskitRuntimeService +from .exceptions import IBMNotAuthorizedError +from .api.exceptions import RequestsApiError +from .accounts.management import AccountManager, _DEFAULT_ACCOUNT_CONFIG_JSON_FILE +from .accounts.exceptions import AccountAlreadyExistsError + +Channel = Literal["ibm_quantum", "ibm_cloud"] +T = TypeVar("T") + + +def entry_point() -> None: + """ + This is the entry point for the `qiskit-ibm-runtime` command. At the + moment, we only support one script (save-account), but we want to have a + `qiskit-ibm-runtime` command so users can run `pipx run qiskit-ibm-runtime + save-account`. + """ + # Use argparse to create the --help feature + parser = argparse.ArgumentParser( + prog="qiskit-ibm-runtime", + description="Commands for the Qiskit IBM Runtime Python package", + ) + parser.add_subparsers( + title="Commands", + description="This package supports the following command:", + dest="script", + required=True, + ).add_parser( + "save-account", + description=( + "An interactive command-line interface to save your Qiskit IBM " + "Runtime account locally. This script is interactive-only." + ), + help="Interactive command-line interface to save your account locally.", + ).add_argument( + "--no-color", action="store_true", help="Hide ANSI escape codes in output" + ) + args = parser.parse_args() + use_color = not args.no_color + if args.script == "save-account": + try: + SaveAccountCLI(color=use_color).main() + except KeyboardInterrupt: + sys.exit() + + +class Formatter: + """Format using terminal escape codes""" + + # pylint: disable=missing-function-docstring + # + def __init__(self, *, color: bool) -> None: + self.color = color + + def box(self, lines: List[str]) -> str: + """Print lines in a box using Unicode box-drawing characters""" + width = max(len(line) for line in lines) + styled_lines = [self.text(line.ljust(width), "bold") for line in lines] + box_lines = [ + "╭─" + "─" * width + "─╮", + *(f"│ {line} │" for line in styled_lines), + "╰─" + "─" * width + "─╯", + ] + return "\n".join(box_lines) + + def text(self, text: str, styles: str) -> str: + if not self.color: + return text + codes = { + "bold": 1, + "green": 32, + "red": 31, + "cyan": 36, + } + ansi_start = "".join([f"\033[{codes[style]}m" for style in styles.split(" ")]) + ansi_end = "\033[0m" + return f"{ansi_start}{text}{ansi_end}" + + +class SaveAccountCLI: + """ + This class contains the save-account command and helper functions. + """ + + def __init__(self, *, color: bool) -> None: + self.fmt = Formatter(color=color) + + def main(self) -> None: + """ + A CLI that guides users through getting their account information and + saving it to disk. + """ + print(self.fmt.box(["Qiskit IBM Runtime save account"])) + channel = self.get_channel() + token = self.get_token(channel) + print("Verifying, this might take few seconds...") + try: + service = QiskitRuntimeService(channel=channel, token=token) + except (ApiException, IBMNotAuthorizedError, RequestsApiError) as err: + print( + self.fmt.text("\nError while authorizing with your token\n", "red bold") + + self.fmt.text(err.message or "", "red"), + file=sys.stderr, + ) + sys.exit(1) + instance = self.get_instance(service) + self.save_to_disk( + { + "channel": channel, + "token": token, + "instance": instance, + } + ) + + def get_channel(self) -> Channel: + """Ask user which channel to use""" + print(self.fmt.text("Select a channel", "bold")) + return UserInput.select_from_list(["ibm_quantum", "ibm_cloud"], self.fmt) + + def get_token(self, channel: Channel) -> str: + """Ask user for their token""" + token_url = { + "ibm_quantum": "https://quantum.ibm.com", + "ibm_cloud": "https://cloud.ibm.com/iam/apikeys", + }[channel] + styled_token_url = self.fmt.text(token_url, "cyan") + print( + self.fmt.text("\nPaste your API token", "bold") + + f"\nYou can get this from {styled_token_url}." + + "\nFor security, you might not see any feedback when typing." + ) + return UserInput.token() + + def get_instance(self, service: QiskitRuntimeService) -> str: + """ + Ask user which instance to use, or select automatically if only one + is available. + """ + instances = service.instances() + if len(instances) == 1: + instance = instances[0] + print("Using instance " + self.fmt.text(instance, "green bold")) + return instance + print(self.fmt.text("\nSelect a default instance", "bold")) + return UserInput.select_from_list(instances, self.fmt) + + def save_to_disk(self, account: dict) -> None: + """ + Save account details to disk, confirming if they'd like to overwrite if + one exists already. Display a warning that token is stored in plain + text. + """ + try: + AccountManager.save(**account) + except AccountAlreadyExistsError: + response = UserInput.input( + message="\nDefault account already exists, would you like to overwrite it? (y/N):", + is_valid=lambda response: response.strip().lower() in ["y", "yes", "n", "no", ""], + ) + if response.strip().lower() in ["y", "yes"]: + AccountManager.save(**account, overwrite=True) + else: + print("Account not saved.", file=sys.stderr) + return + + print("Account saved to " + self.fmt.text(_DEFAULT_ACCOUNT_CONFIG_JSON_FILE, "green bold")) + print( + self.fmt.box( + [ + "⚠️ Warning: your token is saved to disk in plain text.", + "If on a shared computer, make sure to revoke your token", + "by regenerating it in your account settings when finished.", + ] + ), + file=sys.stderr, + ) + + +class UserInput: + """ + Helper functions to get different types input from user. + """ + + @staticmethod + def input(message: str, is_valid: Callable[[str], bool]) -> str: + """ + Repeatedly ask user for input until they give us something that satisifies + `is_valid`. + """ + while True: + response = input(message + " ").strip() + if response in ["q", "quit"]: + sys.exit() + if is_valid(response): + return response + print("Did not understand input, trying again... (or type 'q' to quit)") + + @staticmethod + def token() -> str: + """Ask for API token, prompting again if empty""" + while True: + token = getpass("Token: ").strip() + if token != "": + return token + + @staticmethod + def select_from_list(options: List[T], formatter: Formatter) -> T: + """ + Prompt user to select from a list of options by entering a number. + """ + print() + for index, option in enumerate(options): + print(f" ({index+1}) {option}") + print() + response = UserInput.input( + message=f"Enter a number 1-{len(options)} and press enter:", + is_valid=lambda response: response.isdigit() + and int(response) in range(1, len(options) + 1), + ) + choice = options[int(response) - 1] + print("Selected " + formatter.text(str(choice), "green bold")) + return choice diff --git a/test/unit/test_cli.py b/test/unit/test_cli.py new file mode 100644 index 000000000..46b495dfe --- /dev/null +++ b/test/unit/test_cli.py @@ -0,0 +1,196 @@ +# This code is part of Qiskit. +# +# (C) Copyright IBM 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. + +"""Tests CLI that saves user account to disk.""" +from typing import List +import unittest +from unittest.mock import patch +from textwrap import dedent + +from qiskit_ibm_runtime._cli import SaveAccountCLI, UserInput, Formatter + +from qiskit_ibm_runtime.accounts.account import IBM_CLOUD_API_URL, IBM_QUANTUM_API_URL +from ..ibm_test_case import IBMTestCase + + +class MockIO: + """Mock `input` and `getpass`""" + + # pylint: disable=missing-function-docstring + + def __init__(self, inputs: List[str]): + self.inputs = inputs + self.output = "" + + def mock_input(self, *args, **_kwargs): + if args: + self.mock_print(args[0]) + return self.inputs.pop(0) + + def mock_print(self, *args, **_kwargs): + self.output += " ".join(args) + "\n" + + +class TestCLI(IBMTestCase): + """Tests for the save-account CLI.""" + + # pylint: disable=missing-class-docstring, missing-function-docstring + + def test_select_from_list(self): + """Test the `UserInput.select_from_list` helper method""" + self.maxDiff = 3000 # pylint: disable=invalid-name + + # Check a bunch of invalid inputs before entering a valid one + mockio = MockIO(["", "0", "-1", "3.14", "9", " 3"]) + + @patch("builtins.input", mockio.mock_input) + @patch("builtins.print", mockio.mock_print) + def run_test(): + choice = UserInput.select_from_list(["a", "b", "c", "d"], Formatter(color=False)) + self.assertEqual(choice, "c") + + run_test() + self.assertEqual(mockio.inputs, []) + self.assertEqual( + mockio.output, + dedent( + """ + (1) a + (2) b + (3) c + (4) d + + Enter a number 1-4 and press enter:• + Did not understand input, trying again... (or type 'q' to quit) + Enter a number 1-4 and press enter:• + Did not understand input, trying again... (or type 'q' to quit) + Enter a number 1-4 and press enter:• + Did not understand input, trying again... (or type 'q' to quit) + Enter a number 1-4 and press enter:• + Did not understand input, trying again... (or type 'q' to quit) + Enter a number 1-4 and press enter:• + Did not understand input, trying again... (or type 'q' to quit) + Enter a number 1-4 and press enter:• + Selected c + """.replace( + "•", " " + ) + ), + ) + + def test_cli_multiple_instances_saved_account(self): + """ + Full CLI: User has many instances and account saved. + """ + token = "Password123" + instances = ["my/instance/1", "my/instance/2", "my/instance/3"] + selected_instance = 2 # == instances[1] + + class MockRuntimeService: + def __init__(self, *_args, **_kwargs): + pass + + def instances(self): + return instances + + expected_saved_account = dedent( + f""" + {{ + "default": {{ + "channel": "ibm_quantum", + "instance": "{instances[selected_instance-1]}", + "private_endpoint": false, + "token": "{token}", + "url": "{IBM_QUANTUM_API_URL}" + }} + }} + """ + ) + + existing_account = dedent( + """ + { + "default": { + "channel": "ibm_quantum", + "instance": "my/instance/0", + "private_endpoint": false, + "token": "super-secret-token", + "url": "https://auth.quantum-computing.ibm.com/api" + } + } + """ + ) + + mockio = MockIO(["1", token, str(selected_instance), "yes"]) + mock_open = unittest.mock.mock_open(read_data=existing_account) + + @patch("builtins.input", mockio.mock_input) + @patch("builtins.open", mock_open) + @patch("builtins.print", mockio.mock_print) + @patch("os.path.isfile", lambda *args: True) + @patch("qiskit_ibm_runtime._cli.getpass", mockio.mock_input) + @patch("qiskit_ibm_runtime._cli.QiskitRuntimeService", MockRuntimeService) + def run_cli(): + SaveAccountCLI(color=True).main() + + run_cli() + self.assertEqual(mockio.inputs, []) + + written_output = "".join(call.args[0] for call in mock_open().write.mock_calls) + self.assertEqual(written_output.strip(), expected_saved_account.strip()) + + def test_cli_one_instance_no_saved_account(self): + """ + Full CLI: user only has one instance and no account saved. + """ + token = "QJjjbOxSfzZiskMZiyty" + instance = "my/only/instance" + + class MockRuntimeService: + def __init__(self, *_args, **_kwargs): + pass + + def instances(self): + return [instance] + + expected_saved_account = dedent( + f""" + {{ + "default": {{ + "channel": "ibm_cloud", + "instance": "{instance}", + "private_endpoint": false, + "token": "{token}", + "url": "{IBM_CLOUD_API_URL}" + }} + }} + """ + ) + + mockio = MockIO(["2", token]) + mock_open = unittest.mock.mock_open(read_data="{}") + + @patch("builtins.input", mockio.mock_input) + @patch("builtins.open", mock_open) + @patch("builtins.print", mockio.mock_print) + @patch("os.path.isfile", lambda *args: False) + @patch("qiskit_ibm_runtime._cli.getpass", mockio.mock_input) + @patch("qiskit_ibm_runtime._cli.QiskitRuntimeService", MockRuntimeService) + def run_cli(): + SaveAccountCLI(color=True).main() + + run_cli() + self.assertEqual(mockio.inputs, []) + + written_output = "".join(call.args[0] for call in mock_open().write.mock_calls) + # The extra "{}" is runtime ensuring the file exists + self.assertEqual(written_output.strip(), "{}" + expected_saved_account.strip())