diff --git a/docs/examples/session_handling/cookie_handler.py b/docs/examples/session_handling/cookie_handler.py new file mode 100644 index 000000000..8b94f166e --- /dev/null +++ b/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() diff --git a/docs/examples/session_handling/galaxy_api.py b/docs/examples/session_handling/galaxy_api.py new file mode 100644 index 000000000..c052fc609 --- /dev/null +++ b/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