Как написать предикат для Streaming.Prelude.takeWhile - PullRequest
Новая
       8

Как написать предикат для Streaming.Prelude.takeWhile

0 голосов
/ 15 мая 2018

Я пишу программу для использования конечной точки JSON API. Сложность состоит в том, что я хочу продолжать итерацию страниц (путем передачи параметра _page), и итерация должна прекратиться, как только API вернет пустой список (обратите внимание, что это означает, что нам нужно изучить ответ, чтобы решить, следует ли нам прекрати итерацию).

Я использую servant, чтобы упростить вызов API, и нетрудно написать рекурсивную функцию для сбора результатов и завершения в вышеупомянутом условии. Но это может занять много памяти, если страниц результатов много.

Итак, я начал изучать библиотеку streaming. Мне удалось создать поток вызовов API, но я пытаюсь написать условие завершения. Я действительно хочу написать завершающее условие без выполнения метода runClientM клиента. Поэтому вместо написания рекурсивной функции, которая явно вызывает API и собирает результаты, я хотел бы сделать что-то вроде: 

map runClientM $ takeWhile hasData $ map createPageRequest $ [1..]

Я дошел до того, что могу создавать и запускать поток запросов, но не могу написать условие завершения. Я не уверен, что не могу заставить работать типы, потому что мне не хватает функции, связывающей все типы (liftM или что-то в этом роде ...), или потому что система типов мешает мне делать то, что я пытаюсь сделать, потому что это глупо (все делают незаконные состояния представительными идея).

Во всяком случае, вот код (мне нужно что-то заменить _, predicate1 и predicate2 не компилировать). Любые указатели будут оценены. 

#!/usr/bin/env stack

{-# LANGUAGE DeriveGeneric #-}
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE TypeOperators #-}

import Servant.Client
import Network.HTTP.Client (newManager, defaultManagerSettings)

import Data.Proxy
import Servant.API

import Data.Aeson
import GHC.Generics

import Streaming
import qualified Streaming.Prelude as S

-- data type
data BlogPost = BlogPost
  { id :: Integer
  , title :: String
  } deriving (Show, Generic)

instance FromJSON BlogPost


-- api client
type API = "posts" :> QueryParam "_page" Integer :> Get '[JSON] [BlogPost]
api :: Proxy API
api = Proxy
posts :: Maybe Integer -> ClientM [BlogPost]
posts = client api


requestStream :: (Monad m) => Stream (Of (ClientM [BlogPost])) m ()
requestStream = S.takeWhile _ $ S.map posts $ S.each pages
  where
    pages = [Just p | p <- [1..]]
    predicate1 (Right v) = True
    predicate1 (Left e) = False
    predicate2 request = do
      r <- request
      case r of
        Right v -> return $ length v /= 0
        Left e -> return False


main :: IO ()
main = do
  manager' <- newManager defaultManagerSettings
  let url = ClientEnv manager' (BaseUrl Http "jsonplaceholder.typicode.com" 80 "")
  S.print $ S.mapM (\x -> runClientM x url) requestStream
  print "done"

РЕДАКТИРОВАТЬ: я просто делаю S.print на данный момент, но я хочу сделать некоторые другие действия, такие как отправить эти данные в базу данных.

РЕДАКТИРОВАТЬ: Вот ошибка, которую я получаю при использовании, например, predicate1. Я не понимаю, почему фактический тип теперь имеет Either a0 b0 вместо ClientM [BlogPost], то есть без вызова S.takeWhile. 

test.hs:37:17: error:
    • Couldn't match type ‘Either a0 b0’ with ‘ClientM [BlogPost]’
      Expected type: Stream (Of (ClientM [BlogPost])) m ()
        Actual type: Stream (Of (Either a0 b0)) m ()
    • In the expression:
        S.takeWhile predicate1 $ S.map posts $ S.each pages
      In an equation for ‘requestStream’:
          requestStream
            = S.takeWhile predicate1 $ S.map posts $ S.each pages
            where
                pages = [Just p | p <- [1 .. ]]
                predicate1 (Right v) = True
                predicate1 (Left e) = False
                predicate2 request
                  = do r <- request
                       ....
   |
37 | requestStream = S.takeWhile predicate1 $ S.map posts $ S.each pages
   |                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

test.hs:37:42: error:
    • Couldn't match type ‘ClientM [BlogPost]’ with ‘Either a0 b0’
      Expected type: Stream (Of (Either a0 b0)) m ()
        Actual type: Stream (Of (ClientM [BlogPost])) m ()
    • In the second argument of ‘($)’, namely
        ‘S.map posts $ S.each pages’
      In the expression:
        S.takeWhile predicate1 $ S.map posts $ S.each pages
      In an equation for ‘requestStream’:
          requestStream
            = S.takeWhile predicate1 $ S.map posts $ S.each pages
            where
                pages = [Just p | p <- [1 .. ]]
                predicate1 (Right v) = True
                predicate1 (Left e) = False
                predicate2 request
                  = do r <- request
                       ....
   |
37 | requestStream = S.takeWhile predicate1 $ S.map posts $ S.each pages
   |                                          ^^^^^^^^^^^^^^^^^^^^^^^^^^

1 Ответ

0 голосов
/ 15 мая 2018

пример серванта-клиента с потоковой передачей и без нее

Документы серванта содержат большую часть информации, необходимой для начала работы с потоковой передачей, но на самом деле они перестают вызывать streamAPI в примере. Ссылка на пример .

пример без потоковой передачи 

{-# LANGUAGE DataKinds #-}
{-# LANGUAGE DeriveGeneric #-}
{-# LANGUAGE TypeOperators #-}

module Main where

import Data.Aeson
import Data.Proxy
import GHC.Generics
import Network.HTTP.Client (newManager, defaultManagerSettings)
import Servant.API
import Servant.Client

-- data type
data BlogPost = BlogPost
  { id :: Integer
  , title :: String
  } deriving (Show, Generic)

instance FromJSON BlogPost

-- api client
type API = "posts" :> QueryParam "_page" Integer :> Get '[JSON] [BlogPost]

api :: Proxy API
api = Proxy

posts :: Maybe Integer -> ClientM [BlogPost]
posts = client api


main :: IO ()
main = do
  manager' <- newManager defaultManagerSettings
  res <- runClientM 
           (posts (Just 1)) 
           (mkClientEnv manager' 
             (BaseUrl 
               Http 
               "jsonplaceholder.typicode.com" 
               80 
               ""))
  case res of
    Left err -> putStrLn $ "Error: " ++ show err
    Right (post) -> print post

пример потоковой передачи

Добавьте следующий код в файл выше. 

type StreamAPI = "post" :> QueryParam "_page" Integer :> StreamGet NewlineFraming JSON (ResultStream [BlogPost])

streamAPI :: Proxy StreamAPI
streamAPI = Proxy

posStream :: Maybe Integer -> ClientM (ResultStream [BlogPost])
posStream = client streamAPI

printResultStream :: Show a => ResultStream a -> IO ()
printResultStream (ResultStream k) = k $ \getResult ->
       let loop = do
            r <- getResult
            case r of
                Nothing -> return ()
                Just x -> print x >> loop
       in loop

Затем выполните в main следующую команду: 

streamRes <- 
  runClientM 
    (posStream (Just 1)) 
    (mkClientEnv 
      manager' 
      (BaseUrl 
        Http 
        "jsonplaceholder.typicode.com" 
        80 
        ""))

case streamRes of
  Left err -> putStrLn $ "Error: " ++ show err
  Right (stream) -> printResultStream stream

Редактировать: изначально я указал на некоторые ошибки в предоставленном коде.A изменил это на полностью компилируемый пример.

...