[GH-1181] Ingest Workflow Outputs into TDR

api/deps.edn

@@ -75,6 +75,7 @@
 
   :test
   {:extra-deps  {lambdaisland/kaocha           {:mvn/version "1.0.632"}
+                 org.apache.commons/commons-io {:mvn/version "1.3.2"}
                  org.liquibase/liquibase-core  {:mvn/version "4.0.0"
                                                 :exclusions [ch.qos.logback/logback-classic]}}
    :extra-paths ["test"]

api/src/wfl/service/cromwell.clj

@@ -46,8 +46,8 @@
 (defn ^:private request-json
   "Response to REQUEST with :body parsed as JSON."
   [request]
-  (let [{:keys [body] :as response} (http/request request)]
-    (assoc response :body (json/read-str body :key-fn keyword))))
+  (-> (http/request request)
+      (update :body (fnil util/parse-json "null"))))
 
 (def ^:private bogus-key-character-map
   "Map bogus characters in metadata keys to replacements."
@@ -72,12 +72,9 @@
   "Assemble PARTS into a multipart HTML body and post it to the Cromwell
   server specified by URL, and return the workflow ID."
   [url parts]
-  (letfn [(multipartify [[k v]] {:name (name k) :content v})]
-    (-> {:method    :post               ; :debug true :debug-body true
-         :url       url
-         :headers   (auth/get-auth-header)
-         :multipart (map multipartify parts)}
-        request-json #_debug/dump :body)))
+  (util/response-body-json
+   (http/post url {:headers   (auth/get-auth-header)
+                   :multipart (util/multipart-body parts)})))
 
 (defn make-workflow-labels
   "Return workflow labels for WDL."

api/src/wfl/service/datarepo.clj

@@ -44,7 +44,7 @@
   ([dataset-id profile-id source target]
    (ingest-file dataset-id profile-id source target {})))
 
-(defn ingest-dataset
+(defn ingest-table
   "Ingest TABLE at PATH to DATASET-ID and return the job ID."
   [dataset-id path table]
   (ingest

api/src/wfl/service/google/storage.clj

@@ -135,19 +135,22 @@
               "https://cloud.google.com/storage/docs/naming#requirements"
               (s/explain-str ::bucket-name bucket))))))
 
+(defn upload-content
+  "Upload CONTENT to BUCKET with name OBJECT."
+  ([content bucket object]
+   (-> (str upload-url bucket "/o")
+       (http/post {:query-params {:uploadType "media" :name object}
+                   :content-type (.detect (new Tika) content)
+                   :headers      (auth/get-auth-header)
+                   :body         content})
+       util/response-body-json))
+  ([content url]
+   (apply upload-content content (parse-gs-url url))))
+
 (defn upload-file
   "Upload FILE to BUCKET with name OBJECT."
   ([file bucket object]
-   (let [body (io/file file)]
-     (-> {:method       :post           ; :debug true :debug-body true
-          :url          (str upload-url bucket "/o")
-          :query-params {:uploadType "media"
-                         :name       object}
-          :content-type (.detect (new Tika) body)
-          :headers      (auth/get-auth-header)
-          :body         body}
-         http/request :body
-         (json/read-str :key-fn keyword))))
+   (upload-content (io/file file) bucket object))
   ([file url]
    (apply upload-file file (parse-gs-url url))))
 

api/src/wfl/service/terra.clj

@@ -1,9 +1,10 @@
 (ns wfl.service.terra
   "Analyze data in Terra using the Firecloud/Terra API."
-  (:require [clojure.data.json :as json]
-            [clj-http.client :as http]
-            [wfl.auth :as auth]
-            [wfl.util :as util]))
+  (:require [clj-http.client   :as http]
+            [clojure.data.json :as json]
+            [clojure.string    :as str]
+            [wfl.auth          :as auth]
+            [wfl.util          :as util]))
 
 (defn workspace-api-url
   [terra-url workspace]
@@ -38,6 +39,24 @@
         response (http/get submission-url {:headers (auth/get-auth-header)})]
     (util/parse-json (:body response))))
 
+(defn get-workflow
+  "Query the `firecloud-url` for the the `workflow` created by the `submission`
+   in the Terra `workspace`."
+  [firecloud-url workspace submission-id workflow-id]
+  (-> (workspace-api-url firecloud-url workspace)
+      (str (str/join "/" ["" "submissions" submission-id "workflows" workflow-id]))
+      (http/get {:headers (auth/get-auth-header)})
+      util/response-body-json))
+
+(defn get-workflow-outputs
+  "Query the `firecloud-url` for the outputs of the `workflow` created by
+   the `submission` in the Terra `workspace`."
+  [firecloud-url workspace submission-id workflow-id]
+  (-> (workspace-api-url firecloud-url workspace)
+      (str (str/join "/" ["" "submissions" submission-id "workflows" workflow-id "outputs"]))
+      (http/get {:headers (auth/get-auth-header)})
+      util/response-body-json))
+
 (defn get-workflow-status-by-entity
   "Get workflow status given a Terra submission-id and entity-name."
   [terra-url workspace {:keys [uuid inputs] :as _item}]
@@ -46,3 +65,14 @@
        (filter #(= (:entity-name inputs) (get-in % [:workflowEntity :entityName])))
        (first)
        (:status)))
+
+(defn describe-wdl
+  "Use `firecloud-url` to describe the WDL at `wdl-url`"
+  [firecloud-url wdl-url]
+  (-> (str firecloud-url "/api/womtool/v1/describe")
+      (http/post {:headers   (auth/get-auth-header)
+                  :multipart (util/multipart-body
+                              {:workflowUrl         wdl-url
+                               :workflowTypeVersion "1.0"
+                               :workflowType        "WDL"})})
+      util/response-body-json))

api/src/wfl/util.clj

@@ -369,3 +369,9 @@
       (use resource)
       (finally
         (release resource)))))
+
+(defn multipart-body
+  "Assemble PARTS into a multipart HTML body."
+  [parts]
+  (letfn [(make-part [[k v]] {:name (name k) :content v})]
+    (map make-part parts)))

api/test/resources/datasets/assemble-refbased-outputs.json

@@ -6,10 +6,6 @@
       {
         "name": "assemble_refbased_outputs",
         "columns": [
-          {
-            "name": "id",
-            "datatype": "string"
-          },
           {
             "name": "assembly_fasta",
             "datatype": "fileref"
@@ -151,9 +147,7 @@
             "datatype": "string"
           }
         ],
-        "primaryKey": [
-          "id"
-        ],
+        "primaryKey": [],
         "partitionMode": "date",
         "datePartitionOptions": {
           "column": "datarepo_ingest_date"