Fixed login persist to file error

solvebio/__init__.py

@@ -13,6 +13,7 @@
 
 import os as _os
 import logging as _logging
+from typing import Literal
 from .help import open_help as _open_help
 
 # Capture warnings (specifically from urllib3)
@@ -149,8 +150,8 @@ def login(
                 api_key=YOUR_API_KEY
             )
     """
-    token_type = None
-    token = None
+    token_type: Literal["Bearer", "Token"] = None
+    token: str = None
 
     if access_token:
         token_type = "Bearer"
@@ -160,8 +161,8 @@ def login(
         token = api_key
 
     if api_host or token or debug:
-        client._host, client._auth = authenticate(
-            api_host, token, token_type=token_type, debug=debug
+        client.set_credentials(
+            api_host, token, token_type=token_type, raise_on_missing=not debug, debug=debug
         )
 
     client.set_user_agent(name=name, version=version)

solvebio/auth.py

@@ -104,7 +104,7 @@ def authenticate(
             if token:
                 source_token = 'creds'
 
-    if (not host and raise_on_missing) or debug:
+    if debug:
         # this will tell the user where QB Client found the credentials from
         creds_path = netrc.path()
         print('\n'.join([
@@ -127,8 +127,11 @@ def authenticate(
             "   QUARTZBIO_API_KEY",
         ]))
 
-        if not debug:
-            raise SolveError("No SolveBio API host is set")
+    if not host:
+        if raise_on_missing and not debug:
+            raise SolveError("No QuartzBio API host is set")
+        else:
+            return host, None
 
     host = validate_api_host_url(host)
 

solvebio/cli/auth.py

@@ -25,6 +25,18 @@ def login_and_save_credentials(*args):
     )
 
     # Print information about the current user
+    if not client.is_logged_in():
+        print("login: client is not logged in!")
+
+        # Verify if user has provided the wrong credentials file
+        if client._host:
+            if suggested_host := client.validate_host_is_www_url(client._host):
+                print(
+                    f"Provided API host is: `{client._host}`. "
+                    f"Did you perhaps mean `{suggested_host}`?"
+                )
+        return
+
     user = client.whoami()
     print_user(user)
 

solvebio/client.py

@@ -5,6 +5,7 @@
 import time
 import inspect
 import os
+from typing import Literal
 
 import solvebio
 
@@ -84,59 +85,30 @@ class SolveClient(object):
     _host: str = None
     _auth: SolveBioTokenAuth = None
 
-    def __init__(self, host=None, token=None, token_type='Token',
-                 include_resources=True, retry_all=None):
-        self._host, self._auth = authenticate(host, token, token_type)
+    def __init__(
+        self,
+        host=None,
+        token=None,
+        token_type: Literal["Bearer", "Token"] = "Token",
+        include_resources=True,
+        retry_all: bool = None,
+    ):
+        self._host: str = None
+        self._auth: SolveBioTokenAuth = None
+        self._session: Session = None
+
         self._headers = {
             'Content-Type': 'application/json',
             'Accept': 'application/json',
             'Accept-Encoding': 'gzip,deflate'
         }
-        self.set_user_agent()
-
-        if retry_all is None:
-            retry_all = bool(os.environ.get("SOLVEBIO_RETRY_ALL"))
-
-        if bool(retry_all):
-            logger.info("Retries enabled for all API requests")
-            allowed_methods = frozenset([
-                "HEAD",
-                "GET",
-                "PUT",
-                "POST",
-                "PATCH",
-                "DELETE",
-                "OPTIONS",
-                "TRACE"
-            ])
-
-            retries = Retry(
-                total=5,
-                backoff_factor=2,
-                status_forcelist=[
-                    codes.bad_gateway,
-                    codes.service_unavailable,
-                    codes.gateway_timeout
-                ],
-                allowed_methods=allowed_methods
-            )
-        else:
-            logger.info("Retries enabled for read-only API requests")
-            retries = Retry(
-                total=5,
-                backoff_factor=2,
-                status_forcelist=[
-                    codes.bad_gateway,
-                    codes.service_unavailable,
-                    codes.gateway_timeout
-                ]
-            )
+        self.retry_all = bool(retry_all)
+        if self.retry_all is None:
+            self.retry_all = bool(os.environ.get("SOLVEBIO_RETRY_ALL"))
 
-        # Use a session with a retry policy to handle
-        # intermittent connection errors.
-        adapter = adapters.HTTPAdapter(max_retries=retries)
-        self._session = Session()
-        self._session.mount(self._host, adapter)
+        # this class is created before any commands, so it shouldn't raise a missing host exception
+        self.set_credentials(host, token, token_type, raise_on_missing=False)
+        self.set_user_agent()
 
         # Import all resources into the client
         if include_resources:
@@ -166,6 +138,44 @@ def set_user_agent(self, name=None, version=None):
 
         self._headers['User-Agent'] = ua
 
+    def set_credentials(
+        self, host: str, token: str, token_type: Literal["Bearer", "Token"],
+        *, debug: bool = False, raise_on_missing: bool = True
+    ):
+        self._host, self._auth = authenticate(
+            host, token, token_type, debug=debug, raise_on_missing=raise_on_missing
+        )
+
+        if self._host:
+            retry_kwargs = {}
+            if self.retry_all:
+                logger.info("Retries enabled for all API requests")
+                retry_kwargs["allow_redirects"] = frozenset(
+                    ["HEAD", "GET", "PUT", "POST", "PATCH", "DELETE", "OPTIONS", "TRACE"]
+                )
+            else:
+                logger.info("Retries enabled for read-only API requests")
+
+            retries = Retry(
+                total=5,
+                backoff_factor=2,
+                status_forcelist=[
+                    codes.bad_gateway,
+                    codes.service_unavailable,
+                    codes.gateway_timeout,
+                ],
+                **retry_kwargs
+            )
+
+            # Use a session with a retry policy to handle
+            # intermittent connection errors.
+            adapter = adapters.HTTPAdapter(max_retries=retries)
+            self._session = Session()
+            self._session.mount(self._host, adapter)
+
+    def is_logged_in(self):
+        return bool(self._host and self._auth and self._session)
+
     def whoami(self):
         return self.get('/v1/user', {})
 
@@ -232,6 +242,8 @@ def request(self, method, url, **kwargs):
         repsonse if valid the object will be JSON encoded. Otherwise
         it will be the request.reposne object.
         """
+        if not self.is_logged_in():
+            raise SolveError("HTTP request: client is not logged in!")
 
         opts = {
             'allow_redirects': True,
@@ -284,8 +296,29 @@ def request(self, method, url, **kwargs):
         try:
             return response.json()
         except Exception:
-            raise SolveError("Could not parse JSON response: {}"
-                             .format(response.content))
+            if b"<!DOCTYPE html>" in response.content[:40]:
+                helper_msg = self._host
+                if suggested_host := self.validate_host_is_www_url(self._host):
+                    helper_msg = f"Provided API host is: `{self._host}`. Did you perhaps mean `{suggested_host}`?"
+
+                html_response = response.text[:30].replace("\n", "")
+                raise SolveError(
+                    f"QuartzBio error: HTML response received from API. {html_response}...\n"
+                    "  Please confirm that API Host doesn't point to EDP's Web URL:\n"
+                    f"  {helper_msg}"
+                )
+            elif self._host is None:
+                # shouldn't happen, maybe in rare race conditions
+                raise SolveError("No EDP API host was set!")
+
+            raise SolveError(
+                f"Could not parse JSON response: {response.content}"
+            )
+
+    def validate_host_is_www_url(self, host):
+        # returns corrected API's URL, if it suspects that host is a WWW url, not API
+        if ".api.edp" not in host and ".edp" in host:
+            return host.replace(".edp", ".api.edp")
 
     def _log_raw_request(self, method, url, **kwargs):
         from requests import Request, Session