Merge pull request #97 from yetanalytics/pip-81

[PIP 81] - Toggle JSON-Only

Author: cliffcaseyyet

doc/options.md

@@ -48,6 +48,7 @@ All options:
   -p, --xapi-get-param KEY=VALUE        {}                    xAPI GET Parameters
       --source-username USERNAME                              Source LRS BASIC Auth username
       --source-password PASSWORD                              Source LRS BASIC Auth password
+      --json-only                                             Only operate in JSON statement mode for data transfer, ignoring Attachments/multipart (for compatibility issues)
       --source-auth-uri URI                                   Source LRS OAuth autentication URI
       --source-client-id ID                                   Source LRS OAuth client ID
       --source-client-secret SECRET                           Source LRS OAuth client secret

src/cli/com/yetanalytics/xapipe/cli.clj

@@ -143,8 +143,8 @@
       (log/debugf "Adding shutdown hook for job %s" (:id job))
       (.addShutdownHook (Runtime/getRuntime)
                         (Thread. ^Runnable
-                                 (fn []
-                                   (force-stop-job! stop states))))
+                         (fn []
+                           (force-stop-job! stop states))))
       (let [_ (log/debugf "Waiting for job %s" (:id job))
             {{:keys [status]} :state
              :as job-result} (-> states
@@ -201,6 +201,7 @@
    :get-params                 [:source :get-params]
    :source-username            [:source :request-config :username]
    :source-password            [:source :request-config :password]
+   :json-only                  [:source :request-config :json-only]
    :source-auth-uri            [:source :request-config :oauth-params :auth-uri]
    :source-client-id           [:source :request-config :oauth-params :client-id]
    :source-client-secret       [:source :request-config :oauth-params :client-secret]
@@ -393,8 +394,8 @@
 (defn list-store-jobs
   [store]
   (doseq [[page batch] (->> (store/list-jobs store)
-                           (partition-all 100)
-                           (map-indexed vector))]
+                            (partition-all 100)
+                            (map-indexed vector))]
     (log/infof "Page %d%s"
                page
                (with-out-str

src/cli/com/yetanalytics/xapipe/cli/options.clj

@@ -54,12 +54,10 @@
                `boolean?
                :else
                `string?)
-        arg-spec `#{
-                    ~(format "--%s" long-command-name)
+        arg-spec `#{~(format "--%s" long-command-name)
                     ~@(if short-command
                         [short-command]
-                        [])
-                    }]
+                        [])}]
     `(list
       ;; actual key spec
       (s/def ~spec-kw
@@ -251,7 +249,9 @@
                     v))
            m)))]
     [nil "--source-username USERNAME" "Source LRS BASIC Auth username"]
-    [nil "--source-password PASSWORD" "Source LRS BASIC Auth password"]]
+    [nil "--source-password PASSWORD" "Source LRS BASIC Auth password"]
+    [nil "--json-only" "Only operate in JSON statement mode for data transfer, ignoring Attachments/multipart (for compatibility issues)"
+     :default false]]
    (concat
     (oauth-opts "source")
     (backoff-opts "source"))))

src/cli/com/yetanalytics/xapipe/main.clj

@@ -41,8 +41,7 @@ Delete a Job:
             ?json-file     :json-file
             ?json-out      :json-out
             :as            options} :options
-           :keys         [summary]} (opts/args->options args)
-          ]
+           :keys         [summary]} (opts/args->options args)]
       (if help?
         {:status  0
          :message (str
@@ -87,7 +86,7 @@ Delete a Job:
                         job/upgrade-job
                         (cond->
                           ;; If the user has requested force resume we clear
-                          force-resume?
+                         force-resume?
                           (-> (update :state state/clear-errors)
                               (update :state state/set-status :paused)))
                         (job/reconfigure-job
@@ -146,6 +145,7 @@ Delete a Job:
                                :message (format "Wrote job %s to %s"
                                                 job-id ?json-out)})
                 :else (do
+
                         (log/infof
                          (if new?
                            "Starting job %s"

src/lib/com/yetanalytics/xapipe.clj

@@ -293,8 +293,7 @@
               target-client-opts]
        :or {conn-mgr-opts {}
             source-client-opts {}
-            target-client-opts {}
-            }} :client-opts
+            target-client-opts {}}} :client-opts
       reporter :reporter
       :or {reporter (metrics/->NoopReporter)}}]
   (let [{:keys [id]
@@ -410,7 +409,7 @@
                           {:state (-> state-before
                                       (cond->
                                           ;; If a job is paused, re-init
-                                          (= :paused (:status state-before))
+                                       (= :paused (:status state-before))
                                         (state/set-status :init))
                                       (assoc :updated init-stamp))})
                    states-chan
@@ -542,9 +541,7 @@
         (log/infof "store result: %s" result)))
     (def stop-fn stop))
 
-  (clojure.pprint/pprint (stop-fn))
-
-  )
+  (clojure.pprint/pprint (stop-fn)))
 
 (comment
   ;; Same as above, but with redis as a store
@@ -605,8 +602,7 @@
       (let [result (a/<! store-result)]
         (log/infof "store result: %s" result)))
     (def stop-fn stop))
-  (stop-fn)
-  )
+  (stop-fn))
 
 (comment
   ;; Use OAuth source LRS
@@ -647,6 +643,4 @@
         (log/infof "store result: %s" result)))
     (def stop-fn stop))
 
-  (clojure.pprint/pprint (stop-fn))
-
-  )
+  (clojure.pprint/pprint (stop-fn)))

src/lib/com/yetanalytics/xapipe/client.clj

@@ -8,11 +8,11 @@
             [clojure.spec.gen.alpha :as sgen]
             [clojure.tools.logging :as log]
             [com.yetanalytics.xapipe.client.multipart-mixed :as multipart]
+            [com.yetanalytics.xapipe.client.json-only :as json-only]
             [com.yetanalytics.xapipe.client.oauth :as oauth]
             [com.yetanalytics.xapipe.metrics :as metrics]
             [com.yetanalytics.xapipe.util :as u]
             [xapi-schema.spec :as xs]
-            [xapi-schema.spec.resources :as xsr]
             [com.yetanalytics.xapipe.util.time :as t]
             [com.yetanalytics.xapipe.spec.common :as cspec])
   (:import [org.apache.http.impl.client CloseableHttpClient]
@@ -28,6 +28,16 @@
                  status)
       (update resp :body slurp))))
 
+;; Add json-only output coercion
+(defmethod client/coerce-response-body :json-only
+  [req {:keys [status] :as resp}]
+  (if (= 200 status)
+    (json-only/parse-response req resp)
+    (do
+      (log/warnf "Received json response with non-200 status %d"
+                 status)
+      (update resp :body slurp))))
+
 ;; Config needed for all requests
 (s/def ::url-base
   string?)
@@ -39,6 +49,9 @@
 (s/def ::username string?)
 (s/def ::password string?)
 
+;; json-only mode support
+(s/def ::json-only boolean?)
+
 ;; token support
 (s/def ::token string?)
 
@@ -48,6 +61,7 @@
    :opt-un [::xapi-prefix
             ::username
             ::password
+            ::json-only
             ::token
             ::oauth/oauth-params]))
 
@@ -117,12 +131,12 @@
                 :related_agents
                 :format]))
 
-(def get-request-base
+(defn get-request-base [json-only?]
   {:headers      {"x-experience-api-version" "1.0.3"}
    :method       :get
-   :as           :multipart/mixed
+   :as           (if json-only? :json-only :multipart/mixed)
    :query-params {:ascending   true
-                  :attachments true}})
+                  :attachments (not json-only?)}})
 
 (s/def ::more string?) ;; more link
 
@@ -139,19 +153,20 @@
            username
            password
            token
-           oauth-params]
+           oauth-params
+           json-only]
     :or   {xapi-prefix "/xapi"}}
    get-params
    & [?more]]
   (cond-> (if (not-empty ?more)
             ;; Using More Link
-            (-> get-request-base
+            (-> (get-request-base json-only)
                 (assoc :url
                        (format "%s%s"
                                url-base
                                ?more))
                 (dissoc :query-params))
-            (-> get-request-base
+            (-> (get-request-base json-only)
                 (assoc :url
                        (format "%s%s/statements"
                                url-base
@@ -201,7 +216,7 @@
                   (format "multipart/mixed; boundary=%s" boundary))
         (cond->
           ;; support token if provided
-          (not-empty token)
+         (not-empty token)
           (assoc :oauth-token token)
           ;; support basic auth if provided
           (and (not-empty username)

src/lib/com/yetanalytics/xapipe/client/json_only.clj

@@ -0,0 +1,27 @@
+(ns com.yetanalytics.xapipe.client.json-only
+  "JSON-only statement response handling. Necessary over default JSON parsing
+   to shim attachments which will not exist"
+  (:require [clj-http.client    :as http-client]
+            [clojure.spec.alpha :as s]
+            [com.yetanalytics.xapipe.client.multipart-mixed :as mm]))
+
+(s/def ::body
+  (s/keys :req-un
+          [:xapi.statements.GET.response/statement-result
+           ::mm/attachments]))
+
+(s/fdef parse-response
+  :args (s/cat :response map?)
+  :ret (s/keys :req-un [::body]))
+
+(defn parse-response
+  "Parse + close a json body"
+  [req resp]
+  (let [{:keys [body]} (http-client/coerce-json-body req resp false)]
+    (assoc resp :body  {:attachments []
+                        :statement-result
+                        (reduce-kv
+                         (fn [m k v]
+                           (assoc m (keyword k) v))
+                         {}
+                         body)})))

src/lib/com/yetanalytics/xapipe/job.clj

@@ -210,9 +210,9 @@
         (-> job
             (assoc :config config)
             (cond->
-                (and
-                 (:pattern filter-cfg)
-                 (not (:pattern filter-state)))
+             (and
+              (:pattern filter-cfg)
+              (not (:pattern filter-state)))
               (assoc-in [:state :filter :pattern] {})))
         (-> job
             (update :state

src/lib/com/yetanalytics/xapipe/job/config.clj

@@ -121,8 +121,8 @@
                 :poll-interval poll-interval)
          (assoc-in [:get-params :limit] get-batch-size)
          (cond->
-             ?since (update-in [:get-params :since] t/normalize-stamp)
-             ?until (update-in [:get-params :until] t/normalize-stamp)))
+          ?since (update-in [:get-params :since] t/normalize-stamp)
+          ?until (update-in [:get-params :until] t/normalize-stamp)))
      :target
      (assoc target-config
             :batch-size post-batch-size

src/test/com/yetanalytics/xapipe/cli/options_test.clj

@@ -57,6 +57,7 @@
             :source-poll-interval 1000,
             :get-params {},
             :source-backoff-budget 10000,
+            :json-only false
             :source-backoff-max-attempt 10}
            nil
 
@@ -66,6 +67,7 @@
             :source-poll-interval 1000,
             :get-params {},
             :source-backoff-budget 10000,
+            :json-only false
             :source-backoff-max-attempt 10}
            ["Failed to validate \"--source-batch-size 0\": Must be a positive integer"]
 
@@ -80,6 +82,7 @@
             :get-params {:since "2021-10-26T17:51:06.530464Z"
                          :related_agents true},
             :source-backoff-budget 10000,
+            :json-only false
             :source-backoff-max-attempt 10}
            nil))
 
@@ -129,6 +132,7 @@
    :target-backoff-budget 10000,
    :force-resume false,
    :redis-prefix "xapipe",
+   :json-only false
    :storage :file,
    :get-buffer-size 10,
    :target-batch-size 50,
@@ -157,9 +161,10 @@
     :cleanup-buffer-size 50,
     :source
     {:request-config
-     {:url-base "http://0.0.0.0:8080", :xapi-prefix "/xapi"},
+     {:url-base "http://0.0.0.0:8080",
+      :xapi-prefix "/xapi"},
      :batch-size 50,
-     :backoff-opts {:budget 10000, :max-attempt 10},
+     :backoff-opts {:budget 10000, :max-attempt 10}
      :poll-interval 1000,
      :get-params {:limit 50}},
     :target
@@ -215,6 +220,7 @@
     "--xapi-get-param" "format=exact"
     "--source-username" "foo"
     "--source-password" "bar"
+    "--json-only"
     "--source-backoff-budget" "1"
     "--source-backoff-max-attempt" "1"
     "--source-backoff-j-range" "1"
@@ -275,6 +281,7 @@
     :force-resume true,
     :redis-prefix "my-xapipe",
     :source-backoff-j-range 1,
+    :json-only true,
     :storage :file,
     :get-buffer-size 1,
     :target-backoff-initial 1,
@@ -291,15 +298,15 @@
     :metrics-reporter "prometheus"
     :prometheus-push-gateway "localhost:1234"
     :filter-ensure-paths [[["id"]]]
-    :filter-match-paths [[[["verb"]["id"]]
+    :filter-match-paths [[[["verb"] ["id"]]
                           "http://example.com/verb"]
                          [[["actor"]] {"mbox" "mailto:bob@example.com"
                                        "objectType" "Agent"}]]
     :filter-concept-profile-urls ["http://example.org/profile.jsonld"]
     :filter-concept-types ["Verb"]
-    :filter-activity-type-ids [ "http://example.org/profile.jsonld#activity-type"]
+    :filter-activity-type-ids ["http://example.org/profile.jsonld#activity-type"]
     :filter-verb-ids ["http://example.org/profile.jsonld#verb"]
-    :filter-attachment-usage-types [ "http://example.org/profile.jsonld#aut"]
+    :filter-attachment-usage-types ["http://example.org/profile.jsonld#aut"]
     :cleanup-buffer-size 1}
    [])
   (testing "no defaults"