streaming-postgresql-simple 0.1.0.0 → 0.2.0.0
raw patch · 3 files changed
+111/−37 lines, 3 filesdep −mtl
Dependencies removed: mtl
Files
- ChangeLog.md +28/−0
- Database/PostgreSQL/Simple/Streaming.hs +82/−35
- streaming-postgresql-simple.cabal +1/−2
ChangeLog.md view
@@ -1,5 +1,33 @@ # Revision history for streaming-postgresql-simple +## 0.2.0.0 -- 2017-02-03++### Correctly perform finalisation in `query` functions.++The previous implementation would perform the necessary finalisation only if the+stream was drained. Some handling was in-place such that exceptions wouldn't+cause the stream to end prematurely, but this isn't enough. We now use +`MonadResource` to register an action to drain the stream.++Users should now wrap calls using `query` with `runResourceT`:++```haskell+>>> runResourceT (S.mapM_ print (query c "SELECT * FROM t"))+```++### Correctly deal with transactions in `stream`++`stream` requires a transaction in order to function. If there isn't a +transaction open, `stream` would create one, but if you manually called `commit`+or `rollback` from within the stream, the internal state would become +inconsistent. This would lead to confusing error messages.++We now watch the transaction state as we pull items out from the stream, and+inform the user if the internal state is not what we expected. Further more,+cleanup actions (commit/rolling back the transaction) now only happen if there+is still a transaction open. ++ ## 0.1.0.0 -- 2017-02-02 * First version. Released on an unsuspecting world. Mwahaha.
Database/PostgreSQL/Simple/Streaming.hs view
@@ -1,6 +1,9 @@+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RankNTypes #-}-{-# LANGUAGE BangPatterns, OverloadedStrings, RecordWildCards,- ScopedTypeVariables #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-} module Database.PostgreSQL.Simple.Streaming ( -- * Queries that return results@@ -26,18 +29,20 @@ -- * Streaming data in and out of PostgreSQL with @COPY@ , copyIn , copyOut++ -- * Re-exported symbols+ , runResourceT ) where import Control.Exception.Safe- (MonadCatch, MonadMask, catch, catchAny, throwM, mask)+ (MonadMask, catch, throwM, mask) import Control.Monad (unless) import Control.Monad.Catch (onException) import Control.Monad.IO.Class (MonadIO, liftIO)-import Control.Monad.State (put) import Control.Monad.Trans.Class (lift) import Control.Monad.Trans.Reader (runReaderT) import Control.Monad.Trans.Resource- (MonadResource, register, unprotect)+ (MonadResource, register, unprotect, runResourceT) import Control.Monad.Trans.State.Strict (runStateT) import qualified Data.ByteString as B import qualified Data.ByteString as B8@@ -64,7 +69,7 @@ (isFailedTransactionError, beginMode, commit) import Database.PostgreSQL.Simple.TypeInfo (getTypeInfo) import Database.PostgreSQL.Simple.Types (Query(..))-import Streaming (Stream, unfold, distribute)+import Streaming (Stream, unfold) import Streaming.Internal (Stream(..)) import Streaming.Prelude (Of(..), reread, for, untilRight) import qualified Streaming.Prelude as S@@ -112,40 +117,41 @@ -} query- :: (ToRow q, FromRow r)+ :: (ToRow q, FromRow r, MonadResource m) => Connection -> Query -> q- -> Stream (Of r) IO ()+ -> Stream (Of r) m () query conn template qs = doQuery fromRow conn template =<< liftIO (formatQuery conn template qs) -- | A version of 'query' that does not perform query substitution. query_- :: (FromRow r)+ :: (FromRow r, MonadResource m) => Connection -> Query- -> Stream (Of r) IO ()+ -> Stream (Of r) m () query_ conn template@(Query que) = doQuery fromRow conn template que -- | A version of 'query' taking parser as argument. queryWith- :: (ToRow q)+ :: (ToRow q, MonadResource m) => RowParser r -> Connection -> Query -> q- -> Stream (Of r) IO ()+ -> Stream (Of r) m () queryWith parser conn template qs = doQuery parser conn template =<< liftIO (formatQuery conn template qs) -- | A version of 'query_' taking parser as argument. queryWith_- :: RowParser r -> Connection -> Query -> Stream (Of r) IO ()+ :: MonadResource m+ => RowParser r -> Connection -> Query -> Stream (Of r) m () queryWith_ parser conn template@(Query que) = doQuery parser conn template que doQuery- :: (MonadIO m, MonadCatch m)+ :: (MonadResource m) => RowParser r -> Connection -> Query -> B.ByteString -> Stream (Of r) m () doQuery parser conn q que = do ok <-@@ -155,25 +161,42 @@ unless ok $ liftIO (withConnection conn LibPQ.errorMessage) >>= traverse_ (liftIO . throwM . flip QueryError q . C8.unpack)- (_, fin) <-- flip runStateT (return ()) $- distribute $- for (results conn) $ \result -> do- status <- liftIO (LibPQ.resultStatus result)- case status of- LibPQ.EmptyQuery -> put $ throwM $ QueryError "query: Empty query" q- LibPQ.CommandOk ->- put $ throwM $ QueryError "query resulted in a command response" q- LibPQ.CopyOut ->- put $ throwM $ QueryError "query: COPY TO is not supported" q- LibPQ.CopyIn ->- put $ throwM $ QueryError "query: COPY FROM is not supported" q- LibPQ.BadResponse -> put $ throwResultError "query" result status- LibPQ.NonfatalError -> put $ throwResultError "query" result status- LibPQ.FatalError -> put $ throwResultError "query" result status- _ -> catchAny (streamResult conn parser result) (put . throwM)- liftIO fin+ yieldResults `onTermination` drainRemainingResults + where++ drainRemainingResults :: MonadIO m => m ()+ drainRemainingResults =+ S.effects (results conn)++ yieldResults =+ for (results conn) $ \result -> do+ status <- liftIO (LibPQ.resultStatus result)+ case status of+ LibPQ.EmptyQuery ->+ liftIO $ throwM $ QueryError "query: Empty query" q++ LibPQ.CommandOk ->+ liftIO $ throwM $ QueryError "query resulted in a command response" q++ LibPQ.CopyOut ->+ liftIO $ throwM $ QueryError "query: COPY TO is not supported" q++ LibPQ.CopyIn ->+ liftIO $ throwM $ QueryError "query: COPY FROM is not supported" q++ LibPQ.BadResponse ->+ liftIO (throwResultError "query" result status)++ LibPQ.NonfatalError ->+ liftIO (throwResultError "query" result status)++ LibPQ.FatalError ->+ liftIO (throwResultError "query" result status)++ _ ->+ streamResult conn parser result+ results :: MonadIO m => Connection -> Stream (Of LibPQ.Result) m () results = reread (liftIO . flip withConnection LibPQ.getResult) @@ -205,6 +228,16 @@ -- transaction if needed. The cursor is given a unique temporary name, -- so the consumer may itself call 'stream'. --+-- Due to the dependency on transactions, you must ensure that `commit`+-- or `rollback` aren't called on the connection used to form a stream.+-- Doing so causes the stream cursor to be released, making it impossible+-- to stream more results. If you do perform a commit or rollback,+-- 'stream' will raise an exception indicating that a transaction+-- was aborted.+--+-- If you performing transaction writes in a stream, consider instead+-- using save points, which will nest correctly with 'stream'.+-- -- Exceptions that may be thrown: -- -- * 'Database.PostgreSQL.Simple.FormatError': the query string could not be formatted correctly.@@ -288,8 +321,8 @@ case stat of LibPQ.TransIdle -> bracket (liftIO (beginMode transactionMode conn))- (\_ -> liftIO (commit conn))- (\_ -> go `onException` liftIO (rollback conn))+ (\_ -> ifInTransaction $ liftIO (commit conn))+ (\_ -> go `onException` ifInTransaction (liftIO (rollback conn))) LibPQ.TransInTrans -> go LibPQ.TransActive -> fail "foldWithOpts FIXME: PQ.TransActive" -- This _shouldn't_ occur in the current incarnation of@@ -304,12 +337,20 @@ LibPQ.TransUnknown -> fail "foldWithOpts FIXME: PQ.TransUnknown" -- Not sure what this means. where+ ifInTransaction m = do+ stat <- liftIO (withConnection conn LibPQ.transactionStatus)+ case stat of+ LibPQ.TransInTrans -> m+ _ -> return ()+ declare = do name <- newTempName conn _ <- execute_ conn $ mconcat [ "DECLARE ", name, " NO SCROLL CURSOR FOR ", q ] return name- close name = do++ close name =+ ifInTransaction $ (execute_ conn ("CLOSE " <> name) >> return ()) `catch` \ex -> -- Don't throw exception if CLOSE failed because the transaction is -- aborted. Otherwise, it will throw away the original error.@@ -324,6 +365,12 @@ byteString " FROM " <> byteString name) fetches = untilRight $ do++ stat <- liftIO (withConnection conn LibPQ.transactionStatus)+ case stat of+ LibPQ.TransInTrans -> return ()+ _ -> fail "Stream transaction prematurely aborted"+ result <- liftIO (exec conn fetchQ) status <- liftIO (LibPQ.resultStatus result) case status of
streaming-postgresql-simple.cabal view
@@ -1,5 +1,5 @@ name: streaming-postgresql-simple-version: 0.1.0.0+version: 0.2.0.0 synopsis: Stream postgresql-query results using the streaming library description: This package provides incremental streaming functions for the @postgresql-simple@ library. license: BSD3@@ -21,7 +21,6 @@ build-depends: base >=4.9 && <4.10 , bytestring >= 0.10.8.1 && < 0.11 , exceptions >= 0.8.3 && < 0.9- , mtl >= 2.2.1 && < 2.3 , postgresql-libpq >= 0.9.2.0 && < 0.10 , postgresql-simple >=0.5 && <0.6 , resourcet >= 1.1.8.1 && < 1.2