Add example use case for the session cookie handling

docs/examples/session_handling/cookie_handler.py

@@ -0,0 +1,96 @@
+"""
+Cookie handler for Authelia authentication using the Galaxy API
+"""
+
+import getpass
+import sys
+
+from http.cookiejar import LWPCookieJar
+from pathlib import Path
+from pprint import pprint
+import logging as log
+
+import requests
+
+from galaxy_api import *
+
+AUTH_HOSTNAME = "auth.service.org"
+API_HOSTNAME = "galaxy.service.org"
+cookie_path = Path(".galaxy_auth.txt")
+cookie_jar = LWPCookieJar(cookie_path)
+
+
+class ExpiredCookies(Exception):
+    pass
+
+
+class NoCookies(Exception):
+    pass
+
+
+def main():
+    try:
+        cookie_jar.load()  # raises OSError
+        if not cookie_jar:  # if empty due to expirations
+            raise ExpiredCookies()
+    except OSError:
+        print("No cached session found, please authenticate")
+        prompt_authentication()
+    except ExpiredCookies:
+        print("Session has expired, please authenticate")
+        prompt_authentication()
+    run_examples()
+
+
+def prompt_authentication():
+    # --------------------------------------------------------------------------
+    # Prompt for username and password
+
+    username = input("Please enter username: ")
+    password = getpass.getpass(f"Please enter password for {username}: ")
+
+    # --------------------------------------------------------------------------
+    # Prepare authentication packet and authenticate session using Authelia
+
+    login_body = {
+        "username": username,
+        "password": password,
+        "requestMethod": "GET",
+        "keepMeLoggedIn": True,
+        "targetURL": API_HOSTNAME,
+    }
+
+    with requests.sessions.Session() as session:
+        session.cookies = cookie_jar
+        session.verify = True
+
+        auth = session.post(f"https://{AUTH_HOSTNAME}/api/firstfactor", json=login_body)
+
+        response = session.get(f"https://{AUTH_HOSTNAME}/api/user/info")
+        if response.status_code != 200:
+            print("Authentication failed")
+            sys.exit()
+        else:
+            pprint(response.json())
+            session.cookies.save()
+
+
+def run_examples():
+    GALAXY_KEY = "user_api_key"
+    WORKFLOW_NAME = "workflow_name"
+    with requests.sessions.Session() as session:
+        session.cookies = cookie_jar
+
+        print("Running demo to demonstrate how to use the Galaxy API with Authelia")
+
+        print("Getting workflows from Galaxy")
+        response = get_workflows(f"https://{API_HOSTNAME}", GALAXY_KEY, session=session)
+        print(response)
+
+        print("Getting inputs for a workflow")
+        response = get_inputs(f"https://{API_HOSTNAME}", GALAXY_KEY, WORKFLOW_NAME, session=session)
+        print(response)
+
+
+if __name__ == "__main__":
+    main()

docs/examples/session_handling/galaxy_api.py

@@ -0,0 +1,58 @@
+from bioblend.galaxy import GalaxyInstance
+
+
+def get_inputs(server, api_key, workflow_name, session=None):
+    """
+    Function to get an array of inputs for a given galaxy workflow
+
+    Usage:
+        get_inputs(
+            server = "galaxy.server.org",
+            api_key = "user_api_key",
+            workflow_name = "workflow_name",
+        )
+
+    Args:
+        server (string): Galaxy server address
+        api_key (string): User generated string from galaxy instance
+            to create: User > Preferences > Manage API Key > Create a new key
+        workflow_name (string): Target workflow name
+    Returns:
+        inputs (array of strings): Input files expected by the workflow, these will be in the same order as they should be given in the main API call
+    """
+
+    gi = GalaxyInstance(url=server, key=api_key, session=session)
+    api_workflow = gi.workflows.get_workflows(name=workflow_name)
+    steps = gi.workflows.export_workflow_dict(api_workflow[0]["id"])["steps"]
+    inputs = []
+    for step in steps:
+        # Some of the steps don't take inputs so have to skip these
+        if len(steps[step]["inputs"]) > 0:
+            inputs.append(steps[step]["inputs"][0]["name"])
+
+    return inputs
+
+
+def get_workflows(server, api_key, session=None):
+    """
+    Function to get an array of workflows available on a given galaxy instance
+
+    Usage:
+        get_workflows(
+            server = "galaxy.server.org",
+            api_key = "user_api_key",
+        )
+
+    Args:
+        server (string): Galaxy server address
+        api_key (string): User generated string from galaxy instance
+            to create: User > Preferences > Manage API Key > Create a new key
+    Returns:
+        workflows (array of strings): Workflows available to be run on the galaxy instance provided
+    """
+    gi = GalaxyInstance(url=server, key=api_key, session=session)
+    workflows_dict = gi.workflows.get_workflows()
+    workflows = []
+    for item in workflows_dict:
+        workflows.append(item["name"])
+    return workflows