unicore_api_Python3.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "jupyter": {
     "outputs_hidden": true
    }
   },
   "outputs": [],
   "source": [
    "import requests\n",
    "import json\n",
    "import time\n",
    "requests.packages.urllib3.disable_warnings()\n",
    "\n",
    "\"\"\"\n",
    "Helper methods for using UNICORE's REST API\n",
    "\n",
    "For a full API reference and examples, have a look at\n",
    "https://sourceforge.net/p/unicore/wiki/REST_API\n",
    "https://sourceforge.net/p/unicore/wiki/REST_API_Examples\n",
    "\"\"\"\n",
    "\n",
    "def get_sites():\n",
    "    \"\"\" get info about the known sites in the HPC Platform \"\"\"\n",
    "    sites = {}\n",
    "    sites['JUQUEEN'] = {'name': 'JUQUEEN (JSC)', 'id': 'JUQUEEN', \n",
    "                        'url': \"https://hbp-unic.fz-juelich.de:7112/HBP_JUQUEEN/rest/core\" }\n",
    "    sites['JURECA'] = {'name': 'JURECA (JSC)', 'id': 'JURECA',\n",
    "                       'url': \"https://hbp-unic.fz-juelich.de:7112/HBP_JURECA/rest/core\" }\n",
    "    sites['MARCONI'] = {'name': 'MARCONI (CINECA)', 'id': 'MARCONI',\n",
    "                     'url': \"https://grid.hpc.cineca.it:9111/CINECA-MARCONI/rest/core\" }\n",
    "    sites['GALILEO'] = {'name': 'GALILEO (CINECA)', 'id': 'MARCONI',\n",
    "                     'url': \"https://grid.hpc.cineca.it:9111/CINECA-GALILEO/rest/core\" }\n",
    "    sites['DAINT-CSCS'] = {'name': 'PIZDAINT (CSCS)', 'id': 'CSCS',\n",
    "                     'url': \"https://unicoregw.cscs.ch:8080/DAINT-CSCS/rest/core\" }\n",
    "    return sites\n",
    "\n",
    "\n",
    "def get_site(name):\n",
    "    return get_sites().get(name, None)\n",
    "\n",
    "\n",
    "def get_properties(resource, headers={},verbose=True):\n",
    "    \"\"\" get JSON properties of a resource \"\"\"\n",
    "    my_headers = headers.copy()\n",
    "    my_headers['Accept']=\"application/json\"\n",
    "    resp=\"\"\n",
    "    try:\n",
    "        r = requests.get(resource, headers=my_headers, verify=False)\n",
    "        if r.status_code!=200:\n",
    "            if r.status_code==500:\n",
    "                js = \"<script>alert('HPC server issues. Please try again later.');</script>\"\n",
    "                if verbose==True:\n",
    "                    display(HTML(js))\n",
    "                return \"server issues\"\n",
    "            else:\n",
    "                raise RuntimeError(\"Error getting properties: %s\" % r.status_code)\n",
    "        else:\n",
    "            return r.json()\n",
    "    except requests.exceptions.ConnectionError:\n",
    "        return \"Connection refused.\"       \n",
    "    \n",
    "def get_working_directory(job, headers={}, properties=None):\n",
    "    \"\"\" returns the URL of the working directory resource of a job \"\"\"\n",
    "    if properties is None:\n",
    "        properties = get_properties(job,headers)\n",
    "    return properties['_links']['workingDirectory']['href']\n",
    "\n",
    "\n",
    "def invoke_action(resource, action, headers, data={}):\n",
    "    my_headers = headers.copy()\n",
    "    my_headers['Content-Type']=\"application/json\"\n",
    "    action_url = get_properties(resource, headers)['_links']['action:'+action]['href']\n",
    "    r = requests.post(action_url,data=json.dumps(data), headers=my_headers, verify=False)\n",
    "    if r.status_code!=200:\n",
    "        raise RuntimeError(\"Error invoking action: %s\" % r.status_code)\n",
    "    return r\n",
    "\n",
    "\n",
    "def upload(destination, file_desc, headers):\n",
    "    my_headers = headers.copy()\n",
    "    my_headers['Content-Type']=\"application/octet-stream\"\n",
    "    name = file_desc['To']\n",
    "    data = file_desc['Data']\n",
    "    # TODO file_desc could refer to local file\n",
    "    r = requests.put(destination+\"/\"+name, data=data, headers=my_headers, verify=False)\n",
    "    if r.status_code!=204:\n",
    "        raise RuntimeError(\"Error uploading data: %s\" % r.status_code)\n",
    "    else:\n",
    "        jobURL=''\n",
    "\n",
    "\n",
    "def submit(url, job, headers, inputs=[],verbose=True):\n",
    "    \"\"\"\n",
    "    Submits a job to the given URL, which can be the \".../jobs\" URL\n",
    "    or a \".../sites/site_name/\" URL\n",
    "    If inputs is not empty, the listed input data files are\n",
    "    uploaded to the job's working directory, and a \"start\" command is sent\n",
    "    to the job.\n",
    "    \"\"\"\n",
    "    my_headers = headers.copy()\n",
    "    my_headers['Content-Type']=\"application/json\"\n",
    "    if len(inputs)>0:\n",
    "        # make sure UNICORE does not start the job \n",
    "        # before we have uploaded data\n",
    "        job['haveClientStageIn']='true'\n",
    "        \n",
    "    r = requests.post(url,data=json.dumps(job), headers=my_headers, verify=False)\n",
    "    if r.status_code!=201:\n",
    "        if r.status_code==500:\n",
    "            jobURL=''\n",
    "            js = \"<script>alert('System is in maintenance. Please try again later.');</script>\"\n",
    "            if verbose==True:\n",
    "                display(HTML(js))\n",
    "        else:\n",
    "            if r.status_code==403:\n",
    "                jobURL=''\n",
    "                js = \"<script>alert('Authentication service is restarting. Please try again later.');</script>\"\n",
    "                if verbose==True:\n",
    "                    display(HTML(js))\n",
    "            else:\n",
    "                raise RuntimeError(\"Error submitting job: %s\" % r.status_code)\n",
    "    else:\n",
    "        jobURL = r.headers['Location']\n",
    "\n",
    "        #  upload input data and explicitely start job\n",
    "        if len(inputs)>0:\n",
    "            working_directory = get_working_directory(jobURL, headers)\n",
    "            for input in inputs:\n",
    "                upload(working_directory+\"/files\", input, headers)\n",
    "            try:\n",
    "                invoke_action(jobURL, \"start\", headers)\n",
    "            except:\n",
    "                pass    \n",
    "    return jobURL\n",
    "\n",
    "    \n",
    "def is_running(job, headers={}):\n",
    "    \"\"\" check status for a job \"\"\"\n",
    "    properties = get_properties(job,headers)\n",
    "    status = properties['status']\n",
    "    return (\"SUCCESSFUL\"!=status) and (\"FAILED\"!=status)\n",
    "\n",
    "\n",
    "def wait_for_completion(job, headers={}):\n",
    "    \"\"\" wait until job is done \"\"\"\n",
    "    while is_running(job, headers):\n",
    "        time.sleep(3)\n",
    "\n",
    "\n",
    "def file_exists(wd, name, headers):\n",
    "    \"\"\" check if a file with the given name exists\n",
    "        if yes, return its URL\n",
    "        of no, return None\n",
    "    \"\"\"\n",
    "    files_url = get_properties(wd, headers)['_links']['files']['href']\n",
    "    children = get_properties(files_url, headers)['children']\n",
    "    return name in children or \"/\"+name in children\n",
    "\n",
    "\n",
    "def get_file_content(file_url, headers, check_size_limit=True, MAX_SIZE=30240000):\n",
    "    \"\"\" download binary file data \"\"\"\n",
    "    if check_size_limit:\n",
    "        size = get_properties(file_url, headers)['size']\n",
    "        if size>MAX_SIZE:\n",
    "            raise RuntimeError(\"File size too large!\")\n",
    "    my_headers = headers.copy()\n",
    "    my_headers['Accept']=\"application/octet-stream\"\n",
    "    r = requests.get(file_url, headers=my_headers, verify=False)\n",
    "    if r.status_code!=200:\n",
    "        raise RuntimeError(\"Error getting file data: %s\" % r.status_code)\n",
    "    else:\n",
    "        return r.content\n",
    "\n",
    "    \n",
    "def list_files(dir_url, auth, path=\"/\"):\n",
    "    #return get_properties(dir_url+\"/files\"+path, auth)['children']\n",
    "    return list(get_properties(dir_url+\"/files\"+path, auth)['content'].keys())\n",
    "\n",
    "\n",
    "def get_oidc_auth():\n",
    "    \"\"\" returns HTTP headers containing OIDC bearer token \"\"\"\n",
    "    token = clb_oauth.get_token()\n",
    "    session = requests.Session()\n",
    "    session.headers['Authorization'] = f'Bearer {token}'\n",
    "    hdr = session.headers['Authorization']\n",
    "    return {'Authorization': hdr}\n",
    "\n",
    "def get_user_id(url, auth):\n",
    "    \"\"\" returns the user id at the given url \"\"\"\n",
    "    return get_properties(url, auth)['client']['xlogin']['UID']"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}