This library has 2 high level functions

• cursorPipe to stream a query from postgres DB
• connect to handle connection in 'SafeT IO'

Here is how usage looks like from a production excerpt for a table of samples

mkDB :: ByteString
mkDB =
  [qmb|
    create table samples (
      id         uuid not null,
      time       bigint not null,
      value      double precision not null,
    );
 |]

streamSamples
  :: RunSession
  -> Int64 -- ^ first sample time
  -> Producer (UUID, Int64, Double) (SafeT IO) ()
streamSamples (RunSession run) from' =
  yield from'
    >-> cursorPipe
      do run
      do samplesEncoder
      do samplesDecoder
      do Cursor "sample_cursor"
      do Template [qmb| select * from samples where time >= $1 order by time; |]
      do 1000

samplesEncoder :: Params Int64
samplesEncoder = param $ E.nonNullable E.int8

samplesDecoder :: Row (UUID, Int64, Double)
samplesDecoder =
  (,,)
    $ do column . D.nonNullable $ D.uuid
    * do column . D.nonNullable $ D.int8
    * do column . D.nonNullable $ D.float8

localConnect :: Trace IO DatabaseLog -> SafeT IO RunSession
localConnect tracer = do
  connect tracer $
    settings
      "10.1.9.95"
      5432
      "postgres"
      "postgres"
      "postgres"

main :: IO ()
main = do
  ts <- getPOSIXTime
  runSafeT $ do
    run' <- localConnect pPrint
    runEffect $ streamSamples run' (floor ts - 600) >-> P.print