[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,60 @@ 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,
+):
+    """
+    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).
+    if access_token:
+        client._host, client._auth = authenticate(
+            api_host, access_token, token_type="Bearer"
+        )
+    elif api_key:
+        client._host, client._auth = authenticate(api_host, api_key, token_type="Token")
 
-    Lookup order:
-        1. access_token
-        2. api_key
-        3. local credentials
+    client.set_user_agent(name=name, version=version)    
 
-    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
+
+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,146 @@
+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=True,
+) -> 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.
+    """
+
+    # Find credentials from environment variables
+    if not host:
+        host = (
+            os.environ.get("SOLVEBIO_API_HOST", None)
+            or os.environ.get("SOLVEBIO_API_HOST", None)
+            or os.environ.get("EDP_API_HOST", None)
+        )
+
+    if not token:
+        api_key = (
+            os.environ.get("SOLVEBIO_API_KEY", None)
+            or os.environ.get("SOLVEBIO_API_KEY", None)
+            or os.environ.get("EDP_API_KEY", None)
+        )
+
+        access_token = (
+            os.environ.get("SOLVEBIO_ACCESS_TOKEN", None)
+            or os.environ.get("SOLVEBIO_ACCESS_TOKEN", None)
+            or os.environ.get("EDP_ACCESS_TOKEN", None)
+        )
+
+        if access_token:
+            token = access_token
+            token_type = "Bearer"
+        elif api_key:
+            token = api_key
+            token_type = "Token"
+
+    # 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 not host and raise_on_missing:
+        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()

solvebio/cli/auth.py

@@ -9,34 +9,45 @@
 from .credentials import delete_credentials
 
 
-def login_and_save_credentials(*args, **kwargs):
+def login_and_save_credentials(
+    *args,
+    api_host: str = None,
+    api_key: str = None,
+    access_token: str = None,
+    name: str = None,
+    version: str = None,
+):
     """
-    Domain, email and password are used to get the user's API key.
+    CLI command to login and persist credentials to a file
     """
-    if args and args[0].api_key:
-        # Handle command-line arguments if provided.
-        logged_in = solvebio.login(api_key=args[0].api_key,
-                                   api_host=solvebio.api_host)
-    elif args and args[0].access_token:
-        # Handle command-line arguments if provided.
-        logged_in = solvebio.login(access_token=args[0].access_token,
-                                   api_host=solvebio.api_host)
-    elif solvebio.login(**kwargs):
-        logged_in = True
-    else:
-        logged_in = False
+    if args and args[0].access_token:
+        access_token = args[0].access_token
+    elif args and args[0].api_key:
+        api_key = args[0].api_key
 
-    if logged_in:
-        # Print information about the current user
-        user = client.whoami()
-        print_user(user)
-        save_credentials(
-            user['email'].lower(), client._auth.token,
-            client._auth.token_type, solvebio.api_host)
-        print('Updated local credentials file.')
-    else:
-        print('You are not logged-in. Visit '
-              'https://docs.solvebio.com/#authentication to get started.')
+    if args and args[0].api_host:
+        api_host = args[0].api_host
+
+    solvebio.login(
+        api_host=api_host,
+        api_key=api_key,
+        access_token=access_token,
+        name=name,
+        version=version,
+    )
+
+    # Print information about the current user
+    user = client.whoami()
+    print_user(user)
+
+    # fixme: how to detect if login was successful
+    save_credentials(
+        user["email"].lower(),
+        client._auth.token,
+        client._auth.token_type,
+        solvebio.get_api_host(),
+    )
+    print("Updated local credentials file.")
 
 
 def logout(*args):
@@ -69,5 +80,5 @@ def print_user(user):
     """
     email = user['email']
     domain = user['account']['domain']
-    print('You are logged-in to the "{}" domain as {} (server: {}).'
-          .format(domain, email, solvebio.api_host))
+    print(f'You are logged-in to the "{domain}" domain as {email}'
+        f' (server: {solvebio.get_api_host()}).')