Skip to content

Commit

Permalink
Making account code more generic (#1060)
Browse files Browse the repository at this point in the history 
* Making account code more generic by defining subclasses for channel types

* Removed channel parameter from _assert_valid_instance

* mypy, lint and black

* Changed order of decorators

* Code cleanup

* Documentation fixes

* black

---------

Co-authored-by: Kevin Tian <[email protected]>
  • Loading branch information
@merav-aharoni @kt474
merav-aharoni and kt474 authored Sep 25, 2023
1 parent 86eac07 commit 38ca3b2
Show file tree
Hide file tree
Showing 4 changed files with 199 additions and 92 deletions.
239 changes: 173 additions & 66 deletions qiskit_ibm_runtime/accounts/account.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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"]]
Expand All @@ -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,
Expand All @@ -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
Expand All @@ -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):
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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}'."
)
23 changes: 12 additions & 11 deletions qiskit_ibm_runtime/accounts/management.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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"),
Expand Down Expand Up @@ -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),
Expand Down
5 changes: 2 additions & 3 deletions qiskit_ibm_runtime/qiskit_runtime_service.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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()
Expand Down
Loading

0 comments on commit 38ca3b2

Please sign in to comment.