Merge pull request #2 from HakaiInstitute/add-credentials-input

Add a credentials inputs to the Client class

README.md

@@ -48,10 +48,16 @@ client = Client("http://localhost:8666")
 print(client.api_root) # http://localhost:8666
 ```
 
+You can also pass in the credentials string retrieved from the hakai API login page while initiating the Client class.
+
+```python
+# Pass a credentials token as the Client Class is initiated
+client = Client(credentials="CREDENTIAL_TOKEN")
+```
+
 # Development
 The business code for this package is in the file [`hakai_api/Client.py`](hakai_api/Client.py).
 The `./setup.py` file is important for giving details about how to install the package when users do `pip install hakai-api`.
 
 To build a new PyPi package version push a tag matching the pattern `v[0-9]+.[0-9]+.[0-9]+` or `v[0-9]+.[0-9]+.[0-9]+rc[0-9]+` to the origin (e.g. `v0.4.1` or `v0.5.2rc1`).
 Github Actions should take care of packaging and pushing it to Hakai's PyPi repository from there.
-

hakai_api/Client.py

@@ -15,29 +15,38 @@
 
 
 class Client(OAuth2Session):
-    _credentials_file = os.path.expanduser('~/.hakai-api-auth')
-
-    def __init__(self, api_root: str = "https://hecate.hakai.org/api",
-                 login_page: str = "https://hecate.hakai.org/api-client-login"):
+    _credentials_file = os.path.expanduser("~/.hakai-api-auth")
+
+    def __init__(
+        self,
+        api_root: str = "https://hecate.hakai.org/api",
+        login_page: str = "https://hecate.hakai.org/api-client-login",
+        credentials: Union[str, Dict] = None,
+    ):
         """Create a new Client class with credentials.
 
         Params:
             api_root: The base url of the hakai api you want to call.
                       Defaults to the production server.
-        """
+            credentials (str, Dict): Credentials token retrieved from the hakai api login page.
+                                     If `None`, loads cached credentials or prompts for log in.
+        """     
         self._api_root = api_root
         self._authorization_base_url = login_page
 
-        # Try to get cached credentials
-        credentials = self._try_to_load_credentials()
-        if credentials:
+        if credentials is None:
+            # Try to get cached credentials, prompt user if the file can't be loaded
+            self._credentials = self._try_to_load_credentials_file() or self._get_credentials_from_web()
+        elif isinstance(credentials, str):
+            # Parse credentials from string
+            self._credentials = dict(map(lambda x: x.split("="), credentials.split("&")))
+        elif isinstance(credentials, dict):
             self._credentials = credentials
         else:
-            # Acquire and store credentials from web sign-in.
-            credentials = self._get_credentials_from_web()
-            # Cache the credentials
-            self._save_credentials(credentials)
-            self._credentials = credentials
+            raise ValueError("`credentials` parameter should be str, dict, or None type.")
+
+        # Cache the credentials
+        self._save_credentials(self._credentials)
 
         # Init the OAuth2Session parent class with credentials
         super(Client, self).__init__(token=self._credentials)
@@ -59,24 +68,28 @@ def reset_credentials(cls):
 
     def _save_credentials(self, credentials: Dict):
         """Save the credentials object to a file."""
-        with open(self._credentials_file, 'wb') as outfile:
+        with open(self._credentials_file, "wb") as outfile:
             pickle.dump(credentials, outfile)
 
-    def _try_to_load_credentials(self) -> Union[Dict, bool]:
+    def _try_to_load_credentials_file(self) -> Union[Dict, bool]:
         """Try to load the cached credentials file."""
         if not os.path.isfile(self._credentials_file):
             return False
 
-        with open(self._credentials_file, 'rb') as infile:
+        with open(self._credentials_file, "rb") as infile:
             try:
                 credentials = pickle.load(infile)
-                expires_at = int(credentials['expires_at'])
+                expires_at = int(credentials["expires_at"])
             except (KeyError, ValueError):
                 os.remove(self._credentials_file)
                 return False
 
-            now = int((mktime(datetime.now(tz=utc).timetuple()) + datetime.now(
-                tz=utc).microsecond / 1000000.0))  # utc timestamp
+            now = int(
+                (
+                    mktime(datetime.now(tz=utc).timetuple())
+                    + datetime.now(tz=utc).microsecond / 1000000.0
+                )
+            )  # utc timestamp
 
             if now > expires_at:
                 os.remove(self._credentials_file)
@@ -86,11 +99,11 @@ def _try_to_load_credentials(self) -> Union[Dict, bool]:
 
     def _get_credentials_from_web(self) -> Dict:
         """Get user credentials from a web sign-in."""
-        print('Please go here and authorize:')
+        print("Please go here and authorize:")
         print(self._authorization_base_url)
 
         sleep(0.05)  # Workaround for input / print stream race condition
-        response = input('\nCopy and past your credentials from the login page:\n')
+        response = input("\nCopy and past your credentials from the login page:\n")
 
         # Reformat response to dict
         credentials = dict(map(lambda x: x.split("="), response.split("&")))