lib.postgres

PostgreSQL runtime and connection lifecycle

Coordinate PostgreSQL connection startup, readiness, temporary database setup, and shutdown.

1 Overview

The runtime functions operate on a map containing connection settings, an instance atom, and notification state. Optional settings add setup and teardown hooks.

2 Walkthrough

2.1 Start a runtime

(require '[lib.postgres :as postgres])

(def runtime
  {:host "localhost"
   :port 5432
   :dbname "application"
   :instance (atom nil)
   :notifications (atom {})})

(postgres/start-pg runtime)

2.2 Inspect and close

@(:instance runtime)
(postgres/stop-pg runtime)

2.3 Lifecycle options

Temporary database and external runtime options can be added to the same map. The start function waits for availability before creating the connection instance.

run-pg-lifecycle ^

[rt command]

Added 4.1

executes lifecycle commands using std.lib.os/sh arguments

v 4.1

(defn run-pg-lifecycle
  [rt command]
  (cond
    (fn? command)
    (do (try
          (command rt)
          (catch clojure.lang.ArityException _
            (command)))
        rt)

    command
    (do (os/sh command)
        rt)

    :else
    rt))

link
(let [calls (atom [])]
  (with-redefs [os/sh (fn [args]
                        (swap! calls conj args)
                        "ok")]
    (base/run-pg-lifecycle {:dbname "test"} {:args ["supabase" "start"]
                                            :root "docker/supbase"}))
  @calls)
=> [{:args ["supabase" "start"]
     :root "docker/supbase"}]

run-pg-lifecycle-callback ^

Added 4.1

executes lifecycle callbacks when given a function

v 4.1

link
(let [calls (atom [])]
  (with-redefs [os/sh (fn [args]
                        (swap! calls conj args)
                        "ok")]
    (base/run-pg-lifecycle {:dbname "test"}
                           (fn [rt]
                             (swap! calls conj [:fn rt])
                             rt)))
  @calls)
=> [[:fn {:dbname "test"}]]

start-pg ^

[pg]

Added 4.1

runs startup shell hooks before opening the postgres connection

v 4.1

(def ^{:arglists '([pg])} start-pg
  (component/wrap-start
   start-pg-raw
   [{:key :startup
     :setup run-pg-lifecycle}
    {:key :container
     :setup docker/start-runtime
     :teardown docker/stop-runtime}
    {:key :shutdown
     :teardown run-pg-lifecycle}]))

link
(let [calls (atom [])]
  (with-redefs [os/sh (fn [args]
                        (swap! calls conj [:sh args])
                        "ok")
                conn/conn-create (fn [_]
                                   (swap! calls conj [:conn])
                                   (mock-pooled-conn))]
    (base/start-pg {:instance (atom nil)
                    :notifications (atom {})
                    :startup {:args ["supabase" "start"]
                              :root "docker/supbase"}}))
  @calls)
=> [[:sh {:args ["supabase" "start"]
          :root "docker/supbase"}]
    [:conn]]

start-pg-raw ^

[{:keys [instance host port container temp], :as pg}]

Added 4.0

starts the database

v 4.0

(defn start-pg-raw
  ([{:keys [instance host port container temp] :as pg}]
   (when container
     (network/wait-for-port host port {:timeout 5000})
     (wait-for-pg pg 10 1000))
   (when temp
     (start-pg-temp-init pg))
   (when-not @instance
     (reset! instance (conn/conn-create pg)))
   pg))

link
(with-redefs [conn/conn-create (constantly (mock-pooled-conn))]
  (base/start-pg-raw {:instance (atom nil) :temp false}))
=> map?

start-pg-temp-init ^

[{:keys [dbname], :as pg}]

Added 4.0

initialises a temp database

v 4.0

(defn start-pg-temp-init
  ([{:keys [dbname] :as pg}]
   (when-not dbname (f/error "Missing dbname"))
   (with-open [conn (conn/conn-create (assoc pg :dbname "postgres"))]
     (with-open [c (.getConnection conn)]
       (let [args (jdbc/fetch c [(format "SELECT 1 FROM pg_database WHERE datname = '%s'" dbname)])]
         (when (empty? args)
           (jdbc/execute c [(format "CREATE DATABASE "%s"" dbname)])))))))

link
(with-redefs [conn/conn-create (constantly (mock-pooled-conn))
              jdbc/fetch (constantly [])
              jdbc/execute (constantly 1)]
  (base/start-pg-temp-init {:dbname "test"}))
=> 1

stop-pg ^

[pg]

Added 4.1

runs shutdown shell hooks after closing the postgres connection

v 4.1

(def ^{:arglists '([pg])}
  stop-pg
  (component/wrap-stop
   stop-pg-raw
   [{:key :startup
     :setup run-pg-lifecycle}
    {:key :container
     :setup docker/start-runtime
     :teardown docker/stop-runtime}
    {:key :shutdown
     :teardown run-pg-lifecycle}]))

link
(let [calls (atom [])]
  (with-redefs [conn/conn-close (fn [_]
                                  (swap! calls conj [:conn-close])
                                  nil)
                os/sh (fn [args]
                        (swap! calls conj [:sh args])
                        "ok")]
    (base/stop-pg {:instance (atom (mock-pooled-conn))
                   :notifications (atom {})
                  :shutdown {:args ["supabase" "stop"]
                              :root "docker/supbase"
                              :ignore-errors true}}))
  @calls)
=> [[:conn-close]
    [:sh {:args ["supabase" "stop"]
          :root "docker/supbase"
          :ignore-errors true}]]

stop-pg-raw ^

[{:keys [instance notifications temp], :as pg}]

Added 4.0

stops the postgres runtime

v 4.0

(defn stop-pg-raw
  ([{:keys [instance notifications temp] :as pg}]
   (if @instance
     (atom/swap-return! instance (fn [conn]
                                [(doto conn (conn/conn-close)) nil])))
   (collection/map-vals (fn [{:keys [raw]}]
                 (env/close raw))
               @notifications)
   (reset! notifications {})
   (when (and temp (not= temp :create))
     (stop-pg-temp-teardown pg))
   pg))

link
(with-redefs [conn/conn-close (constantly nil)]
  (base/stop-pg-raw {:instance (atom (mock-pooled-conn)) :notifications (atom {})}))
=> map?

stop-pg-temp-teardown ^

[{:keys [dbname], :as pg}]

Added 4.0

tears down a temp database

v 4.0

(defn stop-pg-temp-teardown
  ([{:keys [dbname] :as pg}]
   (when-not dbname (f/error "Missing dbname"))
   (with-open [conn (conn/conn-create (assoc pg :dbname "postgres"))]
     (with-open [c (.getConnection conn)]
       (let [args (jdbc/fetch c [(format "SELECT 1 FROM pg_database WHERE datname = '%s'" dbname)])]
         (when (not (empty? args))
           (jdbc/execute c [(format "DROP DATABASE "%s"" dbname)])))))))

link
(with-redefs [conn/conn-create (constantly (mock-pooled-conn))
              jdbc/fetch (constantly [{:a 1}])
              jdbc/execute (constantly 1)]
  (base/stop-pg-temp-teardown {:dbname "test-temp-db"}))
=> 1

wait-for-pg ^

[{:keys [instance host port container temp], :as pg} limit sleep]

Added 4.0

waits for the postgres database to come online

v 4.0

(defn wait-for-pg
  [{:keys [instance host port container temp] :as pg} limit sleep]
  (let [_   (if (zero? limit) (f/error "Database not responsive"))
        res (try (with-open [conn (conn/conn-create (assoc pg :dbname "postgres"))])
                 (catch com.impossibl.postgres.jdbc.PGSQLSimpleException e
                   (Thread/sleep ^Long sleep)
                   :error))]
    (if (= res :error)
      (recur pg (dec limit) sleep))))

link
(with-redefs [conn/conn-create (constantly (mock-pooled-conn))]
  (base/wait-for-pg {} 1 1))
=> nil

*execute*
+impls+
conn-close
conn-create
conn-execute
load-impl
notify-create
notify-listener

execute ^

NONE

(def ^:dynamic *execute* nil)

link

+impls+ ^

NONE

(def +impls+
  (atom {:impossibl  {:status :pending
                      :ns 'lib.postgres.impl.impossibl}
         :postgresql {:status :pending
                      :ns 'lib.postgres.impl.postgresql}}))

link

conn-close ^

[conn]

Added 4.0

closes a connection

v 4.0

(defn conn-close
  [conn]
  (cond (instance? javax.sql.PooledConnection conn)
        (.close ^javax.sql.PooledConnection conn)

        (instance? java.lang.AutoCloseable conn)
        (.close ^java.lang.AutoCloseable conn)

        (nil? conn)
        conn
        
        :else
        (f/error "Not closeable." {:type (type conn)
                                   :input conn})))

link
(conn/conn-close (mock-pooled-conn)) => nil

conn-create ^

[{:keys [vendor], :or {vendor :impossibl}, :as m}]

Added 4.0

creates a pooled connection

v 4.0

(defn ^PooledConnection conn-create
  ([{:keys [vendor]
     :or {vendor :impossibl}
     :as m}]
   (let [{:keys [ns]} (load-impl vendor)
         create-pool (ns-resolve ns 'create-pool)]
     (create-pool m))))

link
(try (conn/conn-create {:dbname "test"})
     (catch Throwable t t))
=> (any java.sql.SQLException
        com.impossibl.postgres.jdbc.PGPooledConnection
        com.impossibl.postgres.jdbc.PGSQLSimpleException)

(try (conn/conn-create {:dbname "test" :vendor :postgresql})
     (catch Throwable t t))
=> (any java.sql.SQLException
        org.postgresql.ds.PGPooledConnection
        org.postgresql.util.PSQLException)

conn-execute ^

[pool input] [pool input execute]

Added 4.0

executes a command

v 4.0

(defn conn-execute
  ([^PooledConnection pool input]
   (conn-execute pool input (or *execute*
                                jdbc/execute)))
  ([^PooledConnection pool input execute]
   (let [vendor (if (let [cls-name (.getName (class pool))]
                      (or (.startsWith cls-name "org.postgresql")
                          (.startsWith cls-name "lib.postgres.impl.postgresql")))
                  :postgresql
                  :impossibl)
         {:keys [ns]} (load-impl vendor)
         execute-statement (ns-resolve ns 'execute-statement)]
     (execute-statement pool input execute))))

link
(with-redefs [conn/conn-create (constantly (mock-pooled-conn))]
  (let [pool (conn/conn-create {:dbname "test"})]
    (conn/conn-execute pool "select 1;" (constantly [{:?column? 1}]))))
=> [{:?column? 1}]

load-impl ^

[vendor]

Added 4.1

loads a postgres implementation

v 4.1

(defn load-impl [vendor]
  (let [entry (get @+impls+ vendor)]
    (if (= (:status entry) :loaded)
      entry
      (try
        (require (:ns entry))
        (swap! +impls+ assoc-in [vendor :status] :loaded)
        (get @+impls+ vendor)
        (catch Throwable t
          (swap! +impls+ assoc-in [vendor :status] :error)
          (f/error "Implementation not found" {:vendor vendor}))))))

link
(let [original @conn/+impls+]
  (try
    (reset! conn/+impls+ {:stub {:status :pending :ns 'clojure.core}})
    (with-redefs [require (fn [& _] nil)]
      (conn/load-impl :stub))
    (finally
      (reset! conn/+impls+ original))))
=> (contains {:status :loaded})

notify-create ^

[{:keys [vendor], :or {vendor :impossibl}, :as m} config]

Added 4.0

creates a notify channel

v 4.0

(defn notify-create
  ([{:keys [vendor] :or {vendor :impossibl} :as m} config]
   (if (not= vendor :impossibl)
     (f/error "Only impossibl driver supports notifications" {:vendor vendor})
     (let [{:keys [ns]} (load-impl :impossibl)
           create-notify (ns-resolve ns 'create-notify)]
       (create-notify m config)))))

link
(try (conn/notify-create {:dbname "test"} {:channel "ch"})
     (catch Throwable t t))
=> (contains-in
    [com.impossibl.postgres.jdbc.PGDirectConnection])

notify-listener ^

[m]

Added 4.0

creates a notification listener

v 4.0

(defn notify-listener
  [m]
  (let [{:keys [ns]} (load-impl :impossibl)
        notify-listener (ns-resolve ns 'notify-listener)]
    (notify-listener m)))

link
(conn/notify-listener {})
=> (partial instance? com.impossibl.postgres.api.jdbc.PGNotificationListener)

lib.postgres

1 Overview

2 Walkthrough

2.1 Start a runtime

2.2 Inspect and close

2.3 Lifecycle options

3 API

run-pg-lifecycle ^

run-pg-lifecycle-callback ^

start-pg ^

start-pg-raw ^

start-pg-temp-init ^

stop-pg ^

stop-pg-raw ^

stop-pg-temp-teardown ^

wait-for-pg ^

*execute* ^

NONE

+impls+ ^

NONE

conn-close ^

conn-create ^

conn-execute ^

load-impl ^

notify-create ^

notify-listener ^

Related

execute ^