Making account code more generic

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}'."
+            )

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),

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()