Allow user to define a default account as an environment variable

qiskit_ibm_runtime/accounts/management.py

@@ -50,8 +50,10 @@ def save(
         verify: Optional[bool] = None,
         overwrite: Optional[bool] = False,
         channel_strategy: Optional[str] = None,
+        set_default_channel: Optional[bool] = True,
     ) -> None:
         """Save account on disk."""
+        default_channel = channel if set_default_channel else None
         cls.migrate(filename=filename)
         channel = channel or os.getenv("QISKIT_IBM_CHANNEL") or _DEFAULT_CHANNEL_TYPE
         name = name or cls._get_default_account_name(channel)
@@ -72,6 +74,7 @@ def save(
             )
             # avoid storing invalid accounts
             .validate().to_saved_format(),
+            default_channel=default_channel,
         )
 
     @staticmethod
@@ -105,13 +108,17 @@ def _matching_default(account_name: str) -> bool:
             else:
                 return account_name in default_accounts
 
+        account_dict = read_config(filename=filename)
+        if "default_channel" in account_dict:
+            del account_dict["default_channel"]
+
         # load all accounts
         all_accounts = map(
             lambda kv: (
                 kv[0],
                 Account.from_saved_format(kv[1]),
             ),
-            read_config(filename=filename).items(),
+            account_dict.items(),
         )
 
         # filter based on input parameters
@@ -148,6 +155,7 @@ def get(
         Raises:
             AccountNotFoundError: If the input value cannot be found on disk.
         """
+        # pylint: disable=too-many-return-statements
         filename = filename if filename else _DEFAULT_ACCOUNT_CONFIG_JSON_FILE
         filename = os.path.expanduser(filename)
         cls.migrate(filename)
@@ -157,7 +165,13 @@ def get(
                 raise AccountNotFoundError(f"Account with the name {name} does not exist on disk.")
             return Account.from_saved_format(saved_account)
 
-        channel_ = channel or os.getenv("QISKIT_IBM_CHANNEL") or _DEFAULT_CHANNEL_TYPE
+        default_channel = None
+        if os.path.isfile(_DEFAULT_ACCOUNT_CONFIG_JSON_FILE):
+            qiskit_json_data = read_config(_DEFAULT_ACCOUNT_CONFIG_JSON_FILE)
+            default_channel = qiskit_json_data.get("default_channel")
+        channel_ = (
+            channel or default_channel or os.getenv("QISKIT_IBM_CHANNEL") or _DEFAULT_CHANNEL_TYPE
+        )
         env_account = cls._from_env_variables(channel_)
         if env_account is not None:
             return env_account
@@ -174,6 +188,13 @@ def get(
             return Account.from_saved_format(saved_account)
 
         all_config = read_config(filename=filename)
+
+        # When 'channel' parameter is not defined, check for an account with the default channel
+        account_name = cls._get_default_account_name(channel=channel_)
+        if account_name in all_config:
+            return Account.from_saved_format(all_config[account_name])
+
+        # check for any account
         for channel_type in _CHANNEL_TYPES:
             account_name = cls._get_default_account_name(channel=channel_type)
             if account_name in all_config:

qiskit_ibm_runtime/accounts/storage.py

@@ -22,7 +22,9 @@
 logger = logging.getLogger(__name__)
 
 
-def save_config(filename: str, name: str, config: dict, overwrite: bool) -> None:
+def save_config(
+    filename: str, name: str, config: dict, overwrite: bool, default_channel: Optional[str] = None
+) -> None:
     """Save configuration data in a JSON file under the given name."""
     logger.debug("Save configuration data for '%s' in '%s'", name, filename)
     _ensure_file_exists(filename)
@@ -34,9 +36,19 @@ def save_config(filename: str, name: str, config: dict, overwrite: bool) -> None
         raise AccountAlreadyExistsError(
             f"Named account ({name}) already exists. " f"Set overwrite=True to overwrite."
         )
+    if (
+        data.get("default_channel")
+        and data.get("default_channel") != default_channel
+        and not overwrite
+    ):
+        raise AccountAlreadyExistsError(
+            f"default_channel ({name}) already exists. " f"Set overwrite=True to overwrite."
+        )
 
     with open(filename, mode="w", encoding="utf-8") as json_out:
         data[name] = config
+        if default_channel:
+            data["default_channel"] = default_channel
         json.dump(data, json_out, sort_keys=True, indent=4)
 
 

qiskit_ibm_runtime/qiskit_runtime_service.py

@@ -139,6 +139,7 @@ def __init__(
             - Account with the input `name`, if specified.
             - Default account for the `channel` type, if `channel` is specified but `token` is not.
             - Account defined by the input `channel` and `token`, if specified.
+            - Account defined by the `default_channel` if defined in filename
             - Account defined by the environment variables, if defined.
             - Default account for the ``ibm_cloud`` account, if one is available.
             - Default account for the ``ibm_quantum`` account, if one is available.
@@ -271,7 +272,7 @@ def _discover_account(
             if channel and channel not in ["ibm_cloud", "ibm_quantum"]:
                 raise ValueError("'channel' can only be 'ibm_cloud' or 'ibm_quantum'")
             channel = channel or self._get_channel_for_auth(auth=auth)
-            if token:
+            if channel and token:
                 account = Account(
                     channel=channel,
                     token=token,
@@ -291,6 +292,7 @@ def _discover_account(
                 "'channel' or 'auth' is required if 'token', or 'url' is specified but 'name' is not."
             )
 
+        # channel is not defined yet, get it from the AccountManager
         if account is None:
             account = AccountManager.get(filename=filename)
 
@@ -700,6 +702,7 @@ def save_account(
         verify: Optional[bool] = None,
         overwrite: Optional[bool] = False,
         channel_strategy: Optional[str] = None,
+        set_default_channel: Optional[bool] = True,
     ) -> None:
         """Save the account to disk for future use.
 
@@ -720,6 +723,8 @@ def save_account(
             verify: Verify the server's TLS certificate.
             overwrite: ``True`` if the existing account is to be overwritten.
             channel_strategy: Error mitigation strategy.
+            set_default_channel: If ``True``, the channel parameter is saved in filename,
+                as the default channel to use.
         """
 
         AccountManager.save(
@@ -733,6 +738,7 @@ def save_account(
             verify=verify,
             overwrite=overwrite,
             channel_strategy=channel_strategy,
+            set_default_channel=set_default_channel,
         )
 
     @staticmethod

test/account.py

@@ -152,6 +152,7 @@ def get_account_config_contents(
     instance=None,
     verify=None,
     proxies=None,
+    set_default=None,
 ):
     """Generate qiskitrc content"""
     if instance is None:
@@ -177,4 +178,6 @@ def get_account_config_contents(
         out[name]["verify"] = verify
     if proxies is not None:
         out[name]["proxies"] = proxies
+    if set_default:
+        out["default_channel"] = channel
     return out

test/unit/test_account.py

@@ -580,7 +580,7 @@ def test_delete_auth(self):
     def test_delete_filename(self):
         """Test delete accounts with filename parameter."""
 
-        filename = "~/account_to_delete.json"
+        filename = _TEST_FILENAME
         name = "key1"
         channel = "ibm_quantum"
         AccountManager.save(channel=channel, filename=filename, name=name, token="temp_token")
@@ -606,6 +606,84 @@ def test_account_with_filename(self):
         )
         self.assertEqual(account.token, dummy_token)
 
+    @temporary_account_config_file()
+    def test_default_env_channel(self):
+        """Test that if QISKIT_IBM_CHANNEL is set in the environment, this channel will be used"""
+        token = uuid.uuid4().hex
+        # unset default_channel in the environment
+        with temporary_account_config_file(token=token), no_envs("QISKIT_IBM_CHANNEL"):
+            service = FakeRuntimeService()
+        self.assertEqual(service.channel, "ibm_cloud")
+
+        # set channel to default channel in the environment
+        subtests = ["ibm_quantum", "ibm_cloud"]
+        for channel in subtests:
+            channel_env = {"QISKIT_IBM_CHANNEL": channel}
+            with temporary_account_config_file(channel=channel, token=token), custom_envs(
+                channel_env
+            ):
+                service = FakeRuntimeService()
+                self.assertEqual(service.channel, channel)
+
+    @temporary_account_config_file()
+    def test_save_default_channel(self):
+        """Test that if a default_channel is defined in the qiskit-ibm.json file,
+        this channel will be used"""
+        token = uuid.uuid4().hex
+        subtests = ["ibm_quantum", "ibm_cloud"]
+        for channel in subtests:
+            account_name = "default-" + channel.replace("_", "-")
+            account_fields = {"channel": channel, "token": token}
+            if channel == "ibm_cloud":
+                account_fields["instance"] = "some_instance"
+            with temporary_account_config_file(
+                contents={account_name: account_fields, "default_channel": channel}
+            ), no_envs("QISKIT_IBM_CHANNEL"):
+                service = FakeRuntimeService()
+                self.assertEqual(service.channel, channel)
+
+    @temporary_account_config_file()
+    def test_set_channel_precedence(self):
+        """Test the precedence of the various methods to set the channel:
+        parameter > json file > env variable > default channel"""
+        token = uuid.uuid4().hex
+        channel_env = {"QISKIT_IBM_CHANNEL": "ibm_cloud"}
+        contents = {
+            "default-ibm-cloud": {
+                "channel": "ibm_cloud",
+                "token": token,
+                "instance": "some_instance",
+            },
+            "default-ibm-quantum": {
+                "channel": "ibm_quantum",
+                "token": token,
+            },
+        }
+        contents["default_channel"] = "ibm_cloud"
+
+        # parameter channel is selected
+        with temporary_account_config_file(contents=contents), custom_envs(channel_env):
+            service = FakeRuntimeService(channel="ibm_quantum")
+            self.assertEqual(service.channel, "ibm_quantum")
+
+        # default channel from json file
+        contents["default_channel"] = "ibm_quantum"
+        with temporary_account_config_file(contents=contents), custom_envs(channel_env):
+            service = FakeRuntimeService()
+            self.assertEqual(service.channel, "ibm_quantum")
+
+        # channel from environment variable
+        del contents["default_channel"]
+        channel_env = {"QISKIT_IBM_CHANNEL": "ibm_quantum"}
+        with temporary_account_config_file(contents=contents), custom_envs(channel_env):
+            service = FakeRuntimeService()
+            self.assertEqual(service.channel, "ibm_quantum")
+
+        # default channel
+        with temporary_account_config_file(contents=contents), no_envs("QISKIT_IBM_CHANNEL"):
+            service = FakeRuntimeService()
+            self.assertEqual(service.channel, "ibm_cloud")
+
     def tearDown(self) -> None:
         """Test level tear down."""
         super().tearDown()
@@ -735,7 +813,10 @@ def test_enable_account_both_channel(self):
         token = uuid.uuid4().hex
         contents = get_account_config_contents(channel="ibm_cloud", token=token)
         contents.update(get_account_config_contents(channel="ibm_quantum", token=uuid.uuid4().hex))
-        with temporary_account_config_file(contents=contents), no_envs(["QISKIT_IBM_TOKEN"]):
+
+        with temporary_account_config_file(contents=contents), no_envs(
+            ["QISKIT_IBM_TOKEN", "QISKIT_IBM_CHANNEL"]
+        ):
             service = FakeRuntimeService()
         self.assertTrue(service._account)
         self.assertEqual(service._account.token, token)
@@ -754,7 +835,7 @@ def test_enable_account_by_env_channel(self):
                     "QISKIT_IBM_URL": url,
                     "QISKIT_IBM_INSTANCE": "h/g/p" if channel == "ibm_quantum" else "crn:12",
                 }
-                with custom_envs(envs):
+                with custom_envs(envs), no_envs("QISKIT_IBM_CHANNEL"):
                     service = FakeRuntimeService(channel=channel)
 
                 self.assertTrue(service._account)
@@ -871,7 +952,7 @@ def test_enable_account_by_env_pref(self):
                     "QISKIT_IBM_URL": url,
                     "QISKIT_IBM_INSTANCE": "my_crn",
                 }
-                with custom_envs(envs):
+                with custom_envs(envs), no_envs("QISKIT_IBM_CHANNEL"):
                     service = FakeRuntimeService(**extra)
 
                 self.assertTrue(service._account)