Skip to content
Merged
Show file tree
Hide file tree
Changes from 5 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 15 additions & 1 deletion qiskit_ibm_runtime/accounts/account.py
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,7 @@ def __init__(
self,
auth: AccountType,
token: str,
url: Optional[str],
url: Optional[str] = None,
instance: Optional[str] = None,
# TODO: add validation for proxies input format
proxies: Optional[dict] = None,
Expand Down Expand Up @@ -120,3 +120,17 @@ def get_auth_handler(self) -> AuthBase:
return CloudAuth(api_key=self.token, crn=self.instance)

return LegacyAuth(access_token=self.token)

def __eq__(self, other: object) -> bool:
if not isinstance(other, Account):
return False
return all(
[
self.auth == other.auth,
self.token == other.token,
self.url == other.url,
self.instance == other.instance,
self.proxies == other.proxies,
self.verify == other.verify,
]
)
73 changes: 62 additions & 11 deletions qiskit_ibm_runtime/accounts/management.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@
"""Account management related classes and functions."""

import os
from typing import Optional, Union
from typing import Optional, Dict

from .account import Account, AccountType
from .storage import save_config, read_config, delete_config
Expand All @@ -31,8 +31,9 @@
class AccountManager:
"""Class that bundles account management related functionality."""

@staticmethod
@classmethod
def save(
cls,
token: Optional[str] = None,
url: Optional[str] = None,
instance: Optional[str] = None,
Expand All @@ -43,9 +44,10 @@ def save(
) -> None:
"""Save account on disk."""

config_key = name or cls._get_default_account_name(auth)
return save_config(
filename=_DEFAULT_ACCOUNG_CONFIG_JSON_FILE,
name=name,
name=config_key,
config=Account(
token=token,
url=url,
Expand All @@ -57,10 +59,51 @@ def save(
)

@staticmethod
def list() -> Union[dict, None]:
def list(
default: Optional[bool] = None,
auth: Optional[str] = None,
name: Optional[str] = None,
) -> Dict[str, Account]:
"""List all accounts saved on disk."""

return read_config(filename=_DEFAULT_ACCOUNG_CONFIG_JSON_FILE)
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_default(account_name: str) -> bool:
default_accounts = [
_DEFAULT_ACCOUNT_NAME,
_DEFAULT_ACCOUNT_NAME_LEGACY,
_DEFAULT_ACCOUNT_NAME_CLOUD,
]
if default is None:
return True
elif default is False:
return account_name not in default_accounts
else:
return account_name in default_accounts

# load all accounts
all_accounts = map(
lambda kv: (kv[0], Account.from_saved_format(kv[1])),
read_config(filename=_DEFAULT_ACCOUNG_CONFIG_JSON_FILE).items(),
)

# filter based on input parameters
filtered_accounts = dict(
list(
filter(
lambda kv: _matching_auth(kv[1])
and _matching_default(kv[0])
and _matching_name(kv[0]),
all_accounts,
)
)
)

return filtered_accounts

@classmethod
def get(
Expand Down Expand Up @@ -110,10 +153,18 @@ def get(

return None

@staticmethod
def delete(name: Optional[str] = _DEFAULT_ACCOUNT_NAME) -> bool:
@classmethod
def delete(
cls,
name: Optional[str] = None,
auth: Optional[str] = None,
) -> bool:
"""Delete account from disk."""
return delete_config(name=name, filename=_DEFAULT_ACCOUNG_CONFIG_JSON_FILE)

config_key = name or cls._get_default_account_name(auth)
return delete_config(
name=config_key, filename=_DEFAULT_ACCOUNG_CONFIG_JSON_FILE
)

@classmethod
def _from_env_variables(cls, auth: Optional[AccountType]) -> Optional[Account]:
Expand All @@ -129,7 +180,7 @@ def _from_env_variables(cls, auth: Optional[AccountType]) -> Optional[Account]:
@classmethod
def _get_default_account_name(cls, auth: AccountType) -> str:
return (
_DEFAULT_ACCOUNT_NAME_CLOUD
if auth == "cloud"
else _DEFAULT_ACCOUNT_NAME_LEGACY
_DEFAULT_ACCOUNT_NAME_LEGACY
if auth == "legacy"
else _DEFAULT_ACCOUNT_NAME_CLOUD
)
10 changes: 8 additions & 2 deletions qiskit_ibm_runtime/accounts/storage.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,17 +13,20 @@
"""Utility functions related to storing account configuration on disk."""

import json
import logging
import os
from typing import Optional, Dict

logger = logging.getLogger(__name__)


def save_config(
filename: str,
name: str,
config: dict,
) -> 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)

with open(filename, mode="r") as json_in:
Expand All @@ -39,7 +42,7 @@ def read_config(
name: Optional[str] = None,
) -> Optional[Dict]:
"""Save configuration data from a JSON file."""
Comment thread
rathishcholarajan marked this conversation as resolved.
Outdated

logger.debug("Read configuration data for '%s' from '%s'", name, filename)
_ensure_file_exists(filename)

with open(filename) as json_file:
Expand All @@ -58,6 +61,8 @@ def delete_config(
) -> bool:
"""Delete configuration data from a JSON file."""

logger.debug("Delete configuration data for '%s' from '%s'", name, filename)

_ensure_file_exists(filename)
with open(filename, mode="r") as json_in:
data = json.load(json_in)
Expand All @@ -73,6 +78,7 @@ def delete_config(

def _ensure_file_exists(filename: str, initial_content: str = "{}") -> None:
if not os.path.isfile(filename):
logger.debug("Create empty configuration file at %s", filename)

# create parent directories
os.makedirs(os.path.dirname(filename), exist_ok=True)
Expand Down
31 changes: 24 additions & 7 deletions qiskit_ibm_runtime/ibm_runtime_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -517,18 +517,20 @@ def active_account(self) -> Optional[Dict[str, str]]:
return self._account.to_saved_format()

@staticmethod
def delete_account(name: Optional[str]) -> bool:
def delete_account(name: Optional[str] = None, auth: Optional[str] = None) -> bool:
"""Delete a saved account from disk.

Args:
name: Custom name of the saved account. Defaults to "default".
name: Name of the saved account to delete.
auth: Authentication type of the default account to delete.
Ignored if account name is provided.

Returns:
True if the account with the given name was deleted.
False if no account was found for the given name.
True if the account was deleted.
False if no account was found.
"""

return AccountManager.delete(name=name)
return AccountManager.delete(name=name, auth=auth)

@staticmethod
def save_account(
Expand Down Expand Up @@ -565,17 +567,32 @@ def save_account(
)

@staticmethod
def saved_accounts() -> dict:
def saved_accounts(
default: Optional[bool] = None,
auth: Optional[str] = None,
name: Optional[str] = None,
) -> dict:
"""List the accounts saved on disk.

Args:
default: If set to True, only default accounts are returned.
auth: If set, only accounts with the given authentication type are returned.
name: If set, only accounts with the given name are returned.

Returns:
A dictionary with information about the accounts saved on disk.

Raises:
IBMProviderCredentialsInvalidUrl: If invalid IBM Quantum
credentials are found on disk.
"""
return AccountManager.list()

return dict(
map(
lambda kv: (kv[0], Account.to_saved_format(kv[1])),
AccountManager.list(default=default, auth=auth, name=name).items(),
),
)

def get_backend(
self,
Expand Down
Loading