Upserting in Datomic

upsert-auto.clj

(use '[datomic.api :only (q db) :as d])

(def initial-data
  [{:sku "1" :price 0.95M :qty 1}
   {:sku "2" :price 1.99M :qty 0}
   {:sku "3" :price 1.99M :qty 0}
   {:sku "4" :price 5.99M :qty 3}
   {:sku "5" :price 9.99M :qty 2}
   {:sku "6" :price 2.99M :qty 3}
   {:sku "7" :price 2.99M :qty 2}])

;; 1 & 2 not changed, 7 not included
;; 3-6 changed price and/or qty, 8 new
(def latest-data
  [{:sku "1" :price 0.95M :qty 1}
   {:sku "2" :price 1.99M :qty 0}
   {:sku "3" :price 1.99M :qty 1}
   {:sku "4" :price 1.99M :qty 3}
   {:sku "5" :price 9.99M :qty 3}
   {:sku "6" :price 9.99M :qty 0}
   {:sku "8" :price 1.99M :qty 2}])

(def uri "datomic:mem://history")

(d/delete-database uri)

(d/create-database uri)

(def conn (d/connect uri))

(def schema
 [{:db/id (d/tempid :db.part/db)
   :db/ident :sku
   :db/valueType :db.type/string
   :db/cardinality :db.cardinality/one
   :db/unique :db.unique/identity
   :db/index true
   :db.install/_attribute :db.part/db}
  {:db/id (d/tempid :db.part/db)
   :db/ident :price
   :db/valueType :db.type/bigdec
   :db/cardinality :db.cardinality/one
   :db.install/_attribute :db.part/db}
  {:db/id (d/tempid :db.part/db)
   :db/ident :qty
   :db/valueType :db.type/long
   :db/cardinality :db.cardinality/one
   :db.install/_attribute :db.part/db}
  {:db/id (d/tempid :db.part/db)
   :db/ident :source
   :db/valueType :db.type/keyword
   :db/cardinality :db.cardinality/one
   :db.install/_attribute :db.part/db}])

(d/transact conn schema)

(defn data-with-dbid
  [data]
  (map #(merge {:db/id (d/tempid :db.part/user)} %1) data))

(d/transact
 conn
 (conj (data-with-dbid initial-data)
       [:db/add (d/tempid :db.part/tx)
        :source :initial]))

(d/transact
 conn
 (conj (data-with-dbid latest-data)
       [:db/add (d/tempid :db.part/tx)
        :source :updated]))

(pprint (q '[:find ?e ?sku ?price-source ?qty-source
     :where
     [?e :sku ?sku]
     [?e :price _ ?price-tx]
     [?e :qty _ ?qty-tx]
     [?price-tx :source ?price-source]
     [?qty-tx :source ?qty-source]]
           (db conn)))

upsert-manual.clj

(use '[datomic.api :only (q db) :as d])

(def initial-data
  [{:sku "1" :price 0.95M :qty 1}
   {:sku "2" :price 1.99M :qty 0}
   {:sku "3" :price 1.99M :qty 0}
   {:sku "4" :price 5.99M :qty 3}
   {:sku "5" :price 9.99M :qty 2}
   {:sku "6" :price 2.99M :qty 3}
   {:sku "7" :price 2.99M :qty 2}])

;; 1 & 2 not changed, 7 not included
;; 3-6 changed price and/or qty, 8 new
(def latest-data
  [{:sku "1" :price 0.95M :qty 1}
   {:sku "2" :price 1.99M :qty 0}
   {:sku "3" :price 1.99M :qty 1}
   {:sku "4" :price 1.99M :qty 3}
   {:sku "5" :price 9.99M :qty 3}
   {:sku "6" :price 9.99M :qty 0}
   {:sku "8" :price 1.99M :qty 2}])

(def uri "datomic:mem://history")

(d/delete-database uri)

(d/create-database uri)

(def conn (d/connect uri))

(def schema
 [{:db/id (d/tempid :db.part/db)
   :db/ident :sku
   :db/valueType :db.type/string
   :db/cardinality :db.cardinality/one
   :db/unique :db.unique/identity
   :db/index true
   :db.install/_attribute :db.part/db}
  {:db/id (d/tempid :db.part/db)
   :db/ident :price
   :db/valueType :db.type/bigdec
   :db/cardinality :db.cardinality/one
   :db.install/_attribute :db.part/db}
  {:db/id (d/tempid :db.part/db)
   :db/ident :qty
   :db/valueType :db.type/long
   :db/cardinality :db.cardinality/one
   :db.install/_attribute :db.part/db}
  {:db/id (d/tempid :db.part/db)
   :db/ident :source
   :db/valueType :db.type/keyword
   :db/cardinality :db.cardinality/one
   :db.install/_attribute :db.part/db}])

(d/transact conn schema)

(defn data-with-dbid
  [data]
  (map #(merge {:db/id (d/tempid :db.part/user)} %1) data))

(d/transact
 conn
 (conj (data-with-dbid initial-data)
       [:db/add (d/tempid :db.part/tx)
        :source :initial]))

;; tuplify and maps->rel from https://gist.github.com/3068749
(defn tuplify
  "Returns a vector of the vals at keys ks in map."
  [m ks]
  (mapv #(get m %) ks))
(defn maps->rel
  "Returns the tuplification by ks of x, a collection
   of items that support key lookup"
  [x ks]
  (mapv #(tuplify % ks) x))

(count (q '[:find ?sku :where [?e :sku ?sku]] (db conn)))

;; find unchanged data using maps->rel
;; (would prefer to find the *changed* data instead)
(def unchanged-skus
  (->>
  (q '[:find ?sku
       :in $ $latest-data %
       :where
       [?e :sku ?sku]
       [?e :price ?old-price]
       [?e :qty ?old-qty]
       [$latest-data ?sku ?new-price ?new-qty]
       [(= ?old-price ?new-price)]
       [(= ?old-qty ?new-qty)]]
     (db conn)
     (maps->rel latest-data [:sku :price :qty])
     unchanged-rules)
  (map first)))

(defn remove-unchanged-from-data
  [unchanged-ids data]
  (remove #(some #{(:sku %)} (set unchanged-ids)) data))

(d/transact
 conn
 (conj (data-with-dbid (remove-unchanged-from-data unchanged-skus latest-data))
       [:db/add (d/tempid :db.part/tx)
        :source :updated]))

(pprint (q '[:find ?e ?sku ?price-source ?qty-source
     :where
     [?e :sku ?sku]
     [?e :price _ ?price-tx]
     [?e :qty _ ?qty-tx]
     [?price-tx :source ?price-source]
     [?qty-tx :source ?qty-source]]
                   (db conn)))

z-changed.clj

;; okay, this detects the *changed* skus, but it doesn't include the *added* sku 8

(defn changed-price-or-qty
  [old-price new-price old-qty new-qty]
  (or (not= old-price new-price)
      (not= old-qty new-qty)))

(def changed-skus
  (->>
  (q '[:find ?sku
       :in $ $latest-data
       :where
       [?e :sku ?sku]
       [?e :price ?old-price]
       [?e :qty ?old-qty]
       [$latest-data ?sku ?new-price ?new-qty]
       [(user/changed-price-or-qty ?old-price ?new-price ?old-qty ?new-qty)]]
     (db conn)
     (maps->rel latest-data [:sku :price :qty]))
  (map first)))