diff --git a/qiskit_ibm_runtime/accounts/account.py b/qiskit_ibm_runtime/accounts/account.py index b28fd52e7..7e71b95dc 100644 --- a/qiskit_ibm_runtime/accounts/account.py +++ b/qiskit_ibm_runtime/accounts/account.py @@ -12,6 +12,7 @@ """Account related classes and functions.""" +from abc import abstractmethod import logging from typing import Optional, Literal from urllib.parse import urlparse @@ -22,7 +23,6 @@ from .exceptions import InvalidAccountError, CloudResourceNameResolutionError from ..api.auth import QuantumAuth, CloudAuth - from ..utils import resolve_crn AccountType = Optional[Literal["cloud", "legacy"]] @@ -34,13 +34,11 @@ class Account: - """Class that represents an account.""" + """Class that represents an account. This is an abstract class.""" def __init__( self, - channel: ChannelType, token: str, - url: Optional[str] = None, instance: Optional[str] = None, proxies: Optional[ProxyConfiguration] = None, verify: Optional[bool] = True, @@ -57,12 +55,9 @@ def __init__( verify: Whether to verify server's TLS certificate. channel_strategy: Error mitigation strategy. """ - resolved_url = url or ( - IBM_QUANTUM_API_URL if channel == "ibm_quantum" else IBM_CLOUD_API_URL - ) - self.channel = channel + self.channel: str = None + self.url: str = None self.token = token - self.url = resolved_url self.instance = instance self.proxies = proxies self.verify = verify @@ -78,56 +73,65 @@ def to_saved_format(self) -> dict: @classmethod def from_saved_format(cls, data: dict) -> "Account": """Creates an account instance from data saved on disk.""" + channel = data.get("channel") proxies = data.get("proxies") - return cls( - channel=data.get("channel"), - url=data.get("url"), - token=data.get("token"), - instance=data.get("instance"), - proxies=ProxyConfiguration(**proxies) if proxies else None, - verify=data.get("verify", True), - channel_strategy=data.get("channel_strategy"), + proxies = ProxyConfiguration(**proxies) if proxies else None + url = data.get("url") + token = data.get("token") + instance = data.get("instance") + verify = data.get("verify", True) + channel_strategy = data.get("channel_strategy") + return cls.create_account( + channel=channel, + url=url, + token=token, + instance=instance, + proxies=proxies, + verify=verify, + channel_strategy=channel_strategy, ) + @classmethod + def create_account( + cls, + channel: str, + token: str, + url: Optional[str] = None, + instance: Optional[str] = None, + proxies: Optional[ProxyConfiguration] = None, + verify: Optional[bool] = True, + channel_strategy: Optional[str] = None, + ) -> "Account": + """Creates an account for a specific channel.""" + if channel == "ibm_quantum": + return QuantumAccount( + url=url, + token=token, + instance=instance, + proxies=proxies, + verify=verify, + channel_strategy=channel_strategy, + ) + elif channel == "ibm_cloud": + return CloudAccount( + url=url, + token=token, + instance=instance, + proxies=proxies, + verify=verify, + channel_strategy=channel_strategy, + ) + else: + raise InvalidAccountError( + f"Invalid `channel` value. Expected one of " + f"{['ibm_cloud', 'ibm_quantum']}, got '{channel}'." + ) + def resolve_crn(self) -> None: """Resolves the corresponding unique Cloud Resource Name (CRN) for the given non-unique service instance name and updates the ``instance`` attribute accordingly. - - No-op if ``channel`` attribute is set to ``ibm_quantum``. - No-op if ``instance`` attribute is set to a Cloud Resource Name (CRN). - - Raises: - CloudResourceNameResolutionError: if CRN value cannot be resolved. - """ - if self.channel == "ibm_cloud": - crn = resolve_crn( - channel=self.channel, - url=self.url, - token=self.token, - instance=self.instance, - ) - if len(crn) == 0: - raise CloudResourceNameResolutionError( - f"Failed to resolve CRN value for the provided service name {self.instance}." - ) - if len(crn) > 1: - # handle edge-case where multiple service instances with the same name exist - logger.warning( - "Multiple CRN values found for service name %s: %s. Using %s.", - self.instance, - crn, - crn[0], - ) - - # overwrite with CRN value - self.instance = crn[0] - - def get_auth_handler(self) -> AuthBase: - """Returns the respective authentication handler.""" - if self.channel == "ibm_cloud": - return CloudAuth(api_key=self.token, crn=self.instance) - - return QuantumAuth(access_token=self.token) + Relevant for "ibm_cloud" channel only.""" + pass def __eq__(self, other: object) -> bool: if not isinstance(other, Account): @@ -156,7 +160,7 @@ def validate(self) -> "Account": self._assert_valid_channel(self.channel) self._assert_valid_token(self.token) self._assert_valid_url(self.url) - self._assert_valid_instance(self.channel, self.instance) + self._assert_valid_instance(self.instance) self._assert_valid_proxies(self.proxies) self._assert_valid_channel_strategy(self.channel_strategy) return self @@ -204,18 +208,121 @@ def _assert_valid_proxies(config: ProxyConfiguration) -> None: config.validate() @staticmethod - def _assert_valid_instance(channel: ChannelType, instance: str) -> None: + @abstractmethod + def _assert_valid_instance(instance: str) -> None: """Assert that the instance name is valid for the given account type.""" - if channel == "ibm_cloud": - if not (isinstance(instance, str) and len(instance) > 0): + pass + + +class QuantumAccount(Account): + """Class that represents an account with channel 'ibm_quantum.'""" + + def __init__( + self, + token: str, + url: Optional[str] = None, + instance: Optional[str] = None, + proxies: Optional[ProxyConfiguration] = None, + verify: Optional[bool] = True, + channel_strategy: Optional[str] = None, + ): + """Account constructor. + + Args: + token: Account token to use. + url: Authentication URL. + instance: Service instance to use. + proxies: Proxy configuration. + verify: Whether to verify server's TLS certificate. + channel_strategy: Error mitigation strategy. + """ + super().__init__(token, instance, proxies, verify, channel_strategy) + resolved_url = url or IBM_QUANTUM_API_URL + self.channel = "ibm_quantum" + self.url = resolved_url + + def get_auth_handler(self) -> AuthBase: + """Returns the Quantum authentication handler.""" + return QuantumAuth(access_token=self.token) + + @staticmethod + def _assert_valid_instance(instance: str) -> None: + """Assert that the instance name is valid for the given account type.""" + if instance is not None: + try: + from_instance_format(instance) + except: raise InvalidAccountError( - f"Invalid `instance` value. Expected a non-empty string, got '{instance}'." + f"Invalid `instance` value. Expected hub/group/project format, got {instance}" ) - if channel == "ibm_quantum": - if instance is not None: - try: - from_instance_format(instance) - except: - raise InvalidAccountError( - f"Invalid `instance` value. Expected hub/group/project format, got {instance}" - ) + + +class CloudAccount(Account): + """Class that represents an account with channel 'ibm_cloud'.""" + + def __init__( + self, + token: str, + url: Optional[str] = None, + instance: Optional[str] = None, + proxies: Optional[ProxyConfiguration] = None, + verify: Optional[bool] = True, + channel_strategy: Optional[str] = None, + ): + """Account constructor. + + Args: + token: Account token to use. + url: Authentication URL. + instance: Service instance to use. + proxies: Proxy configuration. + verify: Whether to verify server's TLS certificate. + channel_strategy: Error mitigation strategy. + """ + super().__init__(token, instance, proxies, verify, channel_strategy) + resolved_url = url or IBM_CLOUD_API_URL + self.channel = "ibm_cloud" + self.url = resolved_url + + def get_auth_handler(self) -> AuthBase: + """Returns the Cloud authentication handler.""" + return CloudAuth(api_key=self.token, crn=self.instance) + + def resolve_crn(self) -> None: + """Resolves the corresponding unique Cloud Resource Name (CRN) for the given non-unique service + instance name and updates the ``instance`` attribute accordingly. + + No-op if ``instance`` attribute is set to a Cloud Resource Name (CRN). + + Raises: + CloudResourceNameResolutionError: if CRN value cannot be resolved. + """ + crn = resolve_crn( + channel="ibm_cloud", + url=self.url, + token=self.token, + instance=self.instance, + ) + if len(crn) == 0: + raise CloudResourceNameResolutionError( + f"Failed to resolve CRN value for the provided service name {self.instance}." + ) + if len(crn) > 1: + # handle edge-case where multiple service instances with the same name exist + logger.warning( + "Multiple CRN values found for service name %s: %s. Using %s.", + self.instance, + crn, + crn[0], + ) + + # overwrite with CRN value + self.instance = crn[0] + + @staticmethod + def _assert_valid_instance(instance: str) -> None: + """Assert that the instance name is valid for the given account type.""" + if not (isinstance(instance, str) and len(instance) > 0): + raise InvalidAccountError( + f"Invalid `instance` value. Expected a non-empty string, got '{instance}'." + ) diff --git a/qiskit_ibm_runtime/accounts/management.py b/qiskit_ibm_runtime/accounts/management.py index fd65c6e9b..34eab1dd8 100644 --- a/qiskit_ibm_runtime/accounts/management.py +++ b/qiskit_ibm_runtime/accounts/management.py @@ -55,19 +55,20 @@ def save( name = name or cls._get_default_account_name(channel) filename = filename if filename else _DEFAULT_ACCOUNT_CONFIG_JSON_FILE filename = os.path.expanduser(filename) + config = Account.create_account( + channel=channel, + token=token, + url=url, + instance=instance, + proxies=proxies, + verify=verify, + channel_strategy=channel_strategy, + ) return save_config( filename=filename, name=name, overwrite=overwrite, - config=Account( - token=token, - url=url, - instance=instance, - channel=channel, - proxies=proxies, - verify=verify, - channel_strategy=channel_strategy, - ) + config=config # avoid storing invalid accounts .validate().to_saved_format(), set_as_default=set_as_default, @@ -221,7 +222,7 @@ def _from_env_variables(cls, channel: Optional[ChannelType]) -> Optional[Account url = os.getenv("QISKIT_IBM_URL") if not (token and url): return None - return Account( + return Account.create_account( token=token, url=url, instance=os.getenv("QISKIT_IBM_INSTANCE"), @@ -277,7 +278,7 @@ def _from_qiskitrc_file(cls) -> Optional[Account]: filename=_DEFAULT_ACCOUNT_CONFIG_JSON_FILE, name=_DEFAULT_ACCOUNT_NAME_IBM_QUANTUM, overwrite=False, - config=Account( + config=Account.create_account( token=qiskitrc_data.get("token", None), url=qiskitrc_data.get("url", None), instance=qiskitrc_data.get("default_provider", None), diff --git a/qiskit_ibm_runtime/qiskit_runtime_service.py b/qiskit_ibm_runtime/qiskit_runtime_service.py index a3d4bc146..3b087a81a 100644 --- a/qiskit_ibm_runtime/qiskit_runtime_service.py +++ b/qiskit_ibm_runtime/qiskit_runtime_service.py @@ -269,7 +269,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'") if token: - account = Account( + account = Account.create_account( channel=channel, token=token, url=url, @@ -300,8 +300,7 @@ def _discover_account( account.verify = verify # resolve CRN if needed - if account.channel == "ibm_cloud": - self._resolve_crn(account) + self._resolve_crn(account) # ensure account is valid, fail early if not account.validate() diff --git a/test/unit/test_account.py b/test/unit/test_account.py index d389123f9..409166714 100644 --- a/test/unit/test_account.py +++ b/test/unit/test_account.py @@ -42,14 +42,14 @@ custom_envs, ) -_TEST_IBM_QUANTUM_ACCOUNT = Account( +_TEST_IBM_QUANTUM_ACCOUNT = Account.create_account( channel="ibm_quantum", token="token-x", url="https://auth.quantum-computing.ibm.com/api", instance="ibm-q/open/main", ) -_TEST_IBM_CLOUD_ACCOUNT = Account( +_TEST_IBM_CLOUD_ACCOUNT = Account.create_account( channel="ibm_cloud", token="token-y", url="https://cloud.ibm.com", @@ -80,7 +80,7 @@ def test_invalid_channel(self): with self.assertRaises(InvalidAccountError) as err: invalid_channel: Any = "phantom" - Account( + Account.create_account( channel=invalid_channel, token=self.dummy_token, url=self.dummy_ibm_cloud_url, @@ -94,7 +94,7 @@ def test_invalid_token(self): for token in invalid_tokens: with self.subTest(token=token): with self.assertRaises(InvalidAccountError) as err: - Account( + Account.create_account( channel="ibm_cloud", token=token, url=self.dummy_ibm_cloud_url, @@ -110,7 +110,7 @@ def test_invalid_url(self): for params in subtests: with self.subTest(params=params): with self.assertRaises(InvalidAccountError) as err: - Account(**params, token=self.dummy_token).validate() + Account.create_account(**params, token=self.dummy_token).validate() self.assertIn("Invalid `url` value.", str(err.exception)) def test_invalid_instance(self): @@ -124,7 +124,7 @@ def test_invalid_instance(self): for params in subtests: with self.subTest(params=params): with self.assertRaises(InvalidAccountError) as err: - Account( + Account.create_account( **params, token=self.dummy_token, url=self.dummy_ibm_cloud_url ).validate() self.assertIn("Invalid `instance` value.", str(err.exception)) @@ -137,7 +137,7 @@ def test_invalid_channel_strategy(self): for params in subtests: with self.subTest(params=params): with self.assertRaises(InvalidAccountError) as err: - Account( + Account.create_account( **params, token=self.dummy_token, url=self.dummy_ibm_cloud_url, @@ -162,7 +162,7 @@ def test_invalid_proxy_config(self): for params in subtests: with self.subTest(params=params): with self.assertRaises(ValueError) as err: - Account( + Account.create_account( **params, channel="ibm_quantum", token=self.dummy_token, @@ -308,11 +308,11 @@ def test_list(self): contents={ "key1": _TEST_IBM_CLOUD_ACCOUNT.to_saved_format(), "key2": _TEST_IBM_QUANTUM_ACCOUNT.to_saved_format(), - _DEFAULT_ACCOUNT_NAME_IBM_CLOUD: Account( - "ibm_cloud", "token-ibm-cloud", instance="crn:123" + _DEFAULT_ACCOUNT_NAME_IBM_CLOUD: Account.create_account( + channel="ibm_cloud", token="token-ibm-cloud", instance="crn:123" ).to_saved_format(), - _DEFAULT_ACCOUNT_NAME_IBM_QUANTUM: Account( - "ibm_quantum", "token-ibm-quantum" + _DEFAULT_ACCOUNT_NAME_IBM_QUANTUM: Account.create_account( + channel="ibm_quantum", token="token-ibm-quantum" ).to_saved_format(), } ), self.subTest("filtered list of accounts"):