在Conduit中使用持久性

Question

首先,我要完成的任务的简化版本:我有几个大文件(相当于30GB),我想修剪重复的条目.为此,我建立了一个数据散列数据库,逐个打开文件,散列每个项目,并将其记录在数据库和输出文件中,如果它的散列不在数据库中.

我知道如何使用迭代器,枚举器,我想尝试管道.我也知道如何用管道来做,但现在我想使用管道和持久性.我遇到了类型问题,可能还有整个概念ResourceT.

这里有一些伪代码来说明问题:

withSqlConn "foo.db" $ runSqlConn $ runResourceT $ 
     sourceFile "in" $= parseBytes $= dbAction $= serialize $$ sinkFile "out"

问题在于dbAction功能.我想自然地在这里访问数据库.由于它所做的动作基本上只是一个过滤器,我首先想到这样写:

dbAction = CL.mapMaybeM p
     where p :: (MonadIO m, MonadBaseControl IO (SqlPersist m)) => DataType -> m (Maybe DataType)
           p = lift $ putStrLn "foo" -- fine
           insert $ undefined -- type error!
           return undefined

我得到的具体错误是:

Could not deduce (m ~ b0 m0)
from the context (MonadIO m, MonadBaseControl IO (SqlPersist m))
  bound by the type signature for
             p :: (MonadIO m, MonadBaseControl IO (SqlPersist m)) =>
                           DataType -> m (Maybe DataType)
  at tools/clean-wac.hs:(33,1)-(34,34)
  `m' is a rigid type variable bound by
      the type signature for
        p :: (MonadIO m, MonadBaseControl IO (SqlPersist m)) =>
                      DataType -> m (Maybe (DataType))
      at tools/clean-wac.hs:33:1
Expected type: m (Key b0 val0)
  Actual type: b0 m0 (Key b0 val0)

请注意,这可能是由于我在设计类型签名时所做的错误假设.如果我注释掉类型签名并删除该lift语句,则错误消息将变为:

No instance for (PersistStore ResourceT (SqlPersist IO))
  arising from a use of `p'
Possible fix:
  add an instance declaration for
  (PersistStore ResourceT (SqlPersist IO))
In the first argument of `CL.mapMaybeM', namely `p'

那么这意味着我们根本无法访问PersistStore通道ResourceT？

我也不能写自己的导管,不使用CL.mapMaybeM:

dbAction = filterP
filterP :: (MonadIO m, MonadBaseControl IO (SqlPersist m)) => Conduit DataType m DataType
filterP = loop
    where loop = awaitE >>= either return go
          go s = do lift $ insert $ undefined -- again, type error
                    loop

这导致了另一种我不完全理解的类型错误.

Could not deduce (m ~ b0 m0)
from the context (MonadIO m, MonadBaseControl IO (SqlPersist m))
  bound by the type signature for
             filterP :: (MonadIO m,
                                 MonadBaseControl IO (SqlPersist m)) =>
                                Conduit DataType m DataType
     `m' is a rigid type variable bound by
      the type signature for
        filterP :: (MonadIO m,
                            MonadBaseControl IO (SqlPersist m)) =>
                           Conduit DataType m DataType
Expected type: Conduit DataType m DataType
  Actual type: Pipe
                 DataType DataType DataType () (b0 m0) ()
In the expression: loop
In an equation for `filterP'

所以,我的问题是:是否有可能像我打算在管道内使用持久性？如果,如何？我知道因为我可以liftIO在管道中使用,所以我可以去使用,比如说HDBC,但是我想明确地使用持久性来理解它是如何工作的,因为我喜欢它的db-backend不可知论.

Answer 1

下面的代码对我来说很好.框架是否有可能在此期间继续发展并且事情现在正常起作用？

但请注意,由于世界已经发生了一些变化,或者我没有完成所有代码,因此我必须进行以下更改.我使用conduit-1.0.9.3和持久性1.3.0与GHC 7.6.3.

• 省略parseBytes,serialise因为我没有你的定义DataType = ByteString而是定义了.

• Proxy为该undefined值引入了一个参数和一个显式类型签名,以避免类型族注入的问题.这些可能不会出现在您的真实代码中,因为它将具有具体或外部确定的类型val.

• 使用await而不是awaitE仅用()作替代Left案例的类型,awaitE已经退休.

• 通过虚拟Connection创建函数withSqlConn- 也许我应该使用一些Sqlite特定的功能？

这是代码:

{-# LANGUAGE FlexibleContexts, NoMonomorphismRestriction,
             TypeFamilies, ScopedTypeVariables #-}

module So133331988 where

import Control.Monad.Trans
import Database.Persist.Sql
import Data.ByteString
import Data.Conduit
import Data.Conduit.Binary
import Data.Proxy

test proxy =
    withSqlConn (return (undefined "foo.db")) $ runSqlConn $ runResourceT $ 
         sourceFile "in" $= dbAction proxy $$ sinkFile "out"

dbAction = filterP

type DataType = ByteString

filterP
    :: forall m val
     . ( MonadIO m, MonadBaseControl IO (SqlPersist m)
       , PersistStore m, PersistEntity val
       , PersistEntityBackend val ~ PersistMonadBackend m)
    => Proxy val
    -> Conduit DataType m DataType
filterP Proxy = loop
    where loop = await >>= maybe (return ()) go
          go s = do lift $ insert (undefined :: val)
                    loop