cassava-conduit 0.2.2 → 0.3.0
raw patch · 3 files changed
+80/−21 lines, 3 filesdep +text
Dependencies added: text
Files
- CHANGELOG.md +41/−0
- cassava-conduit.cabal +2/−1
- src/Data/Csv/Conduit.hs +37/−20
CHANGELOG.md view
@@ -1,3 +1,44 @@+# 0.3.x++## 0.2.2 -> 0.3.0++Some new error types, and error contain `T.Text` and not `String` now++``` Haskell+data CsvParseError =+ CsvParseError BS.ByteString T.Text+ | IncrementalError T.Text+ deriving (Show, Eq)++-- | When you want to include errors in the stream, this error type represents errors that halt the stream.+-- They do not appear inside the conduit and will instead get returned from running the conduit.+--+data CsvStreamHaltParseError = HaltingCsvParseError BS.ByteString T.Text -- ^ the remaining bytestring that was read in but not parsed yet, and the stringy error msg describing the fail.+ deriving (Show, Eq)++-- | When you want to include errors in the stream, these are the errors that can be included in the stream,+-- they are usually problems restricted to individual records, and streaming can resume from the next record+-- you just have to decide on something sensible to do with the per record errors.+--+data CsvStreamRecordParseError = CsvStreamRecordParseError T.Text deriving (Show, Eq) -- ^ The stringy error describing why this record could not be parsed.+```++New error types are to separate out errors that stop streaming (and hence imply there are valid records that might be omitted) and errors that can be skipped, allowing valid records after to be processed...++``` Haskell+-- |+-- Same as `fromCsv` but allows for errors to be handled in the pipeline instead+--+fromCsvStreamError :: (FromRecord a, MonadError e m) => DecodeOptions -> HasHeader -> (CsvStreamHaltParseError -> e) -> Conduit BS.ByteString m (Either CsvStreamRecordParseError a)+fromCsvStreamError opts h f = {-# SCC fromCsvStreamError_p #-} streamParser f $ decodeWith opts h++-- |+-- Like `fromNamedCsvStream` but allows for errors to be handled in the pipeline itself.+--+fromNamedCsvStreamError :: (FromNamedRecord a, MonadError e m) => DecodeOptions -> (CsvStreamHaltParseError -> e) -> Conduit BS.ByteString m (Either CsvStreamRecordParseError a)+fromNamedCsvStreamError opts f = {-# SCC fromCsvStreamError_p #-} streamHeaderParser f $ decodeByNameWith opts+```+ # 0.2.x ## 0.1.0 -> 0.2.0
cassava-conduit.cabal view
@@ -1,5 +1,5 @@ name: cassava-conduit-version: 0.2.2+version: 0.3.0 license: BSD3 license-file: etc/LICENCE.md author: Dom De Re@@ -38,6 +38,7 @@ , conduit == 1.2.* , conduit-extra == 1.1.* , mtl == 2.2.*+ , text == 1.2.* ghc-options: -Wall
src/Data/Csv/Conduit.hs view
@@ -12,6 +12,8 @@ module Data.Csv.Conduit ( -- * Types CsvParseError(..)+ , CsvStreamHaltParseError(..)+ , CsvStreamRecordParseError(..) -- * Conduits , fromCsv , fromCsvLiftError@@ -24,6 +26,7 @@ import LocalPrelude +import Control.Monad.Trans ( MonadTrans(..) ) import Control.Monad.Error.Class ( MonadError(..) ) import Data.Bifunctor ( first ) import qualified Data.ByteString as BS@@ -33,11 +36,25 @@ import Data.Csv ( FromNamedRecord, FromRecord, ToRecord, DecodeOptions, EncodeOptions, HasHeader, encodeWith ) import Data.Csv.Incremental ( HeaderParser(..), Parser(..), decodeByNameWith, decodeWith ) import Data.Foldable ( mapM_ )+import qualified Data.Text as T data CsvParseError =- CsvParseError BS.ByteString String- | IncrementalError String+ CsvParseError BS.ByteString T.Text+ | IncrementalError T.Text+ deriving (Show, Eq) +-- | When you want to include errors in the stream, this error type represents errors that halt the stream.+-- They do not appear inside the conduit and will instead get returned from running the conduit.+--+data CsvStreamHaltParseError = HaltingCsvParseError BS.ByteString T.Text -- ^ the remaining bytestring that was read in but not parsed yet, and the stringy error msg describing the fail.+ deriving (Show, Eq)++-- | When you want to include errors in the stream, these are the errors that can be included in the stream,+-- they are usually problems restricted to individual records, and streaming can resume from the next record+-- you just have to decide on something sensible to do with the per record errors.+--+data CsvStreamRecordParseError = CsvStreamRecordParseError T.Text deriving (Show, Eq) -- ^ The stringy error describing why this record could not be parsed.+ -- | -- Streams parsed records, Errors are not received in the stream but instead after the pipeline is executed, -- If you want to handle errors as they come and resume, see `fromCsvStreamError`@@ -72,14 +89,14 @@ -- | -- Same as `fromCsv` but allows for errors to be handled in the pipeline instead ---fromCsvStreamError :: (Monad m, FromRecord a) => DecodeOptions -> HasHeader -> Conduit BS.ByteString m (Either CsvParseError a)-fromCsvStreamError opts h = {-# SCC fromCsvStreamError_p #-} streamParser $ decodeWith opts h+fromCsvStreamError :: (FromRecord a, MonadError e m) => DecodeOptions -> HasHeader -> (CsvStreamHaltParseError -> e) -> Conduit BS.ByteString m (Either CsvStreamRecordParseError a)+fromCsvStreamError opts h f = {-# SCC fromCsvStreamError_p #-} streamParser f $ decodeWith opts h -- | -- Like `fromNamedCsvStream` but allows for errors to be handled in the pipeline itself. ---fromNamedCsvStreamError :: (Monad m, FromNamedRecord a) => DecodeOptions -> Conduit BS.ByteString m (Either CsvParseError a)-fromNamedCsvStreamError opts = {-# SCC fromCsvStreamError_p #-} streamHeaderParser $ decodeByNameWith opts+fromNamedCsvStreamError :: (FromNamedRecord a, MonadError e m) => DecodeOptions -> (CsvStreamHaltParseError -> e) -> Conduit BS.ByteString m (Either CsvStreamRecordParseError a)+fromNamedCsvStreamError opts f = {-# SCC fromCsvStreamError_p #-} streamHeaderParser f $ decodeByNameWith opts -- | -- Streams from csv to text, does not create headers...@@ -89,27 +106,27 @@ -- helpers -streamHeaderParser :: (Monad m) => HeaderParser (Parser a) -> Conduit BS.ByteString m (Either CsvParseError a)-streamHeaderParser (FailH rest errMsg) = {-# SCC streamHeaderParser_FailH_p #-} yield $ Left $ CsvParseError rest errMsg-streamHeaderParser (PartialH p) = {-# SCC streamHeaderParser_PartialH_p #-} await >>= maybe (return ()) (streamHeaderParser . p)-streamHeaderParser (DoneH _ p) = {-# SCC streamHeaderParser_DoneH_p #-} streamParser p+streamHeaderParser :: (MonadError e m) => (CsvStreamHaltParseError -> e) -> HeaderParser (Parser a) -> Conduit BS.ByteString m (Either CsvStreamRecordParseError a)+streamHeaderParser f (FailH rest errMsg) = {-# SCC streamHeaderParser_FailH_p #-} lift . throwError . f $ HaltingCsvParseError rest (T.pack errMsg)+streamHeaderParser f (PartialH p) = {-# SCC streamHeaderParser_PartialH_p #-} await >>= maybe (return ()) (streamHeaderParser f . p)+streamHeaderParser f (DoneH _ p) = {-# SCC streamHeaderParser_DoneH_p #-} streamParser f p -streamParser :: (Monad m) => Parser a -> Conduit BS.ByteString m (Either CsvParseError a)-streamParser (Fail rest errMsg) = yield $ Left $ CsvParseError rest errMsg-streamParser (Many rs p) = do+streamParser :: (MonadError e m) => (CsvStreamHaltParseError -> e) -> Parser a -> Conduit BS.ByteString m (Either CsvStreamRecordParseError a)+streamParser f (Fail rest errMsg) = {-# SCC streamParser_Fail_p #-} lift . throwError . f $ HaltingCsvParseError rest (T.pack errMsg)+streamParser f (Many rs p) = {-# SCC streamParser_Many_p #-} do -- send the results down the stream..- mapM_ (yield . first IncrementalError) rs+ mapM_ (yield . first (CsvStreamRecordParseError . T.pack)) rs -- wait for more.. more <- await- maybe (return ()) (streamParser . p) more-streamParser (Done rs) = mapM_ (yield . first IncrementalError) rs+ maybe (return ()) (streamParser f . p) more+streamParser _ (Done rs) = {-# SCC streamParser_Done_p #-} mapM_ (yield . first (CsvStreamRecordParseError . T.pack)) rs terminatingStreamHeaderParser :: (Monad m, MonadError e m) => (CsvParseError -> e) -> HeaderParser (Parser a) -> Conduit BS.ByteString m a-terminatingStreamHeaderParser f (FailH rest errMsg) = {-# SCC terminatingStreamHeaderParser_FailH_p #-} mapM $ const $ throwError $ f $ CsvParseError rest errMsg+terminatingStreamHeaderParser f (FailH rest errMsg) = {-# SCC terminatingStreamHeaderParser_FailH_p #-} lift . throwError . f . CsvParseError rest . T.pack $ errMsg terminatingStreamHeaderParser f (PartialH p) = {-# SCC terminatingStreamHeaderParser_PartialH_p #-} await >>= maybe (return ()) (terminatingStreamHeaderParser f . p) terminatingStreamHeaderParser f (DoneH _ p) = {-# SCC terminatingStreamHeaderParser_DoneH_p #-} terminatingStreamParser f p @@ -118,11 +135,11 @@ => (CsvParseError -> e) -> Parser a -> Conduit BS.ByteString m a-terminatingStreamParser f (Fail rest errMsg) = {-# SCC terminatingStreamParser_Fail_p #-} mapM $ const $ throwError $ f $ CsvParseError rest errMsg+terminatingStreamParser f (Fail rest errMsg) = {-# SCC terminatingStreamParser_Fail_p #-} lift . throwError . f . CsvParseError rest . T.pack $ errMsg terminatingStreamParser f (Many ers p) = {-# SCC terminatingStreamParser_Many_p #-} let errorHandler :: (Monad m, MonadError e m) => (CsvParseError -> e) -> String -> Conduit BS.ByteString m a- errorHandler f' = mapM . const . throwError . f' . IncrementalError+ errorHandler f' = mapM . const . throwError . f' . IncrementalError . T.pack safeResultHandler :: (Monad m)@@ -137,5 +154,5 @@ in -- send the results down the stream.. either (errorHandler f) (safeResultHandler p (terminatingStreamParser f)) (sequence ers)-terminatingStreamParser f (Done rs) = {-# SCC terminatingStreamParser_Done_p #-} either (mapM . const . throwError . f . IncrementalError) (mapM_ yield) (sequence rs)+terminatingStreamParser f (Done rs) = {-# SCC terminatingStreamParser_Done_p #-} either (lift . throwError . f . IncrementalError . T.pack) (mapM_ yield) (sequence rs)