diff --git a/pipes-postgresql-simple.cabal b/pipes-postgresql-simple.cabal
--- a/pipes-postgresql-simple.cabal
+++ b/pipes-postgresql-simple.cabal
@@ -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       ,
diff --git a/src/Pipes/PostgreSQL/Simple.hs b/src/Pipes/PostgreSQL/Simple.hs
--- a/src/Pipes/PostgreSQL/Simple.hs
+++ b/src/Pipes/PostgreSQL/Simple.hs
@@ -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 #-}
diff --git a/src/Pipes/PostgreSQL/Simple/SafeT.hs b/src/Pipes/PostgreSQL/Simple/SafeT.hs
new file mode 100644
--- /dev/null
+++ b/src/Pipes/PostgreSQL/Simple/SafeT.hs
@@ -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 #-}