hasql-pipes (empty) → 0.1.0.0

raw patch · 6 files changed

+279/−0 lines, 6 filesdep +basedep +bytestring-tree-builderdep +hasqlsetup-changed

Dependencies added: base, bytestring-tree-builder, hasql, pipes, pipes-safe, protolude

Files

ChangeLog.md +3/−0
LICENSE +30/−0
README.md +1/−0
Setup.hs +2/−0
hasql-pipes.cabal +44/−0
src/Hasql/Pipes.hs +199/−0

+ ChangeLog.md view

@@ -0,0 +1,3 @@+# Changelog for hasql-pipes++## Unreleased changes

+ LICENSE view

@@ -0,0 +1,30 @@+Copyright Author name here (c) 2020++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++    * Redistributions of source code must retain the above copyright+      notice, this list of conditions and the following disclaimer.++    * Redistributions in binary form must reproduce the above+      copyright notice, this list of conditions and the following+      disclaimer in the documentation and/or other materials provided+      with the distribution.++    * Neither the name of Author name here nor the names of other+      contributors may be used to endorse or promote products derived+      from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

+ README.md view

@@ -0,0 +1,1 @@+# hasql-pipes

+ Setup.hs view

@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain

+ hasql-pipes.cabal view

@@ -0,0 +1,44 @@+cabal-version: 1.12++-- This file has been generated from package.yaml by hpack version 0.33.0.+--+-- see: https://github.com/sol/hpack+--+-- hash: 2fc484dece0baf332928773545e2531ae503253a571ed39a54fea894455adb7d++name:           hasql-pipes+version:        0.1.0.0+synopsis:       A pipe to stream a postgres database cursor in the hasql ecosystem+description:    Please see the README on GitLab at <https://gitlab.com/paolo.veronelli/hasql-pipes>+category:       Database, Streaming+homepage:       https://github.com/https://gitlab.com/paolo.veronelli/hasql-pipes#readme+bug-reports:    https://github.com/https://gitlab.com/paolo.veronelli/hasql-pipes/issues+author:         Paolo Veronelli+maintainer:     paolo.veronelli@gmail.com+copyright:      2020 Paolo Veronelli+license:        BSD3+license-file:   LICENSE+build-type:     Simple+extra-source-files:+    README.md+    ChangeLog.md++source-repository head+  type: git+  location: https://github.com/https://gitlab.com/paolo.veronelli/hasql-pipes++library+  exposed-modules:+      Hasql.Pipes+  other-modules:+      Paths_hasql_pipes+  hs-source-dirs:+      src+  build-depends:+      base >=4.7 && <5+    , bytestring-tree-builder+    , hasql+    , pipes+    , pipes-safe+    , protolude+  default-language: Haskell2010

+ src/Hasql/Pipes.hs view

@@ -0,0 +1,199 @@+{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes #-}++-- | 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+--+-- +-- @+++module Hasql.Pipes where++import qualified ByteString.TreeBuilder as TB+import Hasql.Connection (acquire, ConnectionError, Settings)+import Hasql.Decoders (Row, rowList)+import qualified Hasql.Decoders as D+import Hasql.Encoders (Params)+import qualified Hasql.Encoders as E+import Hasql.Session+import Hasql.Statement+import Pipes+import qualified Pipes as P+import Pipes.Safe+import qualified Pipes.Safe as P+import Protolude+import qualified Hasql.Connection as HC++-- | cursor name+newtype Cursor = Cursor ByteString deriving (IsString)++-- | query to run+newtype Template = Template ByteString deriving (IsString)++-- | a statement to declare a cursor parametrized over some parameters+declareCursor+  :: E.Params a -- ^ paramenters encoding+  -> Cursor -- ^ cursor name+  -> Template -- ^ query template+  -> Statement a ()+declareCursor encoder (Cursor name) (Template template) =+  Statement sql' encoder D.noResult False+  where+    sql' =+      TB.toByteString $+        "DECLARE "+          <> TB.byteString name+          <> " NO SCROLL CURSOR FOR "+          <> TB.byteString template++-- | a statement to close the cursor+closeCursor+  :: Cursor -- ^ cursor name+  -> Statement () ()+closeCursor (Cursor name) =+  Statement sql' E.noParams D.noResult True+  where+    sql' = "CLOSE " <> name++-- | a statement to fetch given number of rows from cursor forward and apply decoders+fetchFromCursor+  :: Cursor -- ^ cursor name+  -> Batch -- ^ max number of rows to fetch+  -> D.Result result -- ^ row decoders+  -> Statement () result+fetchFromCursor (Cursor name) (Batch batch) decoder =+  Statement sql' E.noParams decoder True+  where+    sql' =+      TB.toByteString $+        "FETCH FORWARD "+          <> TB.asciiIntegral batch+          <> " FROM "+          <> TB.byteString name++beginTransaction :: Session ()+beginTransaction = sql "BEGIN TRANSACTION"++endTransaction :: Session ()+endTransaction = sql "END TRANSACTION"++-- | number of rows+newtype Batch = Batch Int deriving (Num)++-- | stream rows for queries of the same template+cursorPipe+  :: (forall b. Session b -> IO b) -- ^ execute a session command+  -> Params z -- ^ query parameters encoders+  -> Row a -- ^ row decoders+  -> Cursor -- ^ desidered cursor name+  -> Template -- ^ query template+  -> Batch -- ^ number of rows to repeat fetching+  -> Pipe z a (SafeT IO) ()+cursorPipe runS encoders decoders cursor template batch = forever $ do+  parameters <- await+  void $+    liftIO $ runS do+      beginTransaction+      statement parameters $ declareCursor encoders cursor template+  transaction <- lift $+    register $ runS do+      statement () $ closeCursor cursor+      endTransaction+  let loop = do+        xs <-+          liftIO $+            runS $+              statement () $+                fetchFromCursor cursor batch $+                  rowList decoders+        case xs of+          [] -> pure ()+          _ -> P.each xs >> loop+  loop+  lift $ P.release transaction++data DatabaseLog+  = ConnectionReady+  | ConnectionClosed+  | ConnectionFailed ConnectionError+  | QueryFailed QueryError+  deriving (Show)++newtype RunSession = RunSession (forall a. Session a -> IO a)++connect :: (DatabaseLog -> IO ()) -> Settings -> SafeT IO RunSession+connect tracer settings' = do+  mconnection <- liftIO (acquire settings')+  connection <- case mconnection of +    Left e -> do +      liftIO $ tracer $ ConnectionFailed e+      panic "no connection"+    Right c -> pure c+  liftIO $ tracer ConnectionReady+  void $+    register $ do+      liftIO $ HC.release connection+      tracer ConnectionClosed+  pure $ RunSession \s -> do+    result <- run s connection+    case result of+      Left e -> tracer (QueryFailed e) >> panic "query failed"+      Right r -> pure r