Improve test coverage and documentation for proxy configuration

qiskit_ibm_runtime/accounts/account.py

@@ -16,9 +16,10 @@
 from typing import Optional
 from urllib.parse import urlparse
 
-from typing_extensions import Literal
 from requests.auth import AuthBase
-
+from typing_extensions import Literal
+from ..proxies import ProxyConfiguration
+from ..utils.hgp import from_instance_format
 from ..api.auth import LegacyAuth, CloudAuth
 
 AccountType = Optional[Literal["cloud", "legacy"]]
@@ -31,15 +32,15 @@ def _assert_valid_auth(auth: AccountType) -> None:
     """Assert that the auth parameter is valid."""
     if not (auth in ["cloud", "legacy"]):
         raise ValueError(
-            f"Inappropriate `auth` value. Expected one of ['cloud', 'legacy'], got '{auth}'."
+            f"Invalid `auth` value. Expected one of ['cloud', 'legacy'], got '{auth}'."
         )
 
 
 def _assert_valid_token(token: str) -> None:
     """Assert that the token is valid."""
     if not (isinstance(token, str) and len(token) > 0):
         raise ValueError(
-            f"Inappropriate `token` value. Expected a non-empty string, got '{token}'."
+            f"Invalid `token` value. Expected a non-empty string, got '{token}'."
         )
 
 
@@ -48,17 +49,30 @@ def _assert_valid_url(url: str) -> None:
     try:
         urlparse(url)
     except:
-        raise ValueError(f"Inappropriate `url` value. Failed to parse '{url}' as URL.")
+        raise ValueError(f"Invalid `url` value. Failed to parse '{url}' as URL.")
+
+
+def _assert_valid_proxies(config: ProxyConfiguration) -> None:
+    """Assert that the proxy configuration is valid."""
+    if config is not None:
+        config.validate()
 
 
 def _assert_valid_instance(auth: AccountType, instance: str) -> None:
     """Assert that the instance name is valid for the given account type."""
     if auth == "cloud":
         if not (isinstance(instance, str) and len(instance) > 0):
             raise ValueError(
-                f"Inappropriate `instance` value. Expected a non-empty string."
+                f"Invalid `instance` value. Expected a non-empty string, got '{instance}'."
             )
-    # TODO: add validation for legacy instance when adding test coverage
+    if auth == "legacy":
+        if instance is not None:
+            try:
+                from_instance_format(instance)
+            except:
+                raise ValueError(
+                    f"Invalid `instance` value. Expected hub/group/project format, got {instance}"
+                )
 
 
 class Account:
@@ -70,8 +84,7 @@ def __init__(
         token: str,
         url: Optional[str] = None,
         instance: Optional[str] = None,
-        # TODO: add validation for proxies input format
-        proxies: Optional[dict] = None,
+        proxies: Optional[ProxyConfiguration] = None,
         verify: Optional[bool] = True,
     ):
         """Account constructor.
@@ -81,7 +94,7 @@ def __init__(
             token: Account token to use.
             url: Authentication URL.
             instance: Service instance to use.
-            proxies: Proxies to use.
+            proxies: Proxy configuration.
             verify: Whether to verify server's TLS certificate.
         """
         _assert_valid_auth(auth)
@@ -94,23 +107,31 @@ def __init__(
         _assert_valid_url(resolved_url)
         self.url = resolved_url
 
+        _assert_valid_instance(auth, instance)
         self.instance = instance
+
+        _assert_valid_proxies(proxies)
         self.proxies = proxies
+
         self.verify = verify
 
     def to_saved_format(self) -> dict:
         """Returns a dictionary that represents how the account is saved on disk."""
-        return {k: v for k, v in self.__dict__.items() if v is not None}
+        result = {k: v for k, v in self.__dict__.items() if v is not None}
+        if self.proxies:
+            result["proxies"] = self.proxies.to_dict()
+        return result
 
     @classmethod
     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"),
             url=data.get("url"),
             token=data.get("token"),
             instance=data.get("instance"),
-            proxies=data.get("proxies"),
+            proxies=ProxyConfiguration(**proxies) if proxies else None,
             verify=data.get("verify", True),
         )
 

qiskit_ibm_runtime/accounts/management.py

@@ -16,6 +16,7 @@
 from typing import Optional, Dict
 
 from .account import Account, AccountType
+from ..proxies import ProxyConfiguration
 from .storage import save_config, read_config, delete_config
 
 _DEFAULT_ACCOUNT_CONFIG_JSON_FILE = os.path.join(
@@ -39,7 +40,7 @@ def save(
         instance: Optional[str] = None,
         auth: Optional[AccountType] = None,
         name: Optional[str] = _DEFAULT_ACCOUNT_NAME,
-        proxies: Optional[dict] = None,
+        proxies: Optional[ProxyConfiguration] = None,
         verify: Optional[bool] = None,
     ) -> None:
         """Save account on disk."""

qiskit_ibm_runtime/api/client_parameters.py

@@ -14,9 +14,8 @@
 
 from typing import Dict, Optional, Any, Union
 
-from requests_ntlm import HttpNtlmAuth
-
 from ..api.auth import LegacyAuth, CloudAuth
+from ..proxies import ProxyConfiguration
 
 TEMPLATE_IBM_HUBS = "{prefix}/Network/{hub}/Groups/{group}/Projects/{project}"
 """str: Template for creating an IBM Quantum URL with hub/group/project information."""
@@ -31,7 +30,7 @@ def __init__(
         token: str,
         url: str = None,
         instance: Optional[str] = None,
-        proxies: Optional[Dict] = None,
+        proxies: Optional[ProxyConfiguration] = None,
         verify: bool = True,
     ) -> None:
         """ClientParameters constructor.
@@ -64,15 +63,9 @@ def connection_parameters(self) -> Dict[str, Any]:
             expected by ``requests``. The following keys can be present:
             ``proxies``, ``verify``, and ``auth``.
         """
-        request_kwargs = {"verify": self.verify}
+        request_kwargs: Any = {"verify": self.verify}
 
         if self.proxies:
-            if "urls" in self.proxies:
-                request_kwargs["proxies"] = self.proxies["urls"]
-
-            if "username_ntlm" in self.proxies and "password_ntlm" in self.proxies:
-                request_kwargs["auth"] = HttpNtlmAuth(
-                    self.proxies["username_ntlm"], self.proxies["password_ntlm"]
-                )
+            request_kwargs.update(self.proxies.to_request_params())
 
         return request_kwargs

qiskit_ibm_runtime/api/clients/utils.py

@@ -61,8 +61,8 @@ def ws_proxy_params(client_params: ClientParameters, ws_url: str) -> Dict:
 
     if "auth" in conn_data:
         out["http_proxy_auth"] = (
-            client_params.proxies["username_ntlm"],
-            client_params.proxies["password_ntlm"],
+            client_params.proxies.username_ntlm,
+            client_params.proxies.password_ntlm,
         )
 
     return out

qiskit_ibm_runtime/ibm_runtime_service.py

@@ -24,9 +24,10 @@
 from qiskit.providers.exceptions import QiskitBackendNotFoundError
 from qiskit.providers.providerutils import filter_backends
 
-from qiskit_ibm_runtime import ibm_backend  # pylint: disable=unused-import
+from qiskit_ibm_runtime import ibm_backend
 from .accounts import AccountManager, Account, AccountType
 from .accounts.exceptions import AccountsError
+from .proxies import ProxyConfiguration
 from .api.clients import AuthClient, VersionClient
 from .api.clients.runtime import RuntimeClient
 from .api.exceptions import RequestsApiError
@@ -143,7 +144,11 @@ def __init__(
             name: Name of the account to load.
             instance: The service instance to use. For cloud runtime, this is the Cloud Resource
                 Name (CRN). For legacy runtime, this is the hub/group/project in that format.
-            proxies: Proxy configuration for the server.
+            proxies: Proxy configuration. Supported optional keys are
+                ``urls`` (a dictionary mapping protocol or protocol and host to the URL of the proxy,
+                documented at https://docs.python-requests.org/en/latest/api/#requests.Session.proxies),
+                ```username_ntlm```, ```password_ntlm```(username and password to enable NTLM user
+                authentication)
             verify: Whether to verify the server's TLS certificate.
 
         Returns:
@@ -160,7 +165,7 @@ def __init__(
             instance=instance,
             auth=auth,
             name=name,
-            proxies=proxies,
+            proxies=ProxyConfiguration(**proxies) if proxies else None,
             verify=verify,
         )
         if self._account.auth == "cloud" and not self._account.instance:
@@ -211,7 +216,7 @@ def _discover_account(
         instance: Optional[str] = None,
         auth: Optional[AccountType] = None,
         name: Optional[str] = None,
-        proxies: Optional[dict] = None,
+        proxies: Optional[ProxyConfiguration] = None,
         verify: Optional[bool] = None,
     ) -> Account:
         """Discover account."""
@@ -549,7 +554,11 @@ def save_account(
             instance: The CRN (cloud) or hub/group/project (legacy).
             auth: Authentication type. `cloud` or `legacy`.
             name: Name of the account to save.
-            proxies: Proxy configuration for the server.
+            proxies: Proxy configuration. Supported optional keys are
+                ``urls`` (a dictionary mapping protocol or protocol and host to the URL of the proxy,
+                documented at https://docs.python-requests.org/en/latest/api/#requests.Session.proxies),
+                ```username_ntlm```, ```password_ntlm```(username and password to enable NTLM user
+                authentication)
             verify: Verify the server's TLS certificate.
         """
 
@@ -559,7 +568,7 @@ def save_account(
             instance=instance,
             auth=auth,
             name=name,
-            proxies=proxies,
+            proxies=ProxyConfiguration(**proxies) if proxies else None,
             verify=verify,
         )
 

qiskit_ibm_runtime/proxies/__init__.py

@@ -0,0 +1,17 @@
+# This code is part of Qiskit.
+#
+# (C) Copyright IBM 2022.
+#
+# This code is licensed under the Apache License, Version 2.0. You may
+# obtain a copy of this license in the LICENSE.txt file in the root directory
+# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
+#
+# Any modifications or derivative works of this code must retain this
+# copyright notice, and modified files need to carry a notice indicating
+# that they have been altered from the originals.
+
+"""
+Proxy configuration.
+"""
+
+from .configuration import ProxyConfiguration

qiskit_ibm_runtime/proxies/configuration.py

@@ -0,0 +1,84 @@
+# This code is part of Qiskit.
+#
+# (C) Copyright IBM 2022.
+#
+# This code is licensed under the Apache License, Version 2.0. You may
+# obtain a copy of this license in the LICENSE.txt file in the root directory
+# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
+#
+# Any modifications or derivative works of this code must retain this
+# copyright notice, and modified files need to carry a notice indicating
+# that they have been altered from the originals.
+
+"""Proxy related classes and functions."""
+
+
+from dataclasses import dataclass
+from typing import Optional, Dict
+from requests_ntlm import HttpNtlmAuth
+
+
+@dataclass
+class ProxyConfiguration:
+    """Class for representing a proxy configuration.
+
+    Args
+        urls: a dictionary mapping protocol or protocol and host to the URL of the proxy. Refer to
+            https://docs.python-requests.org/en/latest/api/#requests.Session.proxies for details.
+        username_ntlm: username used to enable NTLM user authentication.
+        password_ntlm: password used to enable NTLM user authentication.
+    """
+
+    urls: Optional[Dict[str, str]] = None
+    username_ntlm: Optional[str] = None
+    password_ntlm: Optional[str] = None
+
+    def validate(self) -> None:
+        """Validate configuration.
+
+        Raises:
+            ValueError: If configuration is invalid.
+        """
+        if not any(
+            [
+                isinstance(self.username_ntlm, str)
+                and isinstance(self.password_ntlm, str),
+                self.username_ntlm is None and self.password_ntlm is None,
+            ]
+        ):
+            raise ValueError(
+                f"Invalid proxy configuration for NTLM authentication. None or both of username and "
+                f"password must be provided. Got username_ntlm={self.username_ntlm}, "
+                f"password_ntlm={self.password_ntlm}."
+            )
+
+        if self.urls is not None and not isinstance(self.urls, dict):
+            raise ValueError(
+                f"Invalid proxy configuration. Expected `urls` to contain a dictionary mapping protocol "
+                f"or protocol and host to the URL of the proxy. Got {self.urls}"
+            )
+
+    def to_dict(self) -> dict:
+        """Transform configuration to dictionary."""
+
+        return {k: v for k, v in self.__dict__.items() if v is not None}
+
+    def to_request_params(self) -> dict:
+        """Transform configuration to request parameters.
+
+        Returns:
+            A dictionary with proxy configuration parameters in the format
+            expected by ``requests``. The following keys can be present:
+            ``proxies``and ``auth``.
+        """
+
+        request_kwargs = {}
+        if self.urls:
+            request_kwargs["proxies"] = self.urls
+
+        if self.username_ntlm and self.password_ntlm:
+            request_kwargs["auth"] = HttpNtlmAuth(
+                self.username_ntlm, self.password_ntlm
+            )
+
+        return request_kwargs