Migrate from "auth" to "channel" parameter

.github/workflows/ci.yml

@@ -111,7 +111,7 @@ jobs:
       matrix:
         python-version: [ 3.9 ]
         os: [ "ubuntu-latest" ]
-        environment: [ "legacy-production" ]
+        environment: [ "ibm-quantum-production" ]
     environment: ${{ matrix.environment }}
     env:
       QISKIT_IBM_TOKEN: ${{ secrets.QISKIT_IBM_TOKEN }}

.github/workflows/integration-tests.yml

@@ -26,7 +26,7 @@ jobs:
       matrix:
         python-version: [ 3.9 ]
         os: [ "ubuntu-latest" ]
-        environment: [ "legacy-production", "legacy-staging" ]
+        environment: [ "ibm-quantum-production", "ibm-quantum-staging" ]
     environment: ${{ matrix.environment }}
     env:
       QISKIT_IBM_TOKEN: ${{ secrets.QISKIT_IBM_TOKEN }}

qiskit_ibm_provider/accounts/__init__.py

@@ -14,7 +14,7 @@
 Account management functionality.
 """
 
-from .account import Account, AccountType
+from .account import Account, AccountType, ChannelType
 from .management import AccountManager
 from .exceptions import (
     AccountNotFoundError,

qiskit_ibm_provider/accounts/account.py

@@ -20,13 +20,14 @@
 from typing_extensions import Literal
 
 from .exceptions import InvalidAccountError
-from ..api.auth import LegacyAuth
+from ..api.auth import QuantumAuth
 from ..proxies import ProxyConfiguration
 from ..utils.hgp import from_instance_format
 
 AccountType = Optional[Literal["cloud", "legacy"]]
+ChannelType = Optional[Literal["ibm_cloud", "ibm_quantum"]]
 
-LEGACY_API_URL = "https://auth.quantum-computing.ibm.com/api"
+IBM_QUANTUM_API_URL = "https://auth.quantum-computing.ibm.com/api"
 logger = logging.getLogger(__name__)
 
 
@@ -35,7 +36,7 @@ class Account:
 
     def __init__(
         self,
-        auth: AccountType,
+        channel: ChannelType,
         token: str,
         url: Optional[str] = None,
         instance: Optional[str] = None,
@@ -45,16 +46,16 @@ def __init__(
         """Account constructor.
 
         Args:
-            auth: Authentication type, ``cloud`` or ``legacy``.
+            channel: Channel type, ``ibm_cloud`` or ``ibm_quantum``.
             token: Account token to use.
             url: Authentication URL.
             instance: Service instance to use.
             proxies: Proxy configuration.
             verify: Whether to verify server's TLS certificate.
         """
-        resolved_url = url or LEGACY_API_URL
+        resolved_url = url or IBM_QUANTUM_API_URL
 
-        self.auth = auth
+        self.channel = channel
         self.token = token
         self.url = resolved_url
         self.instance = instance
@@ -73,7 +74,7 @@ def from_saved_format(cls, data: dict) -> "Account":
         """Creates an account instance from data saved on disk."""
         proxies = data.get("proxies")
         return cls(
-            auth=data.get("auth"),
+            channel=data.get("channel"),
             url=data.get("url"),
             token=data.get("token"),
             instance=data.get("instance"),
@@ -83,14 +84,14 @@ def from_saved_format(cls, data: dict) -> "Account":
 
     def get_auth_handler(self) -> AuthBase:
         """Returns the respective authentication handler."""
-        return LegacyAuth(access_token=self.token)
+        return QuantumAuth(access_token=self.token)
 
     def __eq__(self, other: object) -> bool:
         if not isinstance(other, Account):
             return False
         return all(
             [
-                self.auth == other.auth,
+                self.channel == other.channel,
                 self.token == other.token,
                 self.url == other.url,
                 self.instance == other.instance,
@@ -109,19 +110,20 @@ def validate(self) -> "Account":
             This Account instance.
         """
 
-        self._assert_valid_auth(self.auth)
+        self._assert_valid_channel(self.channel)
         self._assert_valid_token(self.token)
         self._assert_valid_url(self.url)
-        self._assert_valid_instance(self.auth, self.instance)
+        self._assert_valid_instance(self.channel, self.instance)
         self._assert_valid_proxies(self.proxies)
         return self
 
     @staticmethod
-    def _assert_valid_auth(auth: AccountType) -> None:
-        """Assert that the auth parameter is valid."""
-        if not (auth in ["cloud", "legacy"]):
+    def _assert_valid_channel(channel: ChannelType) -> None:
+        """Assert that the channel parameter is valid."""
+        if not (channel in ["ibm_cloud", "ibm_quantum"]):
             raise InvalidAccountError(
-                f"Invalid `auth` value. Expected one of ['cloud', 'legacy'], got '{auth}'."
+                f"Invalid `channel` value. Expected one of "
+                f"{['ibm_cloud', 'ibm_quantum']}, got '{channel}'."
             )
 
     @staticmethod
@@ -149,14 +151,14 @@ def _assert_valid_proxies(config: ProxyConfiguration) -> None:
             config.validate()
 
     @staticmethod
-    def _assert_valid_instance(auth: AccountType, instance: str) -> None:
+    def _assert_valid_instance(channel: ChannelType, instance: str) -> None:
         """Assert that the instance name is valid for the given account type."""
-        if auth == "cloud":
+        if channel == "ibm_cloud":
             if not (isinstance(instance, str) and len(instance) > 0):
                 raise InvalidAccountError(
                     f"Invalid `instance` value. Expected a non-empty string, got '{instance}'."
                 )
-        if auth == "legacy":
+        if channel == "ibm_quantum":
             if instance is not None:
                 try:
                     from_instance_format(instance)

qiskit_ibm_provider/accounts/management.py

@@ -15,7 +15,7 @@
 import os
 from typing import Optional, Dict
 from .exceptions import AccountNotFoundError
-from .account import Account, AccountType
+from .account import Account, ChannelType
 from ..proxies import ProxyConfiguration
 from .storage import save_config, read_config, delete_config
 
@@ -25,8 +25,10 @@
 _DEFAULT_ACCOUNT_NAME = "default"
 _DEFAULT_ACCOUNT_NAME_LEGACY = "default-legacy"
 _DEFAULT_ACCOUNT_NAME_CLOUD = "default-cloud"
-_DEFAULT_ACCOUNT_TYPE: AccountType = "legacy"
-_ACCOUNT_TYPES = [_DEFAULT_ACCOUNT_TYPE, "legacy"]
+_DEFAULT_ACCOUNT_NAME_IBM_QUANTUM = "default-ibm-quantum"
+_DEFAULT_ACCOUNT_NAME_IBM_CLOUD = "default-ibm-cloud"
+_DEFAULT_CHANNEL_TYPE: ChannelType = "ibm_cloud"
+_CHANNEL_TYPES = [_DEFAULT_CHANNEL_TYPE, "ibm_quantum"]
 
 
 class AccountManager:
@@ -38,23 +40,24 @@ def save(
         token: Optional[str] = None,
         url: Optional[str] = None,
         instance: Optional[str] = None,
-        auth: Optional[AccountType] = None,
+        channel: Optional[ChannelType] = None,
         name: Optional[str] = _DEFAULT_ACCOUNT_NAME,
         proxies: Optional[ProxyConfiguration] = None,
         verify: Optional[bool] = None,
         overwrite: Optional[bool] = False,
     ) -> None:
         """Save account on disk."""
-        config_key = name or cls._get_default_account_name(auth)
+        cls.migrate()
+        name = name or cls._get_default_account_name(channel)
         return save_config(
             filename=_DEFAULT_ACCOUNT_CONFIG_JSON_FILE,
-            name=config_key,
+            name=name,
             overwrite=overwrite,
             config=Account(
                 token=token,
                 url=url,
                 instance=instance,
-                auth=auth,
+                channel=channel,
                 proxies=proxies,
                 verify=verify,
             )
@@ -65,22 +68,23 @@ def save(
     @staticmethod
     def list(
         default: Optional[bool] = None,
-        auth: Optional[str] = None,
+        channel: Optional[ChannelType] = None,
         name: Optional[str] = None,
     ) -> Dict[str, Account]:
         """List all accounts saved on disk."""
+        AccountManager.migrate()
 
         def _matching_name(account_name: str) -> bool:
             return name is None or name == account_name
 
-        def _matching_auth(account: Account) -> bool:
-            return auth is None or account.auth == auth
+        def _matching_channel(account: Account) -> bool:
+            return channel is None or account.channel == channel
 
         def _matching_default(account_name: str) -> bool:
             default_accounts = [
                 _DEFAULT_ACCOUNT_NAME,
-                _DEFAULT_ACCOUNT_NAME_LEGACY,
-                _DEFAULT_ACCOUNT_NAME_CLOUD,
+                _DEFAULT_ACCOUNT_NAME_IBM_QUANTUM,
+                _DEFAULT_ACCOUNT_NAME_IBM_CLOUD,
             ]
             if default is None:
                 return True
@@ -91,15 +95,18 @@ def _matching_default(account_name: str) -> bool:
 
         # load all accounts
         all_accounts = map(
-            lambda kv: (kv[0], Account.from_saved_format(kv[1])),
+            lambda kv: (
+                kv[0],
+                Account.from_saved_format(kv[1]),
+            ),
             read_config(filename=_DEFAULT_ACCOUNT_CONFIG_JSON_FILE).items(),
         )
 
         # filter based on input parameters
         filtered_accounts = dict(
             list(
                 filter(
-                    lambda kv: _matching_auth(kv[1])
+                    lambda kv: _matching_channel(kv[1])
                     and _matching_default(kv[0])
                     and _matching_name(kv[0]),
                     all_accounts,
@@ -111,20 +118,21 @@ def _matching_default(account_name: str) -> bool:
 
     @classmethod
     def get(
-        cls, name: Optional[str] = None, auth: Optional[AccountType] = None
+        cls, name: Optional[str] = None, channel: Optional[ChannelType] = None
     ) -> Optional[Account]:
         """Read account from disk.
 
         Args:
             name: Account name. Takes precedence if `auth` is also specified.
-            auth: Account auth type.
+            channel: Channel type.
 
         Returns:
             Account information.
 
         Raises:
             AccountNotFoundError: If the input value cannot be found on disk.
         """
+        cls.migrate()
         if name:
             saved_account = read_config(
                 filename=_DEFAULT_ACCOUNT_CONFIG_JSON_FILE, name=name
@@ -135,23 +143,23 @@ def get(
                 )
             return Account.from_saved_format(saved_account)
 
-        auth_ = auth or _DEFAULT_ACCOUNT_TYPE
-        env_account = cls._from_env_variables(auth_)
+        channel_ = channel or _DEFAULT_CHANNEL_TYPE
+        env_account = cls._from_env_variables(channel_)
         if env_account is not None:
             return env_account
 
-        if auth:
+        if channel:
             saved_account = read_config(
                 filename=_DEFAULT_ACCOUNT_CONFIG_JSON_FILE,
-                name=cls._get_default_account_name(auth),
+                name=cls._get_default_account_name(channel=channel),
             )
             if saved_account is None:
-                raise AccountNotFoundError(f"No default {auth} account saved.")
+                raise AccountNotFoundError(f"No default {channel} account saved.")
             return Account.from_saved_format(saved_account)
 
         all_config = read_config(filename=_DEFAULT_ACCOUNT_CONFIG_JSON_FILE)
-        for account_type in _ACCOUNT_TYPES:
-            account_name = cls._get_default_account_name(account_type)
+        for channel_type in _CHANNEL_TYPES:
+            account_name = cls._get_default_account_name(channel_type)
             if account_name in all_config:
                 return Account.from_saved_format(all_config[account_name])
 
@@ -161,30 +169,70 @@ def get(
     def delete(
         cls,
         name: Optional[str] = None,
-        auth: Optional[str] = None,
+        channel: Optional[ChannelType] = None,
     ) -> bool:
         """Delete account from disk."""
+        cls.migrate()
+        name = name or cls._get_default_account_name(channel)
+        return delete_config(name=name, filename=_DEFAULT_ACCOUNT_CONFIG_JSON_FILE)
 
-        config_key = name or cls._get_default_account_name(auth)
-        return delete_config(
-            name=config_key, filename=_DEFAULT_ACCOUNT_CONFIG_JSON_FILE
-        )
+    @classmethod
+    def migrate(cls) -> None:
+        """Migrate accounts on disk by removing `auth` and adding `channel`."""
+        data = read_config(filename=_DEFAULT_ACCOUNT_CONFIG_JSON_FILE)
+        for key, value in data.items():
+            if key == _DEFAULT_ACCOUNT_NAME_CLOUD:
+                value.pop("auth", None)
+                value.update(channel="ibm_cloud")
+                delete_config(filename=_DEFAULT_ACCOUNT_CONFIG_JSON_FILE, name=key)
+                save_config(
+                    filename=_DEFAULT_ACCOUNT_CONFIG_JSON_FILE,
+                    name=_DEFAULT_ACCOUNT_NAME_IBM_CLOUD,
+                    config=value,
+                    overwrite=False,
+                )
+            elif key == _DEFAULT_ACCOUNT_NAME_LEGACY:
+                value.pop("auth", None)
+                value.update(channel="ibm_quantum")
+                delete_config(filename=_DEFAULT_ACCOUNT_CONFIG_JSON_FILE, name=key)
+                save_config(
+                    filename=_DEFAULT_ACCOUNT_CONFIG_JSON_FILE,
+                    name=_DEFAULT_ACCOUNT_NAME_IBM_QUANTUM,
+                    config=value,
+                    overwrite=False,
+                )
+            else:
+                if hasattr(value, "auth"):
+                    if value["auth"] == "cloud":
+                        value.update(channel="ibm_cloud")
+                    elif value["auth"] == "legacy":
+                        value.update(channel="ibm_quantum")
+                    value.pop("auth", None)
+                    save_config(
+                        filename=_DEFAULT_ACCOUNT_CONFIG_JSON_FILE,
+                        name=key,
+                        config=value,
+                        overwrite=True,
+                    )
 
     @classmethod
-    def _from_env_variables(cls, auth: Optional[AccountType]) -> Optional[Account]:
+    def _from_env_variables(cls, channel: Optional[ChannelType]) -> Optional[Account]:
         """Read account from environment variable."""
         token = os.getenv("QISKIT_IBM_TOKEN")
         url = os.getenv("QISKIT_IBM_URL")
         if not (token and url):
             return None
         return Account(
-            token=token, url=url, instance=os.getenv("QISKIT_IBM_INSTANCE"), auth=auth
+            token=token,
+            url=url,
+            instance=os.getenv("QISKIT_IBM_INSTANCE"),
+            channel=channel,
         )
 
     @classmethod
-    def _get_default_account_name(cls, auth: AccountType) -> str:
+    def _get_default_account_name(cls, channel: ChannelType) -> str:
         return (
-            _DEFAULT_ACCOUNT_NAME_LEGACY
-            if auth == "legacy"
-            else _DEFAULT_ACCOUNT_NAME_CLOUD
+            _DEFAULT_ACCOUNT_NAME_IBM_QUANTUM
+            if channel == "ibm_quantum"
+            else _DEFAULT_ACCOUNT_NAME_IBM_CLOUD
         )

qiskit_ibm_provider/accounts/storage.py

@@ -54,7 +54,6 @@ def read_config(
             return data
         if name in data:
             return data[name]
-
         return None