load script (can restore from dump.sh backup)

Author: stain

load.py

@@ -0,0 +1,118 @@
+#!/usr/bin/env python3
+
+import sys
+import re
+import time
+import gzip
+from urllib.parse import urljoin
+import requests
+import json
+
+re.compile
+
+isCommit = re.compile("^[0-9a-fA-F]{40}$")
+
+HEADERS = {
+    'User-Agent': 'cwlviewer-load/0.0.1',
+    'Accept': 'application/json'
+}
+
+
+def parse_gitinfos(sourceFp):
+    doc = json.load(sourceFp)
+    for wf in doc["content"]:
+        yield wf["retrievedFrom"]
+
+def is_not_commit(gitinfo):
+    return not isCommit.match(gitinfo["branch"])
+
+def make_requests(gitinfos):
+    for git in gitinfos:
+        # dict_keys(['repoUrl', 'branch', 'path', 'packedId', 'url', 'rawUrl', 'type'])
+        req = {
+            "url": git["repoUrl"],
+            "branch": git["branch"],
+            "path": git["path"]
+        }
+        if git["packedId"]:
+            req["packedId"] = git["packedId"]
+        yield req
+
+def send(base, req):
+    url = urljoin(base, "/workflows")
+    r = requests.post(url, data=req, allow_redirects=False, headers=HEADERS)
+    print("Posted: %s" % req)
+    if r.status_code == 202:
+        location =  urljoin(url, r.headers["Location"]) 
+        print("  queued: %s" % location)
+        # need to check later
+        return location
+    if r.status_code == 303:
+        print ("  done: %s" % r.headers["Location"])
+        return None # Already there, all OK
+    print("Unhandled HTTP status code: %s  %s" % 
+                    (r.status_code, r.text))
+
+def send_requests(base, requests):
+    for req in requests:
+        yield send(base, req)
+
+def is_running(location):
+    if not location:
+        return True
+    queued = requests.get(location, allow_redirects=False, headers=HEADERS)
+    if queued.status_code == 303:
+        # Done!
+        return False
+    j = queued.json()
+    if j["cwltoolStatus"] == "RUNNING":
+        return True
+    elif j["cwltoolStatus"] == "ERROR":
+        print("Failed %s: %s" % (location, j["message"]))
+        return False
+    else:
+        raise Exception("Unhandled queue status: %s %s" % (queued.status_code, queued.text))
+
+MAX_CONCURRENT=6 # Maximum number in queue
+SLEEP=0.5 # wait SLEEP seconds if queue is full
+
+def trim_queue(queue):    
+    new_queue = []
+    for q in queue:
+        if is_running(q):
+            #print("Still running %s" % q)
+            new_queue.append(q)
+    print("Trimmed queue from %s to %s" % (len(queue), len(new_queue)))
+    return new_queue
+
+def main(jsonfile="-", base="http://view.commonwl.org:8082/", *args):
+    if jsonfile == "-":
+        source = sys.stdin
+    elif jsonfile.endswith(".gz"):
+        source = gzip.open(jsonfile, "rb")
+    else:
+        source = open(jsonfile, "rb")
+
+    gitinfos = parse_gitinfos(source)
+    if "--no-commits" in args:
+        gitinfos = filter(is_not_commit, gitinfos)
+    
+    requests = make_requests(gitinfos)
+    queued = []
+    for q in send_requests(base, requests):
+        if q:
+            queued.append(q)        
+        while len(queued) >= MAX_CONCURRENT:
+            time.sleep(SLEEP)
+            queued = trim_queue(queued)    
+    # Finish the rest of the queue
+    while queued:
+        queued = trim_queue(queued)
+
+if __name__ == "__main__":
+    if "-h" in sys.argv:
+        print("load.py [jsonfile] [baseurl] [--no-commits]")
+        sys.exit(1)
+
+    main(*sys.argv[1:])
+