Compare commits
9
Commits
3a319f5cf4
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
8de68da1d3 | ||
|
|
22aa590de2 | ||
|
|
9134b891a8 | ||
|
|
fc071af990 | ||
|
|
e088d67308 | ||
|
|
1958ac9c5a | ||
|
|
da20691d2b | ||
|
|
06b575d709 | ||
|
|
ce1c3822a2 |
+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))))
|
||||
Executable
+43
@@ -0,0 +1,43 @@
|
||||
#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
|
||||
HOST="${ELPRICE_DEPLOY_HOST:-citadel}"
|
||||
APP_DIR="${ELPRICE_DEPLOY_DIR:-/home/pmp/apps/elprice}"
|
||||
IMAGE_NAME="${ELPRICE_IMAGE_NAME:-$(basename "$APP_DIR")-app:latest}"
|
||||
PLATFORM="${ELPRICE_PLATFORM:-}"
|
||||
FILES=(docker-compose.yml .env .token)
|
||||
|
||||
if [[ -z "$PLATFORM" ]]; then
|
||||
remote_arch="$(ssh "$HOST" uname -m)"
|
||||
case "$remote_arch" in
|
||||
x86_64 | amd64) PLATFORM="linux/amd64" ;;
|
||||
aarch64 | arm64) PLATFORM="linux/arm64" ;;
|
||||
*)
|
||||
echo "Unsupported target architecture: $remote_arch" >&2
|
||||
echo "Set ELPRICE_PLATFORM manually, e.g. ELPRICE_PLATFORM=linux/amd64" >&2
|
||||
exit 1
|
||||
;;
|
||||
esac
|
||||
fi
|
||||
|
||||
for file in "${FILES[@]}"; do
|
||||
if [[ ! -f "$file" ]]; then
|
||||
echo "Missing required file: $file" >&2
|
||||
exit 1
|
||||
fi
|
||||
done
|
||||
|
||||
echo "Building $IMAGE_NAME locally for $PLATFORM..."
|
||||
docker build --platform "$PLATFORM" -t "$IMAGE_NAME" .
|
||||
|
||||
echo "Preparing $APP_DIR on $HOST..."
|
||||
ssh "$HOST" "mkdir -p '$APP_DIR'"
|
||||
|
||||
echo "Copying docker-compose.yml, .env, and .token to $HOST:$APP_DIR..."
|
||||
scp "${FILES[@]}" "$HOST:$APP_DIR/"
|
||||
|
||||
echo "Copying Docker image $IMAGE_NAME to $HOST..."
|
||||
docker save "$IMAGE_NAME" | ssh "$HOST" docker load
|
||||
|
||||
echo "Starting docker compose on $HOST in $APP_DIR..."
|
||||
ssh "$HOST" "cd '$APP_DIR' && docker compose up -d --no-build"
|
||||
@@ -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)))))
|
||||
@@ -0,0 +1 @@
|
||||
DROP TABLE solar;
|
||||
@@ -0,0 +1,10 @@
|
||||
CREATE TABLE solar (
|
||||
period_start TIMESTAMPTZ PRIMARY KEY,
|
||||
pv_yield_kwh DOUBLE PRECISION NOT NULL,
|
||||
inverter_yield_kwh DOUBLE PRECISION NOT NULL,
|
||||
export_kwh DOUBLE PRECISION NOT NULL,
|
||||
import_kwh DOUBLE PRECISION NOT NULL,
|
||||
charge_kwh DOUBLE PRECISION NOT NULL,
|
||||
discharge_kwh DOUBLE PRECISION NOT NULL,
|
||||
revenue_eur DOUBLE PRECISION NOT NULL
|
||||
);
|
||||
@@ -0,0 +1 @@
|
||||
DROP TABLE elprice;
|
||||
@@ -0,0 +1,6 @@
|
||||
CREATE TABLE elprice (
|
||||
time_dk TIMESTAMP NOT NULL,
|
||||
price_area VARCHAR(3) NOT NULL,
|
||||
price_dkk NUMERIC(10,2) NOT NULL,
|
||||
PRIMARY KEY (time_dk, price_area)
|
||||
);
|
||||
@@ -0,0 +1,3 @@
|
||||
ALTER TABLE elprice DROP COLUMN total_dkk;
|
||||
--;;
|
||||
ALTER TABLE elprice DROP COLUMN tariff_dkk;
|
||||
@@ -0,0 +1,21 @@
|
||||
ALTER TABLE elprice ADD COLUMN tariff_dkk NUMERIC(10,2);
|
||||
--;;
|
||||
ALTER TABLE elprice ADD COLUMN total_dkk NUMERIC(10,2);
|
||||
--;;
|
||||
UPDATE elprice
|
||||
SET tariff_dkk = CASE
|
||||
WHEN EXTRACT(MONTH FROM time_dk) BETWEEN 4 AND 9 THEN
|
||||
CASE
|
||||
WHEN EXTRACT(HOUR FROM time_dk) < 6 THEN 109.80
|
||||
WHEN EXTRACT(HOUR FROM time_dk) >= 17 AND EXTRACT(HOUR FROM time_dk) < 21 THEN 428.30
|
||||
ELSE 164.70
|
||||
END
|
||||
ELSE
|
||||
CASE
|
||||
WHEN EXTRACT(HOUR FROM time_dk) < 6 THEN 109.80
|
||||
WHEN EXTRACT(HOUR FROM time_dk) >= 17 AND EXTRACT(HOUR FROM time_dk) < 21 THEN 988.40
|
||||
ELSE 329.50
|
||||
END
|
||||
END;
|
||||
--;;
|
||||
UPDATE elprice SET total_dkk = price_dkk + tariff_dkk WHERE tariff_dkk IS NOT NULL;
|
||||
@@ -0,0 +1,2 @@
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS idx_day_ahead_prices_time_dk_area
|
||||
ON day_ahead_prices (time_dk, price_area);
|
||||
@@ -0,0 +1 @@
|
||||
DROP INDEX IF EXISTS idx_day_ahead_prices_time_dk_area;
|
||||
+32
-2
@@ -1,8 +1,11 @@
|
||||
-- :name insert-price! :! :n
|
||||
-- :doc Insert a day-ahead price record, skip if already exists
|
||||
-- :doc Insert/update a day-ahead price record
|
||||
INSERT INTO day_ahead_prices (time_utc, time_dk, price_area, price_dkk, price_eur)
|
||||
VALUES (:time-utc, :time-dk, :price-area, :price-dkk, :price-eur)
|
||||
ON CONFLICT (time_dk, price_area) DO NOTHING;
|
||||
ON CONFLICT (time_utc, price_area) DO UPDATE SET
|
||||
time_dk = EXCLUDED.time_dk,
|
||||
price_dkk = EXCLUDED.price_dkk,
|
||||
price_eur = EXCLUDED.price_eur;
|
||||
|
||||
-- :name get-prices-for-date :? :*
|
||||
-- :doc Get all prices for a given date (time_dk local date) and price area
|
||||
@@ -54,6 +57,18 @@ SELECT count(*) AS cnt
|
||||
FROM consumption
|
||||
WHERE time_start = :time-start;
|
||||
|
||||
-- :name count-production-for-date :? :1
|
||||
-- :doc Count production records for a given UTC time_start
|
||||
SELECT count(*) AS cnt
|
||||
FROM production
|
||||
WHERE time_start = :time-start;
|
||||
|
||||
-- :name count-meter-readings-for-date :? :1
|
||||
-- :doc Count meter readings for a given UTC time_start (converted to DK date range)
|
||||
SELECT count(*) AS cnt
|
||||
FROM meter_readings
|
||||
WHERE time_dk >= :from-dk AND time_dk < :to-dk;
|
||||
|
||||
-- :name insert-production! :! :n
|
||||
-- :doc Insert a production record, skip if already exists
|
||||
INSERT INTO production (time_start, hour, kwh)
|
||||
@@ -65,3 +80,18 @@ ON CONFLICT (time_start, hour) DO NOTHING;
|
||||
INSERT INTO consumption (time_start, hour, kwh)
|
||||
VALUES (:time-start, :hour, :kwh)
|
||||
ON CONFLICT (time_start, hour) DO NOTHING;
|
||||
|
||||
-- :name insert-elprice! :! :n
|
||||
-- :doc Upsert an hourly el-price row; refresh tariff/total on conflict
|
||||
INSERT INTO elprice (time_dk, price_area, price_dkk, tariff_dkk, total_dkk)
|
||||
VALUES (:time-dk, :price-area, :price-dkk, :tariff-dkk, :total-dkk)
|
||||
ON CONFLICT (time_dk, price_area) DO UPDATE SET
|
||||
tariff_dkk = EXCLUDED.tariff_dkk,
|
||||
total_dkk = EXCLUDED.total_dkk;
|
||||
|
||||
-- :name get-elprice-for-date :? :*
|
||||
-- :doc Hourly prices for a DK local date, both areas
|
||||
SELECT time_dk, price_area, price_dkk, tariff_dkk, total_dkk
|
||||
FROM elprice
|
||||
WHERE time_dk::date = :date
|
||||
ORDER BY time_dk, price_area;
|
||||
|
||||
+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
|
||||
|
||||
Executable
+68
@@ -0,0 +1,68 @@
|
||||
#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
|
||||
# Run the app locally without Docker.
|
||||
# Loads environment variables from .env and the Eloverblik refresh token from .token.
|
||||
|
||||
ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
cd "$ROOT_DIR"
|
||||
|
||||
load_env_file() {
|
||||
local file="$1"
|
||||
|
||||
[[ -f "$file" ]] || return 0
|
||||
|
||||
while IFS= read -r line || [[ -n "$line" ]]; do
|
||||
# Trim CR from CRLF files.
|
||||
line="${line%$'\r'}"
|
||||
|
||||
# Skip blank lines and full-line comments.
|
||||
[[ -z "${line//[[:space:]]/}" || "$line" =~ ^[[:space:]]*# ]] && continue
|
||||
|
||||
# Allow lines prefixed with "export ".
|
||||
line="${line#export }"
|
||||
|
||||
local key="${line%%=*}"
|
||||
local value="${line#*=}"
|
||||
|
||||
# Trim whitespace around the key only; preserve value contents (e.g. JDBC_URL query strings).
|
||||
key="$(printf '%s' "$key" | xargs)"
|
||||
|
||||
if [[ ! "$key" =~ ^[A-Za-z_][A-Za-z0-9_]*$ || "$line" != *=* ]]; then
|
||||
echo "Skipping invalid env line in $file: $line" >&2
|
||||
continue
|
||||
fi
|
||||
|
||||
# Remove one matching pair of surrounding quotes, if present.
|
||||
if [[ "$value" =~ ^\".*\"$ || "$value" =~ ^\'.*\'$ ]]; then
|
||||
value="${value:1:${#value}-2}"
|
||||
fi
|
||||
|
||||
export "$key=$value"
|
||||
done < "$file"
|
||||
}
|
||||
|
||||
load_env_file ".env"
|
||||
|
||||
if [[ -f ".token" && -z "${ELOVERBLIK_TOKEN:-}" ]]; then
|
||||
ELOVERBLIK_TOKEN="$(tr -d '\r\n' < .token)"
|
||||
export ELOVERBLIK_TOKEN
|
||||
fi
|
||||
|
||||
: "${PORT:=3000}"
|
||||
export PORT
|
||||
|
||||
# Energi Data Service currently serves only the leaf certificate. Allow the JVM
|
||||
# to fetch missing intermediate CA certificates from the certificate AIA URL.
|
||||
export JAVA_TOOL_OPTIONS="${JAVA_TOOL_OPTIONS:-} -Dcom.sun.security.enableAIAcaIssuers=true"
|
||||
|
||||
echo "Starting elprice locally on port $PORT (no Docker)..."
|
||||
|
||||
# `clj` requires rlwrap for REPL command editing. Fall back to `clojure`
|
||||
# when rlwrap is not installed so the app can still start.
|
||||
if command -v rlwrap >/dev/null 2>&1; then
|
||||
exec clj -M:dev -e "(dev-prep!) (go)" -r
|
||||
else
|
||||
echo "rlwrap not found; using clojure instead of clj (REPL editing disabled)." >&2
|
||||
exec clojure -M:dev -e "(dev-prep!) (go)" -r
|
||||
fi
|
||||
@@ -286,15 +286,22 @@
|
||||
[query-fn ^LocalDate date]
|
||||
(let [yesterday (.minusDays (LocalDate/now dk-zone) 1)
|
||||
time-start (utc-start-for-date date)
|
||||
cnt (:cnt (query-fn :count-consumption-for-date
|
||||
{:time-start time-start}))]
|
||||
from-dk (Timestamp/valueOf (.atStartOfDay date))
|
||||
to-dk (Timestamp/valueOf (.atStartOfDay (.plusDays date 1)))
|
||||
cons-cnt (:cnt (query-fn :count-consumption-for-date
|
||||
{:time-start time-start}))
|
||||
prod-cnt (:cnt (query-fn :count-production-for-date
|
||||
{:time-start time-start}))
|
||||
read-cnt (:cnt (query-fn :count-meter-readings-for-date
|
||||
{:from-dk from-dk :to-dk to-dk}))]
|
||||
(cond
|
||||
(not (.isBefore date yesterday))
|
||||
(do (log/info "No data to fetch for" (.format date date-fmt) "(too recent)")
|
||||
{:no-data true :date (.format date date-fmt)})
|
||||
|
||||
(and cnt (pos? cnt))
|
||||
(do (log/info "Data already exists for" (.format date date-fmt))
|
||||
(and (some-> cons-cnt pos?) (some-> prod-cnt pos?) (some-> read-cnt pos?))
|
||||
(do (log/info "Data already exists for" (.format date date-fmt)
|
||||
"- consumption:" cons-cnt "production:" prod-cnt "readings:" read-cnt)
|
||||
{:exists true :date (.format date date-fmt)})
|
||||
:else
|
||||
(do
|
||||
|
||||
@@ -0,0 +1,112 @@
|
||||
(ns pmagnus.elprice.web.controllers.elprice
|
||||
(:require
|
||||
[clojure.data.json :as json]
|
||||
[clojure.tools.logging :as log]
|
||||
[pmagnus.elprice.web.controllers.tariffs :as tariffs])
|
||||
(:import
|
||||
(java.net
|
||||
URI)
|
||||
(java.net.http
|
||||
HttpClient
|
||||
HttpRequest
|
||||
HttpResponse$BodyHandlers)
|
||||
(java.sql
|
||||
Timestamp)
|
||||
(java.time
|
||||
LocalDate
|
||||
LocalDateTime)))
|
||||
|
||||
|
||||
(def ^:private eur-dkk-rate 7.46)
|
||||
|
||||
|
||||
(defn- elprisen-url
|
||||
[^LocalDate date area]
|
||||
(format "https://www.elprisenligenu.dk/api/v1/prices/%04d/%02d-%02d_%s.json"
|
||||
(.getYear date) (.getMonthValue date) (.getDayOfMonth date) area))
|
||||
|
||||
|
||||
(defn- fetch-json
|
||||
[url]
|
||||
(let [client (HttpClient/newHttpClient)
|
||||
request (-> (HttpRequest/newBuilder)
|
||||
(.uri (URI/create url))
|
||||
(.header "Accept" "application/json")
|
||||
(.GET)
|
||||
(.build))
|
||||
response (.send client request (HttpResponse$BodyHandlers/ofString))]
|
||||
(when (= 200 (.statusCode response))
|
||||
(json/read-str (.body response) :key-fn keyword))))
|
||||
|
||||
|
||||
(defn- parse-timestamp
|
||||
[s]
|
||||
(when s
|
||||
(Timestamp/valueOf (.replace ^String s "T" " "))))
|
||||
|
||||
|
||||
(defn- dkk-price
|
||||
[price-dkk price-eur]
|
||||
(or price-dkk
|
||||
(when price-eur (* price-eur eur-dkk-rate))))
|
||||
|
||||
|
||||
(defn- fetch-area
|
||||
"Fetch hourly prices for one price area from elprisenligenu.dk."
|
||||
[^LocalDate date area]
|
||||
(let [url (elprisen-url date area)
|
||||
_ (log/info "Fetching" url)
|
||||
records (fetch-json url)]
|
||||
(when (sequential? records)
|
||||
(map (fn [r]
|
||||
;; time_start is "YYYY-MM-DDTHH:MM:SS+HH:MM" already in DK local;
|
||||
;; drop the 6-char offset suffix, keep the naive wall-clock stamp.
|
||||
(let [ts (subs (:time_start r) 0 19)]
|
||||
{:time-dk ts
|
||||
:area area
|
||||
:dkk (some-> (:DKK_per_kWh r) (* 1000.0)) ; kWh → MWh
|
||||
:eur (some-> (:EUR_per_kWh r) (* 1000.0))}))
|
||||
records))))
|
||||
|
||||
|
||||
(defn- fetch-from-api
|
||||
"Fetch hourly prices for both DK1 and DK2 from elprisenligenu.dk."
|
||||
[^LocalDate date]
|
||||
(concat (fetch-area date "DK1")
|
||||
(fetch-area date "DK2")))
|
||||
|
||||
|
||||
(defn- tariff-dkk-mwh
|
||||
"N1 time-of-use tariff converted from øre/kWh to DKK/MWh (×10)."
|
||||
[^LocalDateTime dt]
|
||||
(* 10.0 (tariffs/tariff-for-hour (.getMonthValue dt) (.getHour dt))))
|
||||
|
||||
|
||||
(defn- save-records!
|
||||
[query-fn records]
|
||||
(doseq [rec records]
|
||||
(when-let [dkk (dkk-price (:dkk rec) (:eur rec))]
|
||||
(let [ts (parse-timestamp (:time-dk rec))
|
||||
tariff (tariff-dkk-mwh (.toLocalDateTime ^Timestamp ts))
|
||||
total (+ (double dkk) tariff)]
|
||||
(query-fn :insert-elprice!
|
||||
{:time-dk ts
|
||||
:price-area (:area rec)
|
||||
:price-dkk dkk
|
||||
:tariff-dkk tariff
|
||||
:total-dkk total})))))
|
||||
|
||||
|
||||
(defn get-or-fetch!
|
||||
"Return hourly el-price rows for the given LocalDate (DK local), fetching
|
||||
from the upstream API and caching on a miss."
|
||||
[query-fn ^LocalDate date]
|
||||
(let [sql-date (java.sql.Date/valueOf date)
|
||||
rows (query-fn :get-elprice-for-date {:date sql-date})]
|
||||
(if (seq rows)
|
||||
rows
|
||||
(let [records (fetch-from-api date)]
|
||||
(when (seq records)
|
||||
(log/info "Saving" (count records) "el-price records")
|
||||
(save-records! query-fn records))
|
||||
(query-fn :get-elprice-for-date {:date sql-date})))))
|
||||
@@ -76,6 +76,16 @@
|
||||
(.replace ^String s "T" " "))))
|
||||
|
||||
|
||||
(def ^:private eur-dkk-rate 7.46)
|
||||
|
||||
|
||||
(defn- dkk-price
|
||||
"Return DKK price, converting from EUR if DKK is nil."
|
||||
[price-dkk price-eur]
|
||||
(or price-dkk
|
||||
(when price-eur (* price-eur eur-dkk-rate))))
|
||||
|
||||
|
||||
(defn save-records!
|
||||
"Insert price records into the database, skipping existing ones."
|
||||
[query-fn records]
|
||||
@@ -84,7 +94,7 @@
|
||||
{:time-utc (parse-timestamp (:TimeUTC rec))
|
||||
:time-dk (parse-timestamp (:TimeDK rec))
|
||||
:price-area (:PriceArea rec)
|
||||
:price-dkk (:DayAheadPriceDKK rec)
|
||||
:price-dkk (dkk-price (:DayAheadPriceDKK rec) (:DayAheadPriceEUR rec))
|
||||
:price-eur (:DayAheadPriceEUR rec)})))
|
||||
|
||||
|
||||
@@ -113,3 +123,13 @@
|
||||
(defn get-tomorrow-prices
|
||||
[query-fn]
|
||||
(get-prices-for-date query-fn (.plusDays (LocalDate/now dk-zone) 1) "DK1"))
|
||||
|
||||
|
||||
(defn ensure-current-prices!
|
||||
"Fetch prices only when today's DK1 prices are missing from the database."
|
||||
[query-fn]
|
||||
(if (seq (get-today-prices query-fn))
|
||||
(do
|
||||
(log/info "Day-ahead prices already exist; skipping API fetch")
|
||||
nil)
|
||||
(fetch-and-save! query-fn)))
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
(ns pmagnus.elprice.web.routes.api
|
||||
(:require
|
||||
[integrant.core :as ig]
|
||||
[pmagnus.elprice.web.controllers.elprice :as elprice]
|
||||
[pmagnus.elprice.web.controllers.health :as health]
|
||||
[pmagnus.elprice.web.middleware.exception :as exception]
|
||||
[pmagnus.elprice.web.middleware.formats :as formats]
|
||||
@@ -8,7 +9,13 @@
|
||||
[reitit.ring.coercion :as coercion]
|
||||
[reitit.ring.middleware.muuntaja :as muuntaja]
|
||||
[reitit.ring.middleware.parameters :as parameters]
|
||||
[reitit.swagger :as swagger]))
|
||||
[reitit.swagger :as swagger]
|
||||
[ring.util.http-response :as http-response])
|
||||
(:import
|
||||
(java.time
|
||||
LocalDate)
|
||||
(java.time.format
|
||||
DateTimeParseException)))
|
||||
|
||||
|
||||
(def route-data
|
||||
@@ -33,9 +40,26 @@
|
||||
exception/wrap-exception]})
|
||||
|
||||
|
||||
(defn- elprice-handler
|
||||
[query-fn {{{:keys [date]} :query} :parameters}]
|
||||
(try
|
||||
(let [d (LocalDate/parse date)
|
||||
rows (elprice/get-or-fetch! query-fn d)]
|
||||
(http-response/ok
|
||||
{:date date
|
||||
:prices (mapv (fn [r] {:time_dk (str (:time_dk r))
|
||||
:price_area (:price_area r)
|
||||
:price_dkk (:price_dkk r)
|
||||
:tariff_dkk (:tariff_dkk r)
|
||||
:total_dkk (:total_dkk r)})
|
||||
rows)}))
|
||||
(catch DateTimeParseException _
|
||||
(http-response/bad-request {:error "date must be YYYY-MM-DD"}))))
|
||||
|
||||
|
||||
;; Routes
|
||||
(defn api-routes
|
||||
[_opts]
|
||||
[{:keys [query-fn]}]
|
||||
[["/swagger.json"
|
||||
{:get {:no-doc true
|
||||
:swagger {:info {:title "pmagnus.elprice API"}}
|
||||
@@ -44,7 +68,11 @@
|
||||
;; note that use of the var is necessary
|
||||
;; for reitit to reload routes without
|
||||
;; restarting the system
|
||||
{:get #'health/healthcheck!}]])
|
||||
{:get #'health/healthcheck!}]
|
||||
["/elprice"
|
||||
{:get {:summary "Hourly DK1 + DK2 electricity prices for a date"
|
||||
:parameters {:query [:map [:date :string]]}
|
||||
:handler (partial elprice-handler query-fn)}}]])
|
||||
|
||||
|
||||
(derive :reitit.routes/api :reitit/routes)
|
||||
|
||||
@@ -184,7 +184,7 @@
|
||||
|
||||
(defn home
|
||||
[query-fn request]
|
||||
(prices/fetch-and-save! query-fn)
|
||||
(prices/ensure-current-prices! query-fn)
|
||||
(page {:lang "en"}
|
||||
[:h1 {:class "text-2xl font-bold text-gray-900 cursor-pointer hover:text-blue-600 transition-colors"
|
||||
:hx-post "/prices/fetch"
|
||||
@@ -211,7 +211,7 @@
|
||||
|
||||
(defn fetch-prices
|
||||
[query-fn request]
|
||||
(prices/fetch-and-save! query-fn)
|
||||
(prices/ensure-current-prices! query-fn)
|
||||
(fragment
|
||||
(price-content :today (prices/get-today-prices query-fn) (has-tomorrow? query-fn))))
|
||||
|
||||
|
||||
Reference in New Issue
Block a user