[EDPEV-2767] Move Code from QuartzBio Client

solvebio/__init__.py

@@ -22,17 +22,6 @@
     # Python 2.6 doesn't support this
     pass
 
-# Read/Write API key
-api_key = _os.environ.get('SOLVEBIO_API_KEY', None)
-# OAuth2 access tokens
-access_token = _os.environ.get('SOLVEBIO_ACCESS_TOKEN', None)
-if access_token is None:
-    access_token = _os.environ.get('EDP_ACCESS_TOKEN', None)
-
-api_host = _os.environ.get('SOLVEBIO_API_HOST', None)
-if api_host is None:
-    api_host = _os.environ.get('EDP_API_HOST', 'https://api.solvebio.com')
-
 
 def help():
     _open_help('/docs')
@@ -99,7 +88,8 @@ def emit(self, record):
 from .query import Query, BatchQuery, Filter, GenomicFilter
 from .global_search import GlobalSearch
 from .annotate import Annotator, Expression
-from .client import SolveClient
+from .client import client, SolveClient
+from .auth import authenticate
 from .resource import (
     Application,
     Beacon,
@@ -125,53 +115,69 @@ def emit(self, record):
 )
 
 
-def login(**kwargs):
+def login(
+    api_host: str = None,
+    api_key: str = None,
+    access_token: str = None,
+    name: str = None,
+    version: str = None,
+    debug: bool = False,
+):
+    """
+    Function to login to the QuartzBio/EDP API when using EDP in a python script.
+    Note that another function is used when CLI command `quartzbio login` is used!
+    EDP checks user credentials & host URL from multiple sources, in the following order:
+
+    1) Parameters provided (e.g. the parameters of this function)
+    2) Environment variables (if the above parameters weren't provided)
+    3) quartzbio credentials file stored in the user's HOME directory
+        (if parameters and environment variables weren't found)
+
+    :param api_host: the QuartzBio EDP instance's URL to access.
+    :param access_token: your user's access token, which you can generate at the EDP website
+        (user menu > `Personal Access Tokens`)
+    :param api_key: Your API key. You can use this instead of providing an access token
+    :param name: name
+    :param version: version
+
+    Example:
+        .. code-block:: python
+
+            import quartzbio
+            quartzbio.login(
+                api_host="https://solvebio.api.az.aws.quartz.bio",
+                api_key=YOUR_API_KEY
+            )
     """
-    Sets up the auth credentials using the provided key/token,
-    or checks the credentials file (if no token provided).
+    token_type = None
+    token = None
 
-    Lookup order:
-        1. access_token
-        2. api_key
-        3. local credentials
+    if access_token:
+        token_type = "Bearer"
+        token = access_token
+    elif api_key:
+        token_type = "Token"
+        token = api_key
 
-    No errors are raised if no key is found.
-    """
-    from .cli.auth import get_credentials
-    global access_token, api_key, api_host
-
-    # Clear any existing auth keys
-    access_token, api_key = None, None
-    # Update the host
-    api_host = kwargs.get('api_host') or api_host
-
-    if kwargs.get('access_token'):
-        access_token = kwargs.get('access_token')
-    elif kwargs.get('api_key'):
-        api_key = kwargs.get('api_key')
-    else:
-        creds = get_credentials()
-        # creds = (host, email, token_type, token)
-        if creds:
-            api_host = creds[0]
-            if creds[2] == 'Bearer':
-                access_token = creds[3]
-            else:
-                # By default, assume it is an API key.
-                api_key = creds[3]
-
-    # Always update the client host, version and agent
-    from solvebio.client import client
-    client.set_host()
-    client.set_user_agent(name=kwargs.get('name'),
-                          version=kwargs.get('version'))
-
-    if not (api_key or access_token):
-        return False
+    if api_host or token or debug:
+        client._host, client._auth = authenticate(
+            api_host, token, token_type=token_type, debug=debug
+        )
+
+    client.set_user_agent(name=name, version=version)
+
+
+def whoami():
+    try:
+        user = client.whoami()
+    except Exception as e:
+        print("{} (code: {})".format(e.message, e.status_code))
     else:
-        # Update the client token
-        client.set_token()
-        return True
+        return user
+
+
+def get_api_host():
+    return client._host
 
 
 __all__ = [

solvebio/__main__.py

@@ -0,0 +1,7 @@
+
+import sys
+
+from .cli.main import main
+
+if __name__ == "__main__":
+    main(sys.argv[1:])

solvebio/auth.py

@@ -0,0 +1,183 @@
+from __future__ import absolute_import
+
+import os
+from typing import Literal, Tuple
+
+from six.moves.urllib.parse import urlparse
+
+import logging
+
+from requests.auth import AuthBase
+import requests
+
+from solvebio import SolveError
+from solvebio.cli.credentials import get_credentials
+
+logger = logging.getLogger("solvebio")
+
+
+class SolveBioTokenAuth(AuthBase):
+    """Custom auth handler for SolveBio API token authentication"""
+
+    def __init__(self, token=None, token_type="Token"):
+        self.token = token
+        self.token_type = token_type
+
+    def __call__(self, r):
+        if self.token:
+            r.headers["Authorization"] = "{0} {1}".format(self.token_type, self.token)
+        return r
+
+    def __repr__(self):
+        if self.token:
+            return self.token_type
+        else:
+            return "Anonymous"
+
+
+def authenticate(
+    host: str,
+    token: str,
+    token_type: Literal["Bearer", "Token"],
+    *,
+    raise_on_missing: bool = True,
+    debug: bool = False
+) -> Tuple[str, SolveBioTokenAuth]:
+    """
+    Sets login credentials for SolveBio API authentication.
+
+    :param str host: API host url
+    :param str token: API access token or API key
+    :param str token_type: API token type. `Bearer` is used for access tokens, while `Token` is used for API Keys.
+    :param bool raise_on_missing: Raise an exception if no credentials are available.
+    """
+    # used for debugging
+    source_host = None
+    source_token = None
+
+    # Find credentials from environment variables
+    if not host:
+        host = (
+            os.environ.get("QUARTZBIO_API_HOST", None)
+            or os.environ.get("EDP_API_HOST", None)
+            or os.environ.get("SOLVEBIO_API_HOST", None)
+        )
+
+    if not token:
+        api_key = (
+            os.environ.get("QUARTZBIO_API_KEY", None)
+            or os.environ.get("EDP_API_KEY", None)
+            or os.environ.get("SOLVEBIO_API_KEY", None)
+        )
+
+        access_token = (
+            os.environ.get("QUARTZ_ACCESS_TOKEN", None)
+            or os.environ.get("EDP_ACCESS_TOKEN", None)
+            or os.environ.get("SOLVEBIO_ACCESS_TOKEN", None)
+        )
+
+        if access_token:
+            token = access_token
+            token_type = "Bearer"
+        elif api_key:
+            token = api_key
+            token_type = "Token"
+
+        if token:
+            source_token = 'envvars'
+    else:
+        source_token = 'params'
+
+    # Find credentials from local credentials file
+    if not token:
+        if creds := get_credentials(host):
+            token_type = creds.token_type
+            token = creds.token
+
+            if host is None:
+                # this happens when user/ennvars provided no API host for the login command
+                # but the credentials file still contains login credentials
+                host = creds.api_host
+
+                if host:
+                    source_host = 'creds'
+            if token:
+                source_token = 'creds'
+
+    if (not host and raise_on_missing) or debug:
+        # this will tell the user where QB Client found the credentials from
+        creds_path = netrc.path()
+        print('\n'.join([
+            "Login Debug:",
+            f"--> Host: {host}\n    (source: {source_host})",
+            f"--> Token Type: {token_type}\n    (source: {source_token})",
+            "\n1) source: params",
+            "   Means that you've passed this through the login CLI command:",
+            "   quartzbio login --host <EDP_HOST> --access_token <EDP_TOKEN>",
+            "\n   or the quartzbio.login function:",
+            "   import quartzbio",
+            "   quartzbio.login(debug=True)",
+            "\n2) source: creds",
+            "   Means that the QB client has saved your credentials in:",
+            f"   {creds_path}",
+            "\n3) source: envvars",
+            "   Means that you've set your credentials through environment variables:",
+            "   QUARTZBIO_API_HOST",
+            "   QUARTZBIO_ACCESS_TOKEN",
+            "   QUARTZBIO_API_KEY",
+        ]))
+
+        if not debug:
+            raise SolveError("No SolveBio API host is set")
+
+    host = validate_api_host_url(host)
+
+    # If the domain ends with .solvebio.com, determine if
+    # we are being redirected. If so, update the url with the new host
+    # and log a warning.
+    if host and host.rstrip("/").endswith(".api.solvebio.com"):
+        old_host = host.rstrip("/")
+        response = requests.head(old_host, allow_redirects=True)
+        # Strip the port number from the host for comparison
+        new_host = validate_api_host_url(response.url).rstrip("/").replace(":443", "")
+
+        if old_host != new_host:
+            logger.warning(
+                'API host redirected from "{}" to "{}", '
+                "please update your local credentials file".format(old_host, new_host)
+            )
+            host = new_host
+
+    if token is not None:
+        auth = SolveBioTokenAuth(token, token_type)
+    else:
+        auth = None
+
+    # TODO: warn user if WWW url is provided in edp_login!
+
+    # @TODO: remove references to solvebio.api_host, etc...
+
+    return host, auth
+
+
+def validate_api_host_url(url):
+    """
+    Validate SolveBio API host url.
+
+    Valid urls must not be empty and
+    must contain either HTTP or HTTPS scheme.
+    """
+
+    # Default to https if no scheme is set
+    if "://" not in url:
+        url = "https://" + url
+
+    parsed = urlparse(url)
+    if parsed.scheme not in ["http", "https"]:
+        raise SolveError(
+            "Invalid API host: %s. " "Missing url scheme (HTTP or HTTPS)." % url
+        )
+    elif not parsed.netloc:
+        raise SolveError("Invalid API host: %s." % url)
+
+    return parsed.geturl()