pipes-postgresql-simple 0.1.0.0 → 0.1.1.0

raw patch · 3 files changed

+72/−23 lines, 3 filesdep +exceptionsdep +pipes-safe

Dependencies added: exceptions, pipes-safe

Files

• pipes-postgresql-simple.cabal +4/−1
• src/Pipes/PostgreSQL/Simple.hs +20/−22
• src/Pipes/PostgreSQL/Simple/SafeT.hs +48/−0

@@ -1,5 +1,5 @@ name: pipes-postgresql-simple-version: 0.1.0.0+version: 0.1.1.0 synopsis: Convert various postgresql-simple calls to work with pipes description: This library provides a few Producers and Consumers that allow   @postgresql-simple@ calls to be made within the @pipes@ framework. Currently,@@ -19,12 +19,15 @@  library   exposed-modules:     Pipes.PostgreSQL.Simple+                       Pipes.PostgreSQL.Simple.SafeT   build-depends:       base              >=4.5 && < 5,                        async             >=2.0       ,+                       exceptions,                        mtl               >=2.1       ,                        bytestring        >=0.9       ,                        text              >=0.11      ,                        pipes             >=3.3       ,+                       pipes-safe                    ,                        postgresql-simple >=0.3.4     ,                        pipes-concurrency >=1.1       ,                        transformers      >=0.3       ,

@@ -10,14 +10,19 @@     toTable     ) where -import Data.String (fromString) import Control.Monad (void)+import Control.Monad.Catch (MonadCatch, catchAll, throwM) import Control.Monad.IO.Class (MonadIO, liftIO) import Data.ByteString (ByteString) import Data.Int (Int64)+import Data.String (fromString) +import Pipes.PostgreSQL.Simple.SafeT (Format(..))+ import qualified Control.Concurrent.Async as Async import qualified Control.Concurrent.STM as STM+import qualified Data.Text as Text+import qualified Data.Text.Encoding as Text import qualified Database.PostgreSQL.Simple as Pg import qualified Database.PostgreSQL.Simple.Copy as Pg import qualified Pipes@@ -44,15 +49,6 @@     liftIO $ Async.link worker     Pipes.fromInput i --- | The PostgreSQL file format, used by the @COPY@ command-data Format = Text | Binary | CSV--showFmt :: Format -> String-showFmt fmt = case fmt of-    Text   -> "text"-    Binary -> "binary"-    CSV    -> "csv"- -------------------------------------------------------------------------------- -- | Convert a table to a byte stream. This is equivilent to a PostgreSQL -- @COPY ... TO@ statement.@@ -66,11 +62,8 @@     -> Pipes.Producer ByteString m Int64 fromTable c fmt tblName = do     liftIO $ Pg.copy_ c $ fromString $ concat-        [ "COPY "-        , tblName-        , " TO STDOUT WITH (FORMAT \""-        , showFmt fmt-        , "\")"+        [ "COPY ", tblName+        , " TO STDOUT WITH (FORMAT ", show fmt , ")"         ]     let go = do             r <- liftIO (Pg.getCopyData c)@@ -88,7 +81,7 @@ -- -- Returns the number of rows processed toTable-    :: MonadIO m+    :: (MonadCatch m, MonadIO m)     => Pg.Connection     -> Format     -> String@@ -96,16 +89,21 @@     -> m Int64 toTable c fmt tblName p0 = do     liftIO $ Pg.copy_ c $ fromString $ concat-        [ "COPY "-        , tblName-        , " FROM STDIN WITH (FORMAT \""-        , showFmt fmt-        , "\")"+        [ "COPY " , tblName+        , " FROM STDIN WITH (FORMAT " , show fmt, "\")"         ]     let go p = do-            x <- Pipes.next p+            x <- Pipes.next p `onException`+                   (liftIO . Pg.putCopyError c .+                      Text.encodeUtf8 . Text.pack . show)             case x of                 Left   ()      -> liftIO (Pg.putCopyEnd c)                 Right (bs, p') -> liftIO (Pg.putCopyData c bs) >> go p'     go p0++ where++  action `onException` handler =++    action `catchAll` \e -> handler e >> throwM e {-# INLINABLE toTable #-}

@@ -0,0 +1,48 @@+{-# LANGUAGE TypeFamilies #-}+module Pipes.PostgreSQL.Simple.SafeT (Format(..), toTable) where++import Control.Monad (void)+import Control.Monad.Catch (catchAll, throwM)+import Control.Monad.IO.Class (MonadIO, liftIO)+import Data.ByteString (ByteString)+import Data.String (fromString)++import qualified Database.PostgreSQL.Simple as Pg+import qualified Database.PostgreSQL.Simple.Copy as Pg+import qualified Data.Text as Text+import qualified Data.Text.Encoding as Text+import qualified Pipes+import qualified Pipes.Safe as Pipes++--------------------------------------------------------------------------------+-- | The PostgreSQL file format, used by the @COPY@ command+data Format = Text | Binary | CSV+  deriving (Show)+++--------------------------------------------------------------------------------+toTable+    :: (MonadIO m, Pipes.MonadSafe m, Pipes.Base m ~ IO)+    => Pg.Connection+    -> Format+    -> String+    -> Pipes.Consumer ByteString m a+toTable c fmt tblName = do+  putCopyEnd <- Pipes.register (void $ Pg.putCopyEnd c)++  Pipes.liftBase $ Pg.copy_ c $ fromString $ concat+    [ "COPY ", tblName+    , " FROM STDIN WITH (FORMAT " , show fmt, ")"+    ]++  Pipes.for Pipes.cat (liftIO . Pg.putCopyData c)+    `onException` (\e -> do Pipes.release putCopyEnd+                            Pipes.liftBase $ Pg.putCopyError c $+                              Text.encodeUtf8 . Text.pack $ show e)++ where++  action `onException` handler =+    action `catchAll` \e -> handler e >> throwM e++{-# INLINABLE toTable #-}