new database
This commit is contained in:
+435
@@ -0,0 +1,435 @@
|
||||
;; Copyright © 2015-2017, JUXT LTD.
|
||||
|
||||
(ns aero.core
|
||||
(:require
|
||||
[aero.alpha.core :refer
|
||||
[expand expand-scalar-repeatedly expand-case eval-tagged-literal
|
||||
reassemble kv-seq]]
|
||||
[aero.impl.walk :refer [postwalk]]
|
||||
#?@(:clj [[clojure.edn :as edn]
|
||||
[aero.impl.macro :as macro]]
|
||||
:cljs [[cljs.tools.reader.edn :as edn]
|
||||
[cljs.tools.reader :refer [default-data-readers *data-readers*]]
|
||||
[cljs.tools.reader.reader-types
|
||||
:refer [source-logging-push-back-reader]
|
||||
:as tools.reader.reader-types]])
|
||||
#?@(:clj [[clojure.java.io :as io]]
|
||||
:cljs [[goog.string :as gstring]
|
||||
goog.string.format
|
||||
[goog.object :as gobj]
|
||||
["fs" :as fs]
|
||||
["path" :as path] ["os" :as os]]))
|
||||
#?(:clj (:import (java.io StringReader)))
|
||||
#?(:cljs (:require-macros [aero.impl.macro :as macro])))
|
||||
|
||||
(defrecord Deferred [delegate])
|
||||
|
||||
(macro/usetime
|
||||
(declare read-config)
|
||||
|
||||
(defmulti reader (fn [opts tag value] tag))
|
||||
|
||||
(defmethod reader :default
|
||||
[_ tag value]
|
||||
(cond
|
||||
;; Given tagification, we now must check data-readers
|
||||
(contains? *data-readers* tag)
|
||||
((get *data-readers* tag) value)
|
||||
|
||||
(contains? default-data-readers tag)
|
||||
((get default-data-readers tag) value)
|
||||
:else
|
||||
(throw (ex-info (#?(:clj format :cljs gstring/format) "No reader for tag %s" tag) {:tag tag :value value}))))
|
||||
|
||||
(defn- get-env [s]
|
||||
#?(:clj (System/getenv (str s)))
|
||||
#?(:cljs (gobj/get js/process.env s)))
|
||||
|
||||
(defmethod reader 'env
|
||||
[opts tag value]
|
||||
(get-env value))
|
||||
|
||||
(defmethod reader 'envf
|
||||
[opts tag value]
|
||||
(let [[fmt & args] value]
|
||||
(apply #?(:clj format :cljs gstring/format) fmt
|
||||
(map #(str (get-env (str %))) args))))
|
||||
|
||||
(defmethod reader 'prop
|
||||
[opts tag value]
|
||||
#?(:clj (System/getProperty (str value))
|
||||
:cljs nil))
|
||||
|
||||
(defmethod reader 'long
|
||||
[opts tag value]
|
||||
#?(:clj (Long/parseLong (str value)))
|
||||
#?(:cljs (js/parseInt (str value))))
|
||||
|
||||
(defmethod reader 'double
|
||||
[opts tag value]
|
||||
#?(:clj (Double/parseDouble (str value)))
|
||||
#?(:cljs (js/parseFloat (str value))))
|
||||
|
||||
(defmethod reader 'keyword
|
||||
[opts tag value]
|
||||
(if (keyword? value)
|
||||
value
|
||||
(keyword (str value))))
|
||||
|
||||
(defmethod reader 'boolean
|
||||
[opts tag value]
|
||||
#?(:clj (Boolean/parseBoolean (str value)))
|
||||
#?(:cljs (= "true" (.toLowerCase (str value)))))
|
||||
|
||||
(defmethod reader 'include
|
||||
[{:keys [resolver source] :as opts} tag value]
|
||||
(read-config
|
||||
(if (map? resolver)
|
||||
(get resolver value)
|
||||
(resolver source value))
|
||||
opts))
|
||||
|
||||
(defmethod reader 'join
|
||||
[opts tag value]
|
||||
(apply str value))
|
||||
|
||||
(defmethod reader 'read-edn
|
||||
[opts tag value]
|
||||
(some-> value str edn/read-string))
|
||||
|
||||
(defmethod reader 'merge
|
||||
[opts tag values]
|
||||
(apply merge values))
|
||||
|
||||
#?(:clj
|
||||
(defn relative-resolver [source include]
|
||||
(let [fl
|
||||
(if (.isAbsolute (io/file include))
|
||||
(io/file include)
|
||||
(when-let [source-file
|
||||
(try (io/file source)
|
||||
;; Handle the case where the source isn't file compatible:
|
||||
(catch java.lang.IllegalArgumentException _ nil))]
|
||||
(io/file (.getParent ^java.io.File source-file) include)))]
|
||||
(if (and fl (.exists fl))
|
||||
fl
|
||||
(StringReader. (pr-str {:aero/missing-include include}))))))
|
||||
|
||||
#?(:clj
|
||||
(defn resource-resolver [_ include]
|
||||
(or
|
||||
(io/resource include)
|
||||
(StringReader. (pr-str {:aero/missing-include include})))))
|
||||
|
||||
#?(:clj
|
||||
(defn root-resolver [_ include]
|
||||
include))
|
||||
|
||||
#?(:clj
|
||||
(defn adaptive-resolver [source include]
|
||||
(let [include (or (io/resource include)
|
||||
include)]
|
||||
(if (string? include)
|
||||
(relative-resolver source include)
|
||||
include)))
|
||||
:cljs
|
||||
(defn adaptive-resolver [source include]
|
||||
(let [fl (if (path/isAbsolute include)
|
||||
include
|
||||
(path/join source ".." include))]
|
||||
(if (fs/existsSync fl)
|
||||
fl
|
||||
(source-logging-push-back-reader
|
||||
(pr-str {:aero/missing-include include}))))))
|
||||
|
||||
|
||||
(def default-opts
|
||||
{:profile :default
|
||||
:resolver adaptive-resolver})
|
||||
|
||||
;; The rationale for deferreds is to realise some values after the
|
||||
;; config has been read. This allows certain expensive operations to
|
||||
;; be performed only after #profile has had a chance to filter out all
|
||||
;; other environments. For example, a :prod profile my do some
|
||||
;; expensive decryption of secrets (which may not be cheap to run for
|
||||
;; all environments which don't need them, and probably won't be
|
||||
;; possible to decrypt, therefore you want to defer until needed).
|
||||
|
||||
(defn- realize-deferreds
|
||||
[config]
|
||||
(postwalk (fn [x] (if (instance? Deferred x) @(:delegate x) x)) config))
|
||||
|
||||
(defn- ref-meta-to-tagged-literal
|
||||
[config]
|
||||
(postwalk
|
||||
(fn [v]
|
||||
(cond
|
||||
(tagged-literal? v)
|
||||
(tagged-literal (:tag v) (ref-meta-to-tagged-literal (:form v)))
|
||||
|
||||
(contains? (meta v) :ref)
|
||||
(tagged-literal 'ref v)
|
||||
|
||||
:else
|
||||
v))
|
||||
config))
|
||||
|
||||
(defn- read-pr-into-tagged-literal
|
||||
[pr]
|
||||
(ref-meta-to-tagged-literal
|
||||
(edn/read
|
||||
{:eof nil
|
||||
;; Make a wrapper of all known readers, this permits mixing of
|
||||
;; post-processed tags with declared data readers
|
||||
:readers (into
|
||||
{}
|
||||
(map (fn [[k v]] [k #(tagged-literal k %)])
|
||||
(merge default-data-readers *data-readers*)))
|
||||
:default tagged-literal}
|
||||
pr)))
|
||||
|
||||
(defn read-config-into-tagged-literal
|
||||
[source]
|
||||
#?(:clj
|
||||
(with-open [pr (-> source io/reader clojure.lang.LineNumberingPushbackReader.)]
|
||||
(try
|
||||
(read-pr-into-tagged-literal pr)
|
||||
(catch Exception e
|
||||
(let [line (.getLineNumber pr)]
|
||||
(throw (ex-info (#?(:clj format :cljs gstring/format) "Config error on line %s" line) {:line line} e))))))
|
||||
:cljs
|
||||
(read-pr-into-tagged-literal
|
||||
(cond
|
||||
(tools.reader.reader-types/source-logging-reader? source)
|
||||
source
|
||||
|
||||
(implements? tools.reader.reader-types/Reader source)
|
||||
(source-logging-push-back-reader source)
|
||||
|
||||
:else
|
||||
(source-logging-push-back-reader
|
||||
(fs/readFileSync source "utf-8")
|
||||
1
|
||||
source)))))
|
||||
|
||||
(defn- rewrap
|
||||
[tl]
|
||||
(fn [v]
|
||||
(tagged-literal (:tag tl) v)))
|
||||
|
||||
(defmethod eval-tagged-literal :default
|
||||
[tl opts env ks]
|
||||
(let [{:keys [:aero.core/incomplete?] :as expansion}
|
||||
(expand (:form tl) opts env ks)]
|
||||
(if incomplete?
|
||||
(update expansion ::value (rewrap tl))
|
||||
(update expansion ::value #(reader opts (:tag tl) %)))))
|
||||
|
||||
(defmethod eval-tagged-literal 'ref
|
||||
[tl opts env ks]
|
||||
(let [{:keys [:aero.core/incomplete? :aero.core/env :aero.core/value
|
||||
:aero.core/incomplete]
|
||||
:or {env env}
|
||||
:as expansion} (expand (:form tl) opts env ks)]
|
||||
(if (or incomplete? (not (contains? env value)))
|
||||
(-> expansion
|
||||
(assoc ::incomplete? true)
|
||||
(update ::value (rewrap tl))
|
||||
(assoc ::incomplete (or incomplete
|
||||
{::path (pop ks)
|
||||
::value tl})))
|
||||
(assoc expansion ::value (get env value)))))
|
||||
|
||||
(defmethod eval-tagged-literal 'profile
|
||||
[tl opts env ks]
|
||||
(expand-case (:profile opts) tl opts env ks))
|
||||
|
||||
(defmethod eval-tagged-literal 'hostname
|
||||
[tl {:keys [hostname] :as opts} env ks]
|
||||
(expand-case (or hostname #?(:clj (env "HOSTNAME")
|
||||
:cljs (os/hostname)))
|
||||
tl opts env ks))
|
||||
|
||||
(defmethod eval-tagged-literal 'user
|
||||
[tl {:keys [user] :as opts} env ks]
|
||||
(expand-case (or user (get-env "USER"))
|
||||
tl opts env ks))
|
||||
|
||||
(defmethod eval-tagged-literal 'or
|
||||
[tl opts env ks]
|
||||
(let [{:keys [:aero.core/incomplete? :aero.core/value] :as expansion}
|
||||
(expand-scalar-repeatedly (:form tl) opts env ks)]
|
||||
(if incomplete?
|
||||
(update expansion ::value rewrap)
|
||||
(loop [[x & xs] value
|
||||
idx 0]
|
||||
(let [{:keys [:aero.core/incomplete? :aero.core/value]
|
||||
:as expansion}
|
||||
(expand x opts env (conj ks idx))]
|
||||
(cond
|
||||
;; We skipped a value, we cannot be sure whether it will be true in the future, so return with the remainder to check (including the skipped)
|
||||
incomplete?
|
||||
{::value (tagged-literal (:tag tl) (cons value xs))
|
||||
::incomplete? true
|
||||
::incomplete (::incomplete expansion)}
|
||||
|
||||
;; We found a value, and it's truthy, and we aren't skipped (because order), we successfully got one!
|
||||
value
|
||||
expansion
|
||||
|
||||
;; Run out of things to check
|
||||
(not (seq xs))
|
||||
nil
|
||||
|
||||
:else
|
||||
;; Falsey value, but not skipped, recur with the rest to try
|
||||
(recur xs (inc idx))))))))
|
||||
|
||||
(defn- assoc-in-kv-seq
|
||||
[x ks v]
|
||||
(let [[k & ks] ks]
|
||||
(let [steps (if (tagged-literal? x)
|
||||
(with-meta
|
||||
[[:tag (:tag x)]
|
||||
[:form (:form x)]]
|
||||
{`reassemble (fn [this queue]
|
||||
(let [{:keys [tag form]} (into {} queue)]
|
||||
(tagged-literal tag form)))})
|
||||
(kv-seq x))]
|
||||
(reassemble
|
||||
steps
|
||||
(map (fn [[stepk stepv :as kv]]
|
||||
(cond
|
||||
(and (not= (first ks) ::k)
|
||||
(= stepk k))
|
||||
(if (seq ks)
|
||||
[stepk (assoc-in-kv-seq stepv ks v)]
|
||||
[stepk v])
|
||||
|
||||
(and (= (first ks) ::k)
|
||||
(= stepk k))
|
||||
(if (seq (rest ks))
|
||||
[(assoc-in-kv-seq stepk (rest ks) v) stepv]
|
||||
[v stepv])
|
||||
|
||||
:else
|
||||
kv))
|
||||
steps)))))
|
||||
|
||||
(defn- dissoc-in-kv-seq
|
||||
[x ks]
|
||||
(let [[k & ks] ks]
|
||||
(if
|
||||
(or (not (seq ks))
|
||||
(= [::k] ks))
|
||||
(let [steps (kv-seq x)]
|
||||
(reassemble
|
||||
steps
|
||||
(filter (fn [[stepk stepv :as kv]]
|
||||
(not= stepk k))
|
||||
steps)) )
|
||||
|
||||
(let [steps (if (tagged-literal? x)
|
||||
(with-meta
|
||||
[[:tag (:tag x)]
|
||||
[:form (:form x)]]
|
||||
{`reassemble (fn [this queue]
|
||||
(let [{:keys [tag form]} (into {} queue)]
|
||||
(tagged-literal tag form)))})
|
||||
(kv-seq x))]
|
||||
(reassemble
|
||||
steps
|
||||
(map (fn [[stepk stepv :as kv]]
|
||||
(cond
|
||||
(and (not= (first ks) ::k)
|
||||
(= stepk k))
|
||||
(if (seq ks)
|
||||
[stepk (dissoc-in-kv-seq stepv ks)]
|
||||
[stepk stepv])
|
||||
|
||||
(and (= (first ks) ::k)
|
||||
(= stepk k))
|
||||
(if (seq (rest ks))
|
||||
[(dissoc-in-kv-seq stepk (rest ks)) stepv]
|
||||
[stepk stepv])
|
||||
|
||||
:else
|
||||
kv))
|
||||
steps))))))
|
||||
|
||||
(defn resolve-tagged-literals
|
||||
[wrapped-config opts]
|
||||
(let [{:keys [:aero.core/incomplete?
|
||||
:aero.core/value]
|
||||
:as expansion}
|
||||
(loop [attempts 0
|
||||
x {::value wrapped-config
|
||||
::incomplete? true}]
|
||||
(let [{:keys [:aero.core/incomplete]
|
||||
:as expansion}
|
||||
(expand (::value x)
|
||||
opts
|
||||
(::env x {})
|
||||
[])]
|
||||
(cond
|
||||
(not (::incomplete? x))
|
||||
expansion
|
||||
|
||||
(and (> attempts 0)
|
||||
(= (-> incomplete ::value :tag) 'ref))
|
||||
(do
|
||||
(binding [*out* #?(:clj *err*
|
||||
:cljs *out*)]
|
||||
(println "WARNING: Unable to resolve"
|
||||
(str \" (pr-str (-> incomplete ::value)) \")
|
||||
"at"
|
||||
(pr-str (-> incomplete ::path))))
|
||||
(recur
|
||||
0
|
||||
(if (= ::k (-> incomplete ::path last))
|
||||
(update expansion
|
||||
::value
|
||||
dissoc-in-kv-seq
|
||||
(-> incomplete ::path))
|
||||
(update expansion
|
||||
::value
|
||||
assoc-in-kv-seq
|
||||
(-> incomplete ::path)
|
||||
nil))))
|
||||
|
||||
(> attempts 1)
|
||||
(throw (ex-info "Max attempts exhausted"
|
||||
{:progress x
|
||||
:attempts attempts}))
|
||||
|
||||
:else
|
||||
(recur (if (= x expansion)
|
||||
(inc attempts)
|
||||
0)
|
||||
expansion))))]
|
||||
(if incomplete?
|
||||
(throw (ex-info "Incomplete resolution" expansion))
|
||||
value)))
|
||||
|
||||
(defn read-config
|
||||
"First argument is a string URL to the file. To read from the
|
||||
current directory just put the file name. To read from the classpath
|
||||
call clojure.java.io/resource on the string before passing it into
|
||||
this function.
|
||||
Optional second argument is a map that can include
|
||||
the following keys:
|
||||
:profile - indicates the profile to use for #profile extension
|
||||
:user - manually set the user for the #user extension
|
||||
:resolver - a function or map used to resolve includes."
|
||||
([source given-opts]
|
||||
(let [opts (merge default-opts given-opts {:source source})
|
||||
wrapped-config (read-config-into-tagged-literal source)]
|
||||
(-> wrapped-config
|
||||
(resolve-tagged-literals opts)
|
||||
(realize-deferreds))))
|
||||
([source] (read-config source {})))
|
||||
)
|
||||
|
||||
(macro/deftime
|
||||
(defmacro deferred [& expr]
|
||||
`(->Deferred (delay ~@expr))))
|
||||
+196
@@ -0,0 +1,196 @@
|
||||
(ns conman.core
|
||||
(:require [clojure.java.io :as io]
|
||||
[clojure.set :refer [rename-keys]]
|
||||
[hikari-cp.core :refer [make-datasource]]
|
||||
[hugsql.core :as hugsql]
|
||||
[hugsql.adapter.next-jdbc :as next-adapter]
|
||||
[to-jdbc-uri.core :refer [to-jdbc-uri]])
|
||||
(:import [clojure.lang IDeref]))
|
||||
|
||||
(hugsql/set-adapter! (next-adapter/hugsql-adapter-next-jdbc))
|
||||
|
||||
(defn validate-files [filenames]
|
||||
(doseq [file filenames]
|
||||
(when-not (or (instance? java.io.File file) (io/resource file))
|
||||
(throw (Exception. (str "conman could not find the query file:" file))))))
|
||||
|
||||
(defn try-snip [[id snip]]
|
||||
[id
|
||||
(update snip :fn
|
||||
(fn [snip]
|
||||
(fn [& args]
|
||||
(try (apply snip args)
|
||||
(catch Exception e
|
||||
(throw (ex-info (ex-message e) {:snip-id id} e)))))))])
|
||||
|
||||
(defn try-query [[id query]]
|
||||
[id
|
||||
(update query :fn
|
||||
(fn [query]
|
||||
(fn
|
||||
([conn params]
|
||||
(try (query conn params)
|
||||
(catch Exception e
|
||||
(throw (ex-info (ex-message e) {:query-id id} e)))))
|
||||
([conn params opts & command-opts]
|
||||
(try (apply query conn params opts command-opts)
|
||||
(catch Exception e
|
||||
(throw (ex-info (ex-message e) {:query-id id} e))))))))])
|
||||
|
||||
(defn load-queries [& args]
|
||||
(let [options? (map? (first args))
|
||||
options (if options? (first args) {})
|
||||
filenames (if options? (rest args) args)]
|
||||
(validate-files filenames)
|
||||
(reduce
|
||||
(fn [queries file]
|
||||
(let [{snips true
|
||||
fns false}
|
||||
(group-by
|
||||
#(-> % second :meta :snip? boolean)
|
||||
(hugsql/map-of-db-fns file options))]
|
||||
(-> queries
|
||||
(update :snips (fnil into {}) (mapv try-snip snips))
|
||||
(update :fns (fnil into {}) (mapv try-query fns)))))
|
||||
{}
|
||||
filenames)))
|
||||
|
||||
(defn intern-fn [ns id meta f]
|
||||
(intern ns (with-meta (symbol (name id)) meta) f))
|
||||
|
||||
(defmacro bind-connection [conn & filenames]
|
||||
`(let [{snips# :snips fns# :fns :as queries#} (conman.core/load-queries ~@filenames)]
|
||||
(doseq [[id# {fn# :fn meta# :meta}] snips#]
|
||||
(conman.core/intern-fn *ns* id# meta# fn#))
|
||||
(doseq [[id# {query# :fn meta# :meta}] fns#]
|
||||
(conman.core/intern-fn *ns* id#
|
||||
;; Need to explicitly set :arglists since we don't use defn.
|
||||
;; Another option would be to generate defns.
|
||||
(assoc meta#
|
||||
:arglists (quote ~'([] [params] [db params options & command-options])))
|
||||
(fn f#
|
||||
([] (query# ~conn {}))
|
||||
([params#] (query# ~conn params#))
|
||||
([conn# params# & args#] (apply query# conn# params# args#)))))
|
||||
queries#))
|
||||
|
||||
(defmacro bind-connection-deref [conn & filenames]
|
||||
`(let [{snips# :snips fns# :fns :as queries#} (conman.core/load-queries ~@filenames)]
|
||||
(doseq [[id# {fn# :fn meta# :meta}] snips#]
|
||||
(conman.core/intern-fn *ns* id# meta# fn#))
|
||||
(doseq [[id# {query# :fn meta# :meta}] fns#]
|
||||
(conman.core/intern-fn *ns* id#
|
||||
(assoc meta#
|
||||
:arglists (quote ~'([] [params] [db params options & command-options])))
|
||||
(fn f#
|
||||
([] (query# (deref ~conn) {}))
|
||||
([params#] (query# (deref ~conn) params#))
|
||||
([conn# params# & args#] (apply query# conn# params# args#)))))
|
||||
queries#))
|
||||
|
||||
(defn bind-connection-map [conn & args]
|
||||
(-> (apply load-queries args)
|
||||
(update :snips
|
||||
(fn [snips]
|
||||
(reduce (fn [acc [id snip]] (assoc acc id snip)) {} snips)))
|
||||
(update :fns
|
||||
(fn [queries]
|
||||
(reduce
|
||||
(fn [acc [id query]]
|
||||
(assoc acc id
|
||||
(update query
|
||||
:fn
|
||||
(fn [query]
|
||||
(fn fn#
|
||||
([] (query conn {}))
|
||||
([params]
|
||||
(query conn params))
|
||||
([conn params & args] (apply query conn params args)))))))
|
||||
{}
|
||||
queries)))))
|
||||
|
||||
(defn find-fn [connection-map query-type k]
|
||||
(or (get-in connection-map [query-type k :fn])
|
||||
(throw (IllegalArgumentException.
|
||||
(str (if (= query-type :snips) "no snippet" "no query")
|
||||
" found for the key: " k
|
||||
"', available queries: " (keys (get connection-map query-type)))))))
|
||||
|
||||
(defn snip [connection-map snip-key & args]
|
||||
"runs a SQL query snippet
|
||||
queries - a map of queries
|
||||
id - keyword indicating the name of the query
|
||||
args - arguments that will be passed to the query"
|
||||
(apply (find-fn connection-map :snips snip-key) args))
|
||||
|
||||
(defn query
|
||||
"runs a database query and returns the result
|
||||
conn - database connection
|
||||
queries - a map of queries
|
||||
id - keyword indicating the name of the query
|
||||
args - arguments that will be passed to the query"
|
||||
([connection-map query-key]
|
||||
((find-fn connection-map :fns query-key)))
|
||||
([connection-map query-key params]
|
||||
((find-fn connection-map :fns query-key) params))
|
||||
([conn connection-map query-key params & opts]
|
||||
(apply (find-fn connection-map :fns query-key) conn params opts)))
|
||||
|
||||
(defn- format-url [pool-spec]
|
||||
(if (:jdbc-url pool-spec)
|
||||
(update pool-spec :jdbc-url to-jdbc-uri)
|
||||
pool-spec))
|
||||
|
||||
(defn make-config [{:keys [jdbc-url adapter datasource datasource-classname] :as pool-spec}]
|
||||
(when (not (or jdbc-url adapter datasource datasource-classname))
|
||||
(throw (Exception. "one of :jdbc-url, :adapter, :datasource, or :datasource-classname is required to initialize the connection!")))
|
||||
(-> pool-spec
|
||||
(format-url)
|
||||
(rename-keys
|
||||
{:auto-commit? :auto-commit
|
||||
:conn-timeout :connection-timeout
|
||||
:min-idle :minimum-idle
|
||||
:max-pool-size :maximum-pool-size})))
|
||||
|
||||
(defn connect!
|
||||
"attempts to create a new connection and set it as the value of the conn atom,
|
||||
does nothing if conn atom is already populated"
|
||||
[pool-spec]
|
||||
(make-datasource (make-config pool-spec)))
|
||||
|
||||
(defn disconnect!
|
||||
"checks if there's a connection and closes it
|
||||
resets the conn to nil"
|
||||
[conn]
|
||||
(when (and (instance? com.zaxxer.hikari.HikariDataSource conn)
|
||||
(not (.isClosed conn)))
|
||||
(.close conn)))
|
||||
|
||||
(defn reconnect!
|
||||
"calls disconnect! to ensure the connection is closed
|
||||
then calls connect! to establish a new connection"
|
||||
[conn pool-spec]
|
||||
(disconnect! conn)
|
||||
(connect! pool-spec))
|
||||
|
||||
(extend-protocol next.jdbc.protocols/Sourceable
|
||||
IDeref
|
||||
(get-datasource [this]
|
||||
(next.jdbc.protocols/get-datasource (deref this))))
|
||||
|
||||
(defmacro with-transaction
|
||||
"Runs the body in a transaction where t-conn is the name of the transaction connection.
|
||||
The body will be evaluated within a binding where conn is set to the transactional
|
||||
connection. The isolation level and readonly status of the transaction may also be specified.
|
||||
(with-transaction [conn {:isolation level :read-only? true}]
|
||||
... t-conn ...)
|
||||
See next.jdbc/transact for more details on the semantics of the :isolation and
|
||||
:read-only options."
|
||||
[[dbsym & opts] & body]
|
||||
`(if (instance? IDeref ~dbsym)
|
||||
(next.jdbc/with-transaction [t-conn# (deref ~dbsym) ~@opts]
|
||||
(binding [~dbsym (delay t-conn#)]
|
||||
~@body))
|
||||
(next.jdbc/with-transaction [t-conn# ~dbsym ~@opts]
|
||||
(binding [~dbsym t-conn#]
|
||||
~@body))))
|
||||
@@ -0,0 +1,70 @@
|
||||
(ns kit.edge.db.postgres
|
||||
(:require
|
||||
[cheshire.core :as cheshire]
|
||||
[next.jdbc]
|
||||
[next.jdbc.prepare :as prepare]
|
||||
[next.jdbc.result-set :as result-set])
|
||||
(:import
|
||||
[clojure.lang IPersistentMap IPersistentVector]
|
||||
[java.sql Array PreparedStatement Timestamp]
|
||||
[java.time Instant LocalDate LocalDateTime]
|
||||
[org.postgresql.util PGobject]))
|
||||
|
||||
(def ->json cheshire/generate-string)
|
||||
(def <-json #(cheshire/parse-string % true))
|
||||
|
||||
(defn ->pgobject
|
||||
"Transforms Clojure data to a PGobject that contains the data as
|
||||
JSON. PGObject type defaults to `jsonb` but can be changed via
|
||||
metadata key `:pgtype`"
|
||||
[x]
|
||||
(let [pgtype (:pgtype (meta x) "jsonb")]
|
||||
(doto (PGobject.)
|
||||
(.setType pgtype)
|
||||
(.setValue (->json x)))))
|
||||
|
||||
(defn <-pgobject
|
||||
"Transform PGobject containing `json` or `jsonb` value to Clojure data"
|
||||
[^PGobject v]
|
||||
(let [type (.getType v)
|
||||
value (.getValue v)]
|
||||
(if (#{"jsonb" "json"} type)
|
||||
(when value
|
||||
(with-meta (<-json value) {:pgtype type}))
|
||||
value)))
|
||||
|
||||
(extend-protocol result-set/ReadableColumn
|
||||
Array
|
||||
(read-column-by-label [^Array v _] (vec (.getArray v)))
|
||||
(read-column-by-index [^Array v _2 _3] (vec (.getArray v)))
|
||||
|
||||
PGobject
|
||||
(read-column-by-label [^PGobject v _] (<-pgobject v))
|
||||
(read-column-by-index [^PGobject v _2 _3] (<-pgobject v)))
|
||||
|
||||
(extend-protocol prepare/SettableParameter
|
||||
Instant
|
||||
(set-parameter [^Instant v ^PreparedStatement ps ^long i]
|
||||
(.setTimestamp ps i (Timestamp/from v)))
|
||||
|
||||
LocalDate
|
||||
(set-parameter [^LocalDate v ^PreparedStatement ps ^long i]
|
||||
(.setTimestamp ps i (Timestamp/valueOf (.atStartOfDay v))))
|
||||
|
||||
LocalDateTime
|
||||
(set-parameter [^LocalDateTime v ^PreparedStatement ps ^long i]
|
||||
(.setTimestamp ps i (Timestamp/valueOf v)))
|
||||
|
||||
IPersistentMap
|
||||
(set-parameter [m ^PreparedStatement s i]
|
||||
(.setObject s i (->pgobject m)))
|
||||
|
||||
IPersistentVector
|
||||
(set-parameter [^clojure.lang.IPersistentVector v ^java.sql.PreparedStatement stmt ^long idx]
|
||||
(let [conn (.getConnection stmt)
|
||||
meta (.getParameterMetaData stmt)
|
||||
type-name (.getParameterTypeName meta idx)]
|
||||
(if-let [elem-type (when (= (first type-name) \_)
|
||||
(apply str (rest type-name)))]
|
||||
(.setObject stmt idx (.createArrayOf conn elem-type (to-array v)))
|
||||
(.setObject stmt idx (->pgobject v))))))
|
||||
@@ -0,0 +1,59 @@
|
||||
(ns kit.edge.db.sql.conman
|
||||
(:require
|
||||
[clojure.tools.logging :as log]
|
||||
[conman.core :as conman]
|
||||
[integrant.core :as ig]
|
||||
[kit.ig-utils :as ig-utils]))
|
||||
|
||||
(defmethod ig/init-key :db.sql/connection
|
||||
[_ pool-spec]
|
||||
(conman/connect! pool-spec))
|
||||
|
||||
(defmethod ig/suspend-key! :db.sql/connection [_ _])
|
||||
|
||||
(defmethod ig/halt-key! :db.sql/connection
|
||||
[_ conn]
|
||||
(conman/disconnect! conn))
|
||||
|
||||
(defmethod ig/resume-key :db.sql/connection
|
||||
[key opts old-opts old-impl]
|
||||
(ig-utils/resume-handler key opts old-opts old-impl))
|
||||
|
||||
(defn queries-dev [load-queries]
|
||||
(fn
|
||||
([query params]
|
||||
(conman/query (load-queries) query params))
|
||||
([conn query params & opts]
|
||||
(conman/query conn (load-queries) query params opts))))
|
||||
|
||||
(defn queries-prod [load-queries]
|
||||
(let [queries (load-queries)]
|
||||
(fn
|
||||
([query params]
|
||||
(conman/query queries query params))
|
||||
([conn query params & opts]
|
||||
(conman/query conn queries query params opts)))))
|
||||
|
||||
(defmethod ig/init-key :db.sql/query-fn
|
||||
[_ {:keys [conn options filename filenames env]
|
||||
:or {options {}}}]
|
||||
(let [filenames (or filenames [filename])
|
||||
load-queries #(apply conman/bind-connection-map conn options filenames)]
|
||||
(with-meta
|
||||
(if (= env :dev)
|
||||
(queries-dev load-queries)
|
||||
(queries-prod load-queries))
|
||||
{:mtimes (mapv ig-utils/last-modified filenames)})))
|
||||
|
||||
(defmethod ig/suspend-key! :db.sql/query-fn [_ _])
|
||||
|
||||
(defmethod ig/resume-key :db.sql/query-fn
|
||||
[k {:keys [filename filenames] :as opts} old-opts old-impl]
|
||||
(let [check-res (and (= opts old-opts)
|
||||
(= (mapv ig-utils/last-modified (or filenames [filename]))
|
||||
(:mtimes (meta old-impl))))]
|
||||
(log/info k "resume check. Same?" check-res)
|
||||
(if check-res
|
||||
old-impl
|
||||
(do (ig/halt-key! k old-impl)
|
||||
(ig/init-key k opts)))))
|
||||
+27
-3
@@ -47,9 +47,33 @@
|
||||
:env #ig/ref :system/env}
|
||||
|
||||
:db.sql/connection
|
||||
#profile {:dev {:jdbc-url #env JDBC_URL}
|
||||
:test {:jdbc-url #env JDBC_URL}
|
||||
:prod {:jdbc-url #env JDBC_URL
|
||||
#profile {:dev {:jdbc-url #or [#env JDBC_URL
|
||||
#join ["jdbc:postgresql://"
|
||||
#or [#env TARGET_HOST "localhost"]
|
||||
":"
|
||||
#or [#env TARGET_PORT "5432"]
|
||||
"/"
|
||||
#or [#env TARGET_DB "elprice"]]]
|
||||
:username #or [#env TARGET_USER "elprice"]
|
||||
:password #or [#env TARGET_PASSWORD "elprice"]}
|
||||
:test {:jdbc-url #or [#env JDBC_URL
|
||||
#join ["jdbc:postgresql://"
|
||||
#or [#env TARGET_HOST "localhost"]
|
||||
":"
|
||||
#or [#env TARGET_PORT "5432"]
|
||||
"/"
|
||||
#or [#env TARGET_DB "elprice"]]]
|
||||
:username #or [#env TARGET_USER "elprice"]
|
||||
:password #or [#env TARGET_PASSWORD "elprice"]}
|
||||
:prod {:jdbc-url #or [#env JDBC_URL
|
||||
#join ["jdbc:postgresql://"
|
||||
#or [#env TARGET_HOST "localhost"]
|
||||
":"
|
||||
#or [#env TARGET_PORT "5432"]
|
||||
"/"
|
||||
#or [#env TARGET_DB "elprice"]]]
|
||||
:username #or [#env TARGET_USER "elprice"]
|
||||
:password #or [#env TARGET_PASSWORD "elprice"]
|
||||
:init-size 1
|
||||
:min-idle 1
|
||||
:max-idle 8
|
||||
|
||||
Reference in New Issue
Block a user