pipes-postgresql-simple (empty) → 0.1.0.0
raw patch · 4 files changed
+176/−0 lines, 4 filesdep +asyncdep +basedep +bytestringsetup-changed
Dependencies added: async, base, bytestring, mtl, pipes, pipes-concurrency, postgresql-simple, stm, text, transformers
Files
- LICENSE +30/−0
- Setup.hs +2/−0
- pipes-postgresql-simple.cabal +33/−0
- src/Pipes/PostgreSQL/Simple.hs +111/−0
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c) 2013, Oliver Charles++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 Oliver Charles 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.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ pipes-postgresql-simple.cabal view
@@ -0,0 +1,33 @@+name: pipes-postgresql-simple+version: 0.1.0.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,+ there is support for:+ .+ * Streaming queries+ .+ * Streaming entire tables to and from a @ByteString@.+license: MIT+license-file: LICENSE+author: Oliver Charles+maintainer: ollie@ocharles.org.uk+copyright: Oliver Charles 2013+category: Database, Pipes+build-type: Simple+cabal-version: >=1.8++library+ exposed-modules: Pipes.PostgreSQL.Simple+ build-depends: base >=4.5 && < 5,+ async >=2.0 ,+ mtl >=2.1 ,+ bytestring >=0.9 ,+ text >=0.11 ,+ pipes >=3.3 ,+ postgresql-simple >=0.3.4 ,+ pipes-concurrency >=1.1 ,+ transformers >=0.3 ,+ stm >=2.4+ hs-source-dirs: src+ GHC-Options: -O2 -Wall
+ src/Pipes/PostgreSQL/Simple.hs view
@@ -0,0 +1,111 @@+-- | Pipes utilities built on top of @postgresql-simple@++module Pipes.PostgreSQL.Simple (+ -- * Querying+ query,++ -- * Serialization and Deserialization+ Format(..),+ fromTable,+ toTable+ ) where++import Data.String (fromString)+import Control.Monad (void)+import Control.Monad.IO.Class (MonadIO, liftIO)+import Data.ByteString (ByteString)+import Data.Int (Int64)++import qualified Control.Concurrent.Async as Async+import qualified Control.Concurrent.STM as STM+import qualified Database.PostgreSQL.Simple as Pg+import qualified Database.PostgreSQL.Simple.Copy as Pg+import qualified Pipes+import qualified Pipes.Concurrent as Pipes++--------------------------------------------------------------------------------+-- | Convert a query to a 'Producer' of rows.+--+-- For example,+--+-- > pg <- connectToPostgresql+-- > query pg "SELECT * FROM widgets WHERE ID = ?" (Only widgetId) >-> print+--+-- Will select all widgets for a given @widgetId@, and then print each row to+-- standard output.+query+ :: (MonadIO m, Pg.FromRow r, Pg.ToRow params)+ => Pg.Connection -> Pg.Query -> params -> Pipes.Producer r m ()+query c q p = do+ (o, i, seal) <- liftIO (Pipes.spawn' Pipes.Single)+ worker <- liftIO $ Async.async $ do+ Pg.fold c q p () (const $ void . STM.atomically . Pipes.send o)+ STM.atomically seal+ 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.+--+-- Returns the number of rows processed.+fromTable+ :: MonadIO m+ => Pg.Connection+ -> Format+ -> String+ -> Pipes.Producer ByteString m Int64+fromTable c fmt tblName = do+ liftIO $ Pg.copy_ c $ fromString $ concat+ [ "COPY "+ , tblName+ , " TO STDOUT WITH (FORMAT \""+ , showFmt fmt+ , "\")"+ ]+ let go = do+ r <- liftIO (Pg.getCopyData c)+ case r of+ Pg.CopyOutRow bs -> do+ Pipes.yield bs+ go+ Pg.CopyOutDone n -> return n+ go+{-# INLINABLE fromTable #-}++--------------------------------------------------------------------------------+-- | Convert a byte stream to a table. This is equivilent to a PostgreSQL+-- @COPY ... FROM@ statement.+--+-- Returns the number of rows processed+toTable+ :: MonadIO m+ => Pg.Connection+ -> Format+ -> String+ -> Pipes.Producer ByteString m ()+ -> m Int64+toTable c fmt tblName p0 = do+ liftIO $ Pg.copy_ c $ fromString $ concat+ [ "COPY "+ , tblName+ , " FROM STDIN WITH (FORMAT \""+ , showFmt fmt+ , "\")"+ ]+ let go p = do+ x <- Pipes.next p+ case x of+ Left () -> liftIO (Pg.putCopyEnd c)+ Right (bs, p') -> liftIO (Pg.putCopyData c bs) >> go p'+ go p0+{-# INLINABLE toTable #-}