building_data_pipelines_2_1.clj

(require '[clojure.spec.alpha :as s])

(defn hyphenize-keys
  "Given a map, hyphenize the first level of keys."
  [m]
  (reduce #(assoc %1
                  (-> (key %2)
                      name
                      str
                      (clojure.string/replace #"_" "-")
                      keyword)
                  (val %2))
          {}
          m))

(s/def ::subscriber
  (s/keys :req-un [::id ::name ::is-cat-owner ::city]))

(s/def ::output-spec ::subscriber)

(def create-table-sql
  "CREATE TABLE subscribers
   (id uuid primary key not null,
    name text not null,
    is_cat_owner boolean not null,
    city text not null)")

(def get-subscribers-query
  "SELECT id, name, is_cat_owner, city FROM subscribers")

(def input
  {:postgresql.query->local
   {:query get-subscribers-query
    :query-options {:row-fn hyphenize-keys}}})

(def output
  {:rabbitmq.local->exchange
   {:exchange :default
    :routing-key :work.filter-cat-owners
    :options {:persistent true}}})

(def job
  {:job 
   {:name :get-subscribers
    :description "Get all subscribers and forward to processing pipeline."
    :input-spec ::subscriber
    :input input
    :output-spec ::subscriber
    :output output
    :handler :data}})