git.iko.soy/typeable/bloodhound
on 977d2e8 help sign in
typeable/bloodhound/src/Database/Bloodhound/Client.hs
1208 lines 53 kB history raw
1{-# LANGUAGE OverloadedStrings #-}
2{-# LANGUAGE RecordWildCards #-}
3{-# LANGUAGE TupleSections #-}
5-------------------------------------------------------------------------------
6-- |
7-- Module : Database.Bloodhound.Client
8-- Copyright : (C) 2014, 2018 Chris Allen
9-- License : BSD-style (see the file LICENSE)
10-- Maintainer : Chris Allen <cma@bitemyapp.com>
11-- Stability : provisional
12-- Portability : OverloadedStrings
13--
14-- Client side functions for talking to Elasticsearch servers.
15--
16-------------------------------------------------------------------------------
18module Database.Bloodhound.Client
19 ( -- * Bloodhound client functions
20 -- | The examples in this module assume the following code has been run.
21 -- The :{ and :} will only work in GHCi. You'll only need the data types
22 -- and typeclass instances for the functions that make use of them.
24 -- $setup
25 withBH
26 -- ** Indices
27 , createIndex
28 , createIndexWith
29 , flushIndex
30 , deleteIndex
31 , updateIndexSettings
32 , getIndexSettings
33 , forceMergeIndex
34 , indexExists
35 , openIndex
36 , closeIndex
37 , listIndices
38 , waitForYellowIndex
39 -- *** Index Aliases
40 , updateIndexAliases
41 , getIndexAliases
42 , deleteIndexAlias
43 -- *** Index Templates
44 , putTemplate
45 , templateExists
46 , deleteTemplate
47 -- ** Mapping
48 , putMapping
49 -- ** Documents
50 , indexDocument
51 , updateDocument
52 , getDocument
53 , documentExists
54 , deleteDocument
55 -- ** Searching
56 , searchAll
57 , searchByIndex
58 , searchByIndices
59 , searchByType
60 , scanSearch
61 , getInitialScroll
62 , getInitialSortedScroll
63 , advanceScroll
64 , refreshIndex
65 , mkSearch
66 , mkAggregateSearch
67 , mkHighlightSearch
68 , bulk
69 , pageSearch
70 , mkShardCount
71 , mkReplicaCount
72 , getStatus
73 -- ** Snapshot/Restore
74 -- *** Snapshot Repos
75 , getSnapshotRepos
76 , updateSnapshotRepo
77 , verifySnapshotRepo
78 , deleteSnapshotRepo
79 -- *** Snapshots
80 , createSnapshot
81 , getSnapshots
82 , deleteSnapshot
83 -- *** Restoring Snapshots
84 , restoreSnapshot
85 -- ** Nodes
86 , getNodesInfo
87 , getNodesStats
88 -- ** Request Utilities
89 , encodeBulkOperations
90 , encodeBulkOperation
91 -- * Authentication
92 , basicAuthHook
93 -- * Reply-handling tools
94 , isVersionConflict
95 , isSuccess
96 , isCreated
97 , parseEsResponse
98 )
99 where
101import qualified Blaze.ByteString.Builder as BB
102import Control.Applicative as A
103import Control.Monad
104import Control.Monad.Catch
105import Control.Monad.IO.Class
106import Data.Aeson
107import Data.ByteString.Lazy.Builder
108import qualified Data.ByteString.Lazy.Char8 as L
109import Data.Foldable (toList)
110import qualified Data.HashMap.Strict as HM
111import Data.Ix
112import qualified Data.List as LS (filter, foldl')
113import Data.List.NonEmpty (NonEmpty (..))
114import Data.Maybe (catMaybes, fromMaybe, isJust)
115import Data.Monoid
116import Data.Text (Text)
117import qualified Data.Text as T
118import qualified Data.Text.Encoding as T
119import Data.Time.Clock
120import qualified Data.Vector as V
121import Network.HTTP.Client
122import qualified Network.HTTP.Types.Method as NHTM
123import qualified Network.HTTP.Types.Status as NHTS
124import qualified Network.HTTP.Types.URI as NHTU
125import qualified Network.URI as URI
126import Prelude hiding (filter, head)
128import Database.Bloodhound.Types
130-- $setup
131-- >>> :set -XOverloadedStrings
132-- >>> :set -XDeriveGeneric
133-- >>> import Database.Bloodhound
134-- >>> import Network.HTTP.Client
135-- >>> let testServer = (Server "http://localhost:9200")
136-- >>> let runBH' = withBH defaultManagerSettings testServer
137-- >>> let testIndex = IndexName "twitter"
138-- >>> let testMapping = MappingName "tweet"
139-- >>> let defaultIndexSettings = IndexSettings (ShardCount 1) (ReplicaCount 0)
140-- >>> data TweetMapping = TweetMapping deriving (Eq, Show)
141-- >>> _ <- runBH' $ deleteIndex testIndex
142-- >>> _ <- runBH' $ deleteIndex (IndexName "didimakeanindex")
143-- >>> import GHC.Generics
144-- >>> import Data.Time.Calendar (Day (..))
145-- >>> import Data.Time.Clock (UTCTime (..), secondsToDiffTime)
146-- >>> :{
147--instance ToJSON TweetMapping where
148-- toJSON TweetMapping =
149-- object ["properties" .=
150-- object ["location" .=
151-- object ["type" .= ("geo_point" :: Text)]]]
152--data Location = Location { lat :: Double
153-- , lon :: Double } deriving (Eq, Generic, Show)
154--data Tweet = Tweet { user :: Text
155-- , postDate :: UTCTime
156-- , message :: Text
157-- , age :: Int
158-- , location :: Location } deriving (Eq, Generic, Show)
159--exampleTweet = Tweet { user = "bitemyapp"
160-- , postDate = UTCTime
161-- (ModifiedJulianDay 55000)
162-- (secondsToDiffTime 10)
163-- , message = "Use haskell!"
164-- , age = 10000
165-- , location = Location 40.12 (-71.34) }
166--instance ToJSON Tweet where
167-- toJSON = genericToJSON defaultOptions
168--instance FromJSON Tweet where
169-- parseJSON = genericParseJSON defaultOptions
170--instance ToJSON Location where
171-- toJSON = genericToJSON defaultOptions
172--instance FromJSON Location where
173-- parseJSON = genericParseJSON defaultOptions
174--data BulkTest = BulkTest { name :: Text } deriving (Eq, Generic, Show)
175--instance FromJSON BulkTest where
176-- parseJSON = genericParseJSON defaultOptions
177--instance ToJSON BulkTest where
178-- toJSON = genericToJSON defaultOptions
179-- :}
181-- | 'mkShardCount' is a straight-forward smart constructor for 'ShardCount'
182-- which rejects 'Int' values below 1 and above 1000.
183--
184-- >>> mkShardCount 10
185-- Just (ShardCount 10)
186mkShardCount :: Int -> Maybe ShardCount
187mkShardCount n
188 | n < 1 = Nothing
189 | n > 1000 = Nothing
190 | otherwise = Just (ShardCount n)
192-- | 'mkReplicaCount' is a straight-forward smart constructor for 'ReplicaCount'
193-- which rejects 'Int' values below 0 and above 1000.
194--
195-- >>> mkReplicaCount 10
196-- Just (ReplicaCount 10)
197mkReplicaCount :: Int -> Maybe ReplicaCount
198mkReplicaCount n
199 | n < 0 = Nothing
200 | n > 1000 = Nothing -- ...
201 | otherwise = Just (ReplicaCount n)
203emptyBody :: L.ByteString
204emptyBody = L.pack ""
206dispatch :: MonadBH m
207 => Method
208 -> Text
209 -> Maybe L.ByteString
210 -> m Reply
211dispatch dMethod url body = do
212 initReq <- liftIO $ parseUrl' url
213 reqHook <- bhRequestHook A.<$> getBHEnv
214 let reqBody = RequestBodyLBS $ fromMaybe emptyBody body
215 req <- liftIO
216 $ reqHook
217 $ setRequestIgnoreStatus
218 $ initReq { method = dMethod
219 , requestHeaders =
220 -- "application/x-ndjson" for bulk
221 ("Content-Type", "application/json") : requestHeaders initReq
222 , requestBody = reqBody }
223 -- req <- liftIO $ reqHook $ setRequestIgnoreStatus $ initReq { method = dMethod
224 -- , requestBody = reqBody }
225 mgr <- bhManager <$> getBHEnv
226 liftIO $ httpLbs req mgr
228joinPath' :: [Text] -> Text
229joinPath' = T.intercalate "/"
231joinPath :: MonadBH m => [Text] -> m Text
232joinPath ps = do
233 Server s <- bhServer <$> getBHEnv
234 return $ joinPath' (s:ps)
236appendSearchTypeParam :: Text -> SearchType -> Text
237appendSearchTypeParam originalUrl st = addQuery params originalUrl
238 where stText = "search_type"
239 params
240 | st == SearchTypeDfsQueryThenFetch = [(stText, Just "dfs_query_then_fetch")]
241 -- used to catch 'SearchTypeQueryThenFetch', which is also the default
242 | otherwise = []
244-- | Severely dumbed down query renderer. Assumes your data doesn't
245-- need any encoding
246addQuery :: [(Text, Maybe Text)] -> Text -> Text
247addQuery q u = u <> rendered
248 where
249 rendered =
250 T.decodeUtf8 $ BB.toByteString $ NHTU.renderQueryText prependQuestionMark q
251 prependQuestionMark = True
253bindM2 :: (Applicative m, Monad m) => (a -> b -> m c) -> m a -> m b -> m c
254bindM2 f ma mb = join (f <$> ma <*> mb)
256-- | Convenience function that sets up a manager and BHEnv and runs
257-- the given set of bloodhound operations. Connections will be
258-- pipelined automatically in accordance with the given manager
259-- settings in IO. If you've got your own monad transformer stack, you
260-- should use 'runBH' directly.
261withBH :: ManagerSettings -> Server -> BH IO a -> IO a
262withBH ms s f = do
263 mgr <- newManager ms
264 let env = mkBHEnv s mgr
265 runBH env f
267-- Shortcut functions for HTTP methods
268delete :: MonadBH m => Text -> m Reply
269delete = flip (dispatch NHTM.methodDelete) Nothing
270get :: MonadBH m => Text -> m Reply
271get = flip (dispatch NHTM.methodGet) Nothing
272head :: MonadBH m => Text -> m Reply
273head = flip (dispatch NHTM.methodHead) Nothing
274put :: MonadBH m => Text -> Maybe L.ByteString -> m Reply
275put = dispatch NHTM.methodPut
276post :: MonadBH m => Text -> Maybe L.ByteString -> m Reply
277post = dispatch NHTM.methodPost
279-- indexDocument s ix name doc = put (root </> s </> ix </> name </> doc) (Just encode doc)
280-- http://hackage.haskell.org/package/http-client-lens-0.1.0/docs/Network-HTTP-Client-Lens.html
281-- https://github.com/supki/libjenkins/blob/master/src/Jenkins/Rest/Internal.hs
283-- | 'getStatus' fetches the 'Status' of a 'Server'
284--
285-- >>> serverStatus <- runBH' getStatus
286-- >>> fmap tagline (serverStatus)
287-- Just "You Know, for Search"
288getStatus :: MonadBH m => m (Maybe Status)
289getStatus = do
290 response <- get =<< url
291 return $ decode (responseBody response)
292 where url = joinPath []
294-- | 'getSnapshotRepos' gets the definitions of a subset of the
295-- defined snapshot repos.
296getSnapshotRepos
297 :: ( MonadBH m
298 , MonadThrow m
299 )
300 => SnapshotRepoSelection
301 -> m (Either EsError [GenericSnapshotRepo])
302getSnapshotRepos sel = fmap (fmap unGSRs) . parseEsResponse =<< get =<< url
303 where
304 url = joinPath ["_snapshot", selectorSeg]
305 selectorSeg = case sel of
306 AllSnapshotRepos -> "_all"
307 SnapshotRepoList (p :| ps) -> T.intercalate "," (renderPat <$> (p:ps))
308 renderPat (RepoPattern t) = t
309 renderPat (ExactRepo (SnapshotRepoName t)) = t
312-- | Wrapper to extract the list of 'GenericSnapshotRepo' in the
313-- format they're returned in
314newtype GSRs = GSRs { unGSRs :: [GenericSnapshotRepo] }
317instance FromJSON GSRs where
318 parseJSON = withObject "Collection of GenericSnapshotRepo" parse
319 where
320 parse = fmap GSRs . mapM (uncurry go) . HM.toList
321 go rawName = withObject "GenericSnapshotRepo" $ \o ->
322 GenericSnapshotRepo (SnapshotRepoName rawName) <$> o .: "type"
323 <*> o .: "settings"
326-- | Create or update a snapshot repo
327updateSnapshotRepo
328 :: ( MonadBH m
329 , SnapshotRepo repo
330 )
331 => SnapshotRepoUpdateSettings
332 -- ^ Use 'defaultSnapshotRepoUpdateSettings' if unsure
333 -> repo
334 -> m Reply
335updateSnapshotRepo SnapshotRepoUpdateSettings {..} repo =
336 bindM2 put url (return (Just body))
337 where
338 url = addQuery params <$> joinPath ["_snapshot", snapshotRepoName gSnapshotRepoName]
339 params
340 | repoUpdateVerify = []
341 | otherwise = [("verify", Just "false")]
342 body = encode $ object [ "type" .= gSnapshotRepoType
343 , "settings" .= gSnapshotRepoSettings
344 ]
345 GenericSnapshotRepo {..} = toGSnapshotRepo repo
349-- | Verify if a snapshot repo is working. __NOTE:__ this API did not
350-- make it into Elasticsearch until 1.4. If you use an older version,
351-- you will get an error here.
352verifySnapshotRepo
353 :: ( MonadBH m
354 , MonadThrow m
355 )
356 => SnapshotRepoName
357 -> m (Either EsError SnapshotVerification)
358verifySnapshotRepo (SnapshotRepoName n) =
359 parseEsResponse =<< bindM2 post url (return Nothing)
360 where
361 url = joinPath ["_snapshot", n, "_verify"]
364deleteSnapshotRepo :: MonadBH m => SnapshotRepoName -> m Reply
365deleteSnapshotRepo (SnapshotRepoName n) = delete =<< url
366 where
367 url = joinPath ["_snapshot", n]
370-- | Create and start a snapshot
371createSnapshot
372 :: (MonadBH m)
373 => SnapshotRepoName
374 -> SnapshotName
375 -> SnapshotCreateSettings
376 -> m Reply
377createSnapshot (SnapshotRepoName repoName)
378 (SnapshotName snapName)
379 SnapshotCreateSettings {..} =
380 bindM2 put url (return (Just body))
381 where
382 url = addQuery params <$> joinPath ["_snapshot", repoName, snapName]
383 params = [("wait_for_completion", Just (boolQP snapWaitForCompletion))]
384 body = encode $ object prs
385 prs = catMaybes [ ("indices" .=) . indexSelectionName <$> snapIndices
386 , Just ("ignore_unavailable" .= snapIgnoreUnavailable)
387 , Just ("ignore_global_state" .= snapIncludeGlobalState)
388 , Just ("partial" .= snapPartial)
389 ]
392indexSelectionName :: IndexSelection -> Text
393indexSelectionName AllIndexes = "_all"
394indexSelectionName (IndexList (i :| is)) = T.intercalate "," (renderIndex <$> (i:is))
395 where
396 renderIndex (IndexName n) = n
399-- | Get info about known snapshots given a pattern and repo name.
400getSnapshots
401 :: ( MonadBH m
402 , MonadThrow m
403 )
404 => SnapshotRepoName
405 -> SnapshotSelection
406 -> m (Either EsError [SnapshotInfo])
407getSnapshots (SnapshotRepoName repoName) sel =
408 fmap (fmap unSIs) . parseEsResponse =<< get =<< url
409 where
410 url = joinPath ["_snapshot", repoName, snapPath]
411 snapPath = case sel of
412 AllSnapshots -> "_all"
413 SnapshotList (s :| ss) -> T.intercalate "," (renderPath <$> (s:ss))
414 renderPath (SnapPattern t) = t
415 renderPath (ExactSnap (SnapshotName t)) = t
418newtype SIs = SIs { unSIs :: [SnapshotInfo] }
421instance FromJSON SIs where
422 parseJSON = withObject "Collection of SnapshotInfo" parse
423 where
424 parse o = SIs <$> o .: "snapshots"
427-- | Delete a snapshot. Cancels if it is running.
428deleteSnapshot :: MonadBH m => SnapshotRepoName -> SnapshotName -> m Reply
429deleteSnapshot (SnapshotRepoName repoName) (SnapshotName snapName) =
430 delete =<< url
431 where
432 url = joinPath ["_snapshot", repoName, snapName]
435-- | Restore a snapshot to the cluster See
436-- <https://www.elastic.co/guide/en/elasticsearch/reference/1.7/modules-snapshots.html#_restore>
437-- for more details.
438restoreSnapshot
439 :: MonadBH m
440 => SnapshotRepoName
441 -> SnapshotName
442 -> SnapshotRestoreSettings
443 -- ^ Start with 'defaultSnapshotRestoreSettings' and customize
444 -- from there for reasonable defaults.
445 -> m Reply
446restoreSnapshot (SnapshotRepoName repoName)
447 (SnapshotName snapName)
448 SnapshotRestoreSettings {..} = bindM2 post url (return (Just body))
449 where
450 url = addQuery params <$> joinPath ["_snapshot", repoName, snapName, "_restore"]
451 params = [("wait_for_completion", Just (boolQP snapRestoreWaitForCompletion))]
452 body = encode (object prs)
455 prs = catMaybes [ ("indices" .=) . indexSelectionName <$> snapRestoreIndices
456 , Just ("ignore_unavailable" .= snapRestoreIgnoreUnavailable)
457 , Just ("include_global_state" .= snapRestoreIncludeGlobalState)
458 , ("rename_pattern" .=) <$> snapRestoreRenamePattern
459 , ("rename_replacement" .=) . renderTokens <$> snapRestoreRenameReplacement
460 , Just ("include_aliases" .= snapRestoreIncludeAliases)
461 , ("index_settings" .= ) <$> snapRestoreIndexSettingsOverrides
462 , ("ignore_index_settings" .= ) <$> snapRestoreIgnoreIndexSettings
463 ]
464 renderTokens (t :| ts) = mconcat (renderToken <$> (t:ts))
465 renderToken (RRTLit t) = t
466 renderToken RRSubWholeMatch = "$0"
467 renderToken (RRSubGroup g) = T.pack (show (rrGroupRefNum g))
470getNodesInfo
471 :: ( MonadBH m
472 , MonadThrow m
473 )
474 => NodeSelection
475 -> m (Either EsError NodesInfo)
476getNodesInfo sel = parseEsResponse =<< get =<< url
477 where
478 url = joinPath ["_nodes", selectionSeg]
479 selectionSeg = case sel of
480 LocalNode -> "_local"
481 NodeList (l :| ls) -> T.intercalate "," (selToSeg <$> (l:ls))
482 AllNodes -> "_all"
483 selToSeg (NodeByName (NodeName n)) = n
484 selToSeg (NodeByFullNodeId (FullNodeId i)) = i
485 selToSeg (NodeByHost (Server s)) = s
486 selToSeg (NodeByAttribute (NodeAttrName a) v) = a <> ":" <> v
488getNodesStats
489 :: ( MonadBH m
490 , MonadThrow m
491 )
492 => NodeSelection
493 -> m (Either EsError NodesStats)
494getNodesStats sel = parseEsResponse =<< get =<< url
495 where
496 url = joinPath ["_nodes", selectionSeg, "stats"]
497 selectionSeg = case sel of
498 LocalNode -> "_local"
499 NodeList (l :| ls) -> T.intercalate "," (selToSeg <$> (l:ls))
500 AllNodes -> "_all"
501 selToSeg (NodeByName (NodeName n)) = n
502 selToSeg (NodeByFullNodeId (FullNodeId i)) = i
503 selToSeg (NodeByHost (Server s)) = s
504 selToSeg (NodeByAttribute (NodeAttrName a) v) = a <> ":" <> v
506-- | 'createIndex' will create an index given a 'Server', 'IndexSettings', and an 'IndexName'.
507--
508-- >>> response <- runBH' $ createIndex defaultIndexSettings (IndexName "didimakeanindex")
509-- >>> respIsTwoHunna response
510-- True
511-- >>> runBH' $ indexExists (IndexName "didimakeanindex")
512-- True
513createIndex :: MonadBH m => IndexSettings -> IndexName -> m Reply
514createIndex indexSettings (IndexName indexName) =
515 bindM2 put url (return body)
516 where url = joinPath [indexName]
517 body = Just $ encode indexSettings
519-- | Create an index, providing it with any number of settings. This
520-- is more expressive than 'createIndex' but makes is more verbose
521-- for the common case of configuring only the shard count and
522-- replica count.
523createIndexWith :: MonadBH m
524 => [UpdatableIndexSetting]
525 -> Int -- ^ shard count
526 -> IndexName
527 -> m Reply
528createIndexWith updates shards (IndexName indexName) =
529 bindM2 put url (return (Just body))
530 where url = joinPath [indexName]
531 body = encode $ object
532 ["settings" .= deepMerge
533 ( HM.singleton "index.number_of_shards" (toJSON shards) :
534 [u | Object u <- toJSON <$> updates]
535 )
536 ]
538-- | 'flushIndex' will flush an index given a 'Server' and an 'IndexName'.
539flushIndex :: MonadBH m => IndexName -> m Reply
540flushIndex (IndexName indexName) = do
541 path <- joinPath [indexName, "_flush"]
542 post path Nothing
544-- | 'deleteIndex' will delete an index given a 'Server' and an 'IndexName'.
545--
546-- >>> _ <- runBH' $ createIndex defaultIndexSettings (IndexName "didimakeanindex")
547-- >>> response <- runBH' $ deleteIndex (IndexName "didimakeanindex")
548-- >>> respIsTwoHunna response
549-- True
550-- >>> runBH' $ indexExists (IndexName "didimakeanindex")
551-- False
552deleteIndex :: MonadBH m => IndexName -> m Reply
553deleteIndex (IndexName indexName) =
554 delete =<< joinPath [indexName]
556-- | 'updateIndexSettings' will apply a non-empty list of setting updates to an index
557--
558-- >>> _ <- runBH' $ createIndex defaultIndexSettings (IndexName "unconfiguredindex")
559-- >>> response <- runBH' $ updateIndexSettings (BlocksWrite False :| []) (IndexName "unconfiguredindex")
560-- >>> respIsTwoHunna response
561-- True
562updateIndexSettings :: MonadBH m => NonEmpty UpdatableIndexSetting -> IndexName -> m Reply
563updateIndexSettings updates (IndexName indexName) =
564 bindM2 put url (return body)
565 where
566 url = joinPath [indexName, "_settings"]
567 body = Just (encode jsonBody)
568 jsonBody = Object (deepMerge [u | Object u <- toJSON <$> toList updates])
571getIndexSettings :: (MonadBH m, MonadThrow m) => IndexName
572 -> m (Either EsError IndexSettingsSummary)
573getIndexSettings (IndexName indexName) =
574 parseEsResponse =<< get =<< url
575 where
576 url = joinPath [indexName, "_settings"]
578-- | 'forceMergeIndex'
579--
580-- The force merge API allows to force merging of one or more indices through
581-- an API. The merge relates to the number of segments a Lucene index holds
582-- within each shard. The force merge operation allows to reduce the number of
583-- segments by merging them.
584--
585-- This call will block until the merge is complete. If the http connection is
586-- lost, the request will continue in the background, and any new requests will
587-- block until the previous force merge is complete.
589-- For more information see
590-- <https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-forcemerge.html#indices-forcemerge>.
591-- Nothing
592-- worthwhile comes back in the reply body, so matching on the status
593-- should suffice.
594--
595-- 'forceMergeIndex' with a maxNumSegments of 1 and onlyExpungeDeletes
596-- to True is the main way to release disk space back to the OS being
597-- held by deleted documents.
598--
599-- >>> let ixn = IndexName "unoptimizedindex"
600-- >>> _ <- runBH' $ deleteIndex ixn >> createIndex defaultIndexSettings ixn
601-- >>> response <- runBH' $ forceMergeIndex (IndexList (ixn :| [])) (defaultIndexOptimizationSettings { maxNumSegments = Just 1, onlyExpungeDeletes = True })
602-- >>> respIsTwoHunna response
603-- True
604forceMergeIndex :: MonadBH m => IndexSelection -> ForceMergeIndexSettings -> m Reply
605forceMergeIndex ixs ForceMergeIndexSettings {..} =
606 bindM2 post url (return body)
607 where url = addQuery params <$> joinPath [indexName, "_forcemerge"]
608 params = catMaybes [ ("max_num_segments",) . Just . showText <$> maxNumSegments
609 , Just ("only_expunge_deletes", Just (boolQP onlyExpungeDeletes))
610 , Just ("flush", Just (boolQP flushAfterOptimize))
611 ]
612 indexName = indexSelectionName ixs
613 body = Nothing
616deepMerge :: [Object] -> Object
617deepMerge = LS.foldl' go mempty
618 where go acc = LS.foldl' go' acc . HM.toList
619 go' acc (k, v) = HM.insertWith merge k v acc
620 merge (Object a) (Object b) = Object (deepMerge [a, b])
621 merge _ b = b
624statusCodeIs :: (Int, Int) -> Reply -> Bool
625statusCodeIs r resp = inRange r $ NHTS.statusCode (responseStatus resp)
627respIsTwoHunna :: Reply -> Bool
628respIsTwoHunna = statusCodeIs (200, 299)
630existentialQuery :: MonadBH m => Text -> m (Reply, Bool)
631existentialQuery url = do
632 reply <- head url
633 return (reply, respIsTwoHunna reply)
636-- | Tries to parse a response body as the expected type @a@ and
637-- failing that tries to parse it as an EsError. All well-formed, JSON
638-- responses from elasticsearch should fall into these two
639-- categories. If they don't, a 'EsProtocolException' will be
640-- thrown. If you encounter this, please report the full body it
641-- reports along with your Elasticsearch verison.
642parseEsResponse :: ( MonadThrow m
643 , FromJSON a
644 )
645 => Reply
646 -> m (Either EsError a)
647parseEsResponse reply
648 | respIsTwoHunna reply = case eitherDecode body of
649 Right a -> return (Right a)
650 Left err ->
651 tryParseError err
652 | otherwise = tryParseError "Non-200 status code"
653 where body = responseBody reply
654 tryParseError originalError
655 = case eitherDecode body of
656 Right e -> return (Left e)
657 -- Failed to parse the error message.
658 Left err -> explode ("Original error was: " <> originalError <> " Error parse failure was: " <> err)
659 explode errorMsg = throwM (EsProtocolException (T.pack errorMsg) body)
661-- | 'indexExists' enables you to check if an index exists. Returns 'Bool'
662-- in IO
663--
664-- >>> exists <- runBH' $ indexExists testIndex
665indexExists :: MonadBH m => IndexName -> m Bool
666indexExists (IndexName indexName) = do
667 (_, exists) <- existentialQuery =<< joinPath [indexName]
668 return exists
670-- | 'refreshIndex' will force a refresh on an index. You must
671-- do this if you want to read what you wrote.
672--
673-- >>> _ <- runBH' $ createIndex defaultIndexSettings testIndex
674-- >>> _ <- runBH' $ refreshIndex testIndex
675refreshIndex :: MonadBH m => IndexName -> m Reply
676refreshIndex (IndexName indexName) =
677 bindM2 post url (return Nothing)
678 where url = joinPath [indexName, "_refresh"]
680-- | Block until the index becomes available for indexing
681-- documents. This is useful for integration tests in which
682-- indices are rapidly created and deleted.
683waitForYellowIndex :: MonadBH m => IndexName -> m Reply
684waitForYellowIndex (IndexName indexName) = get =<< url
685 where url = addQuery q <$> joinPath ["_cluster","health",indexName]
686 q = [("wait_for_status",Just "yellow"),("timeout",Just "10s")]
688stringifyOCIndex :: OpenCloseIndex -> Text
689stringifyOCIndex oci = case oci of
690 OpenIndex -> "_open"
691 CloseIndex -> "_close"
693openOrCloseIndexes :: MonadBH m => OpenCloseIndex -> IndexName -> m Reply
694openOrCloseIndexes oci (IndexName indexName) =
695 bindM2 post url (return Nothing)
696 where ociString = stringifyOCIndex oci
697 url = joinPath [indexName, ociString]
699-- | 'openIndex' opens an index given a 'Server' and an 'IndexName'. Explained in further detail at
700-- <http://www.elasticsearch.org/guide/en/elasticsearch/reference/current/indices-open-close.html>
701--
702-- >>> reply <- runBH' $ openIndex testIndex
703openIndex :: MonadBH m => IndexName -> m Reply
704openIndex = openOrCloseIndexes OpenIndex
706-- | 'closeIndex' closes an index given a 'Server' and an 'IndexName'. Explained in further detail at
707-- <http://www.elasticsearch.org/guide/en/elasticsearch/reference/current/indices-open-close.html>
708--
709-- >>> reply <- runBH' $ closeIndex testIndex
710closeIndex :: MonadBH m => IndexName -> m Reply
711closeIndex = openOrCloseIndexes CloseIndex
713-- | 'listIndices' returns a list of all index names on a given 'Server'
714listIndices :: (MonadThrow m, MonadBH m) => m [IndexName]
715listIndices =
716 parse . responseBody =<< get =<< url
717 where
718 url = joinPath ["_cat/indices?format=json"]
719 parse body = either (\msg -> (throwM (EsProtocolException (T.pack msg) body))) return $ do
720 vals <- eitherDecode body
721 forM vals $ \val ->
722 case val of
723 Object obj ->
724 case HM.lookup "index" obj of
725 (Just (String txt)) -> Right (IndexName txt)
726 v -> Left $ "indexVal in listIndices failed on non-string, was: " <> show v
727 v -> Left $ "One of the values parsed in listIndices wasn't an object, it was: " <> show v
729-- | 'updateIndexAliases' updates the server's index alias
730-- table. Operations are atomic. Explained in further detail at
731-- <https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-aliases.html>
732--
733-- >>> let src = IndexName "a-real-index"
734-- >>> let aliasName = IndexName "an-alias"
735-- >>> let iAlias = IndexAlias src (IndexAliasName aliasName)
736-- >>> let aliasCreate = IndexAliasCreate Nothing Nothing
737-- >>> _ <- runBH' $ deleteIndex src
738-- >>> respIsTwoHunna <$> runBH' (createIndex defaultIndexSettings src)
739-- True
740-- >>> runBH' $ indexExists src
741-- True
742-- >>> respIsTwoHunna <$> runBH' (updateIndexAliases (AddAlias iAlias aliasCreate :| []))
743-- True
744-- >>> runBH' $ indexExists aliasName
745-- True
746updateIndexAliases :: MonadBH m => NonEmpty IndexAliasAction -> m Reply
747updateIndexAliases actions = bindM2 post url (return body)
748 where url = joinPath ["_aliases"]
749 body = Just (encode bodyJSON)
750 bodyJSON = object [ "actions" .= toList actions]
752-- | Get all aliases configured on the server.
753getIndexAliases :: (MonadBH m, MonadThrow m)
754 => m (Either EsError IndexAliasesSummary)
755getIndexAliases = parseEsResponse =<< get =<< url
756 where url = joinPath ["_aliases"]
758-- | Delete a single alias, removing it from all indices it
759-- is currently associated with.
760deleteIndexAlias :: MonadBH m => IndexAliasName -> m Reply
761deleteIndexAlias (IndexAliasName (IndexName name)) = delete =<< url
762 where url = joinPath ["_all","_alias",name]
764-- | 'putTemplate' creates a template given an 'IndexTemplate' and a 'TemplateName'.
765-- Explained in further detail at
766-- <https://www.elastic.co/guide/en/elasticsearch/reference/1.7/indices-templates.html>
767--
768-- >>> let idxTpl = IndexTemplate (TemplatePattern "tweet-*") (Just (IndexSettings (ShardCount 1) (ReplicaCount 1))) [toJSON TweetMapping]
769-- >>> resp <- runBH' $ putTemplate idxTpl (TemplateName "tweet-tpl")
770putTemplate :: MonadBH m => IndexTemplate -> TemplateName -> m Reply
771putTemplate indexTemplate (TemplateName templateName) =
772 bindM2 put url (return body)
773 where url = joinPath ["_template", templateName]
774 body = Just $ encode indexTemplate
776-- | 'templateExists' checks to see if a template exists.
777--
778-- >>> exists <- runBH' $ templateExists (TemplateName "tweet-tpl")
779templateExists :: MonadBH m => TemplateName -> m Bool
780templateExists (TemplateName templateName) = do
781 (_, exists) <- existentialQuery =<< joinPath ["_template", templateName]
782 return exists
784-- | 'deleteTemplate' is an HTTP DELETE and deletes a template.
785--
786-- >>> let idxTpl = IndexTemplate (TemplatePattern "tweet-*") (Just (IndexSettings (ShardCount 1) (ReplicaCount 1))) [toJSON TweetMapping]
787-- >>> _ <- runBH' $ putTemplate idxTpl (TemplateName "tweet-tpl")
788-- >>> resp <- runBH' $ deleteTemplate (TemplateName "tweet-tpl")
789deleteTemplate :: MonadBH m => TemplateName -> m Reply
790deleteTemplate (TemplateName templateName) =
791 delete =<< joinPath ["_template", templateName]
793-- | 'putMapping' is an HTTP PUT and has upsert semantics. Mappings are schemas
794-- for documents in indexes.
795--
796-- >>> _ <- runBH' $ createIndex defaultIndexSettings testIndex
797-- >>> resp <- runBH' $ putMapping testIndex testMapping TweetMapping
798-- >>> print resp
799-- Response {responseStatus = Status {statusCode = 200, statusMessage = "OK"}, responseVersion = HTTP/1.1, responseHeaders = [("content-type","application/json; charset=UTF-8"),("content-encoding","gzip"),("transfer-encoding","chunked")], responseBody = "{\"acknowledged\":true}", responseCookieJar = CJ {expose = []}, responseClose' = ResponseClose}
800putMapping :: (MonadBH m, ToJSON a) => IndexName
801 -> MappingName -> a -> m Reply
802putMapping (IndexName indexName) (MappingName mappingName) mapping =
803 bindM2 put url (return body)
804 where url = joinPath [indexName, "_mapping", mappingName]
805 -- "_mapping" and mappingName above were originally transposed
806 -- erroneously. The correct API call is: "/INDEX/_mapping/MAPPING_NAME"
807 body = Just $ encode mapping
809versionCtlParams :: IndexDocumentSettings -> [(Text, Maybe Text)]
810versionCtlParams cfg =
811 case idsVersionControl cfg of
812 NoVersionControl -> []
813 InternalVersion v -> versionParams v "internal"
814 ExternalGT (ExternalDocVersion v) -> versionParams v "external_gt"
815 ExternalGTE (ExternalDocVersion v) -> versionParams v "external_gte"
816 ForceVersion (ExternalDocVersion v) -> versionParams v "force"
817 where
818 vt = showText . docVersionNumber
819 versionParams v t = [ ("version", Just $ vt v)
820 , ("version_type", Just t)
821 ]
823-- | 'indexDocument' is the primary way to save a single document in
824-- Elasticsearch. The document itself is simply something we can
825-- convert into a JSON 'Value'. The 'DocId' will function as the
826-- primary key for the document. You are encouraged to generate
827-- your own id's and not rely on Elasticsearch's automatic id
828-- generation. Read more about it here:
829-- https://github.com/bitemyapp/bloodhound/issues/107
830--
831-- >>> resp <- runBH' $ indexDocument testIndex testMapping defaultIndexDocumentSettings exampleTweet (DocId "1")
832-- >>> print resp
833-- Response {responseStatus = Status {statusCode = 201, statusMessage = "Created"}, responseVersion = HTTP/1.1, responseHeaders = [("Location","/twitter/tweet/1"),("content-type","application/json; charset=UTF-8"),("content-encoding","gzip"),("transfer-encoding","chunked")], responseBody = "{\"_index\":\"twitter\",\"_type\":\"tweet\",\"_id\":\"1\",\"_version\":1,\"result\":\"created\",\"_shards\":{\"total\":2,\"successful\":1,\"failed\":0},\"created\":true}", responseCookieJar = CJ {expose = []}, responseClose' = ResponseClose}
834indexDocument :: (ToJSON doc, MonadBH m) => IndexName -> MappingName
835 -> IndexDocumentSettings -> doc -> DocId -> m Reply
836indexDocument (IndexName indexName)
837 (MappingName mappingName) cfg document (DocId docId) =
838 bindM2 put url (return body)
839 where url = addQuery params <$> joinPath [indexName, mappingName, docId]
840 parentParams = case idsParent cfg of
841 Nothing -> []
842 Just (DocumentParent (DocId p)) -> [ ("parent", Just p) ]
843 params = versionCtlParams cfg ++ parentParams
844 body = Just (encode document)
846-- | 'updateDocument' provides a way to perform an partial update of a
847-- an already indexed document.
848updateDocument :: (ToJSON patch, MonadBH m) => IndexName -> MappingName
849 -> IndexDocumentSettings -> patch -> DocId -> m Reply
850updateDocument (IndexName indexName)
851 (MappingName mappingName) cfg patch (DocId docId) =
852 bindM2 post url (return body)
853 where url = addQuery (versionCtlParams cfg) <$>
854 joinPath [indexName, mappingName, docId, "_update"]
855 body = Just (encode $ object ["doc" .= toJSON patch])
857-- | 'deleteDocument' is the primary way to delete a single document.
858--
859-- >>> _ <- runBH' $ deleteDocument testIndex testMapping (DocId "1")
860deleteDocument :: MonadBH m => IndexName -> MappingName
861 -> DocId -> m Reply
862deleteDocument (IndexName indexName)
863 (MappingName mappingName) (DocId docId) =
864 delete =<< joinPath [indexName, mappingName, docId]
866-- | 'bulk' uses
867-- <http://www.elasticsearch.org/guide/en/elasticsearch/reference/current/docs-bulk.html Elasticsearch's bulk API>
868-- to perform bulk operations. The 'BulkOperation' data type encodes the
869-- index\/update\/delete\/create operations. You pass a 'V.Vector' of 'BulkOperation's
870-- and a 'Server' to 'bulk' in order to send those operations up to your Elasticsearch
871-- server to be performed. I changed from [BulkOperation] to a Vector due to memory overhead.
872--
873-- >>> let stream = V.fromList [BulkIndex testIndex testMapping (DocId "2") (toJSON (BulkTest "blah"))]
874-- >>> _ <- runBH' $ bulk stream
875-- >>> _ <- runBH' $ refreshIndex testIndex
876bulk :: MonadBH m => V.Vector BulkOperation -> m Reply
877bulk bulkOps =
878 bindM2 post url (return body)
879 where
880 url = joinPath ["_bulk"]
881 body = Just $ encodeBulkOperations bulkOps
883-- | 'encodeBulkOperations' is a convenience function for dumping a vector of 'BulkOperation'
884-- into an 'L.ByteString'
885--
886-- >>> let bulkOps = V.fromList [BulkIndex testIndex testMapping (DocId "2") (toJSON (BulkTest "blah"))]
887-- >>> encodeBulkOperations bulkOps
888-- "\n{\"index\":{\"_type\":\"tweet\",\"_id\":\"2\",\"_index\":\"twitter\"}}\n{\"name\":\"blah\"}\n"
889encodeBulkOperations :: V.Vector BulkOperation -> L.ByteString
890encodeBulkOperations stream = collapsed where
891 blobs =
892 fmap encodeBulkOperation stream
893 mashedTaters =
894 mash (mempty :: Builder) blobs
895 collapsed =
896 toLazyByteString $ mappend mashedTaters (byteString "\n")
898mash :: Builder -> V.Vector L.ByteString -> Builder
899mash = V.foldl' (\b x -> b <> byteString "\n" <> lazyByteString x)
901mkBulkStreamValue :: Text -> Text -> Text -> Text -> Value
902mkBulkStreamValue operation indexName mappingName docId =
903 object [operation .=
904 object [ "_index" .= indexName
905 , "_type" .= mappingName
906 , "_id" .= docId]]
908mkBulkStreamValueAuto :: Text -> Text -> Text -> Value
909mkBulkStreamValueAuto operation indexName mappingName =
910 object [operation .=
911 object [ "_index" .= indexName
912 , "_type" .= mappingName]]
914-- | 'encodeBulkOperation' is a convenience function for dumping a single 'BulkOperation'
915-- into an 'L.ByteString'
916--
917-- >>> let bulkOp = BulkIndex testIndex testMapping (DocId "2") (toJSON (BulkTest "blah"))
918-- >>> encodeBulkOperation bulkOp
919-- "{\"index\":{\"_type\":\"tweet\",\"_id\":\"2\",\"_index\":\"twitter\"}}\n{\"name\":\"blah\"}"
920encodeBulkOperation :: BulkOperation -> L.ByteString
921encodeBulkOperation (BulkIndex (IndexName indexName)
922 (MappingName mappingName)
923 (DocId docId) value) = blob
924 where metadata = mkBulkStreamValue "index" indexName mappingName docId
925 blob = encode metadata `mappend` "\n" `mappend` encode value
927encodeBulkOperation (BulkIndexAuto (IndexName indexName)
928 (MappingName mappingName)
929 value) = blob
930 where metadata = mkBulkStreamValueAuto "index" indexName mappingName
931 blob = encode metadata `mappend` "\n" `mappend` encode value
933encodeBulkOperation (BulkIndexEncodingAuto (IndexName indexName)
934 (MappingName mappingName)
935 encoding) = toLazyByteString blob
936 where metadata = toEncoding (mkBulkStreamValueAuto "index" indexName mappingName)
937 blob = fromEncoding metadata <> "\n" <> fromEncoding encoding
939encodeBulkOperation (BulkCreate (IndexName indexName)
940 (MappingName mappingName)
941 (DocId docId) value) = blob
942 where metadata = mkBulkStreamValue "create" indexName mappingName docId
943 blob = encode metadata `mappend` "\n" `mappend` encode value
945encodeBulkOperation (BulkDelete (IndexName indexName)
946 (MappingName mappingName)
947 (DocId docId)) = blob
948 where metadata = mkBulkStreamValue "delete" indexName mappingName docId
949 blob = encode metadata
951encodeBulkOperation (BulkUpdate (IndexName indexName)
952 (MappingName mappingName)
953 (DocId docId) value) = blob
954 where metadata = mkBulkStreamValue "update" indexName mappingName docId
955 doc = object ["doc" .= value]
956 blob = encode metadata `mappend` "\n" `mappend` encode doc
958encodeBulkOperation (BulkCreateEncoding (IndexName indexName)
959 (MappingName mappingName)
960 (DocId docId) encoding) = toLazyByteString blob
961 where metadata = toEncoding (mkBulkStreamValue "create" indexName mappingName docId)
962 blob = fromEncoding metadata <> "\n" <> fromEncoding encoding
964-- | 'getDocument' is a straight-forward way to fetch a single document from
965-- Elasticsearch using a 'Server', 'IndexName', 'MappingName', and a 'DocId'.
966-- The 'DocId' is the primary key for your Elasticsearch document.
967--
968-- >>> yourDoc <- runBH' $ getDocument testIndex testMapping (DocId "1")
969getDocument :: MonadBH m => IndexName -> MappingName
970 -> DocId -> m Reply
971getDocument (IndexName indexName)
972 (MappingName mappingName) (DocId docId) =
973 get =<< joinPath [indexName, mappingName, docId]
975-- | 'documentExists' enables you to check if a document exists. Returns 'Bool'
976-- in IO
977--
978-- >>> exists <- runBH' $ documentExists testIndex testMapping Nothing (DocId "1")
979documentExists :: MonadBH m => IndexName -> MappingName
980 -> Maybe DocumentParent -> DocId -> m Bool
981documentExists (IndexName indexName) (MappingName mappingName)
982 parent (DocId docId) = do
983 (_, exists) <- existentialQuery =<< url
984 return exists
985 where url = addQuery params <$> joinPath [indexName, mappingName, docId]
986 parentParam = fmap (\(DocumentParent (DocId p)) -> p) parent
987 params = LS.filter (\(_, v) -> isJust v) [("parent", parentParam)]
989dispatchSearch :: MonadBH m => Text -> Search -> m Reply
990dispatchSearch url search = post url' (Just (encode search))
991 where url' = appendSearchTypeParam url (searchType search)
993-- | 'searchAll', given a 'Search', will perform that search against all indexes
994-- on an Elasticsearch server. Try to avoid doing this if it can be helped.
995--
996-- >>> let query = TermQuery (Term "user" "bitemyapp") Nothing
997-- >>> let search = mkSearch (Just query) Nothing
998-- >>> reply <- runBH' $ searchAll search
999searchAll :: MonadBH m => Search -> m Reply
1000searchAll = bindM2 dispatchSearch url . return
1001 where url = joinPath ["_search"]
1003-- | 'searchByIndex', given a 'Search' and an 'IndexName', will perform that search
1004-- against all mappings within an index on an Elasticsearch server.
1006-- >>> let query = TermQuery (Term "user" "bitemyapp") Nothing
1007-- >>> let search = mkSearch (Just query) Nothing
1008-- >>> reply <- runBH' $ searchByIndex testIndex search
1009searchByIndex :: MonadBH m => IndexName -> Search -> m Reply
1010searchByIndex (IndexName indexName) = bindM2 dispatchSearch url . return
1011 where url = joinPath [indexName, "_search"]
1013-- | 'searchByIndices' is a variant of 'searchByIndex' that executes a
1014-- 'Search' over many indices. This is much faster than using
1015-- 'mapM' to 'searchByIndex' over a collection since it only
1016-- causes a single HTTP request to be emitted.
1017searchByIndices :: MonadBH m => NonEmpty IndexName -> Search -> m Reply
1018searchByIndices ixs = bindM2 dispatchSearch url . return
1019 where url = joinPath [renderedIxs, "_search"]
1020 renderedIxs = T.intercalate (T.singleton ',') (map (\(IndexName t) -> t) (toList ixs))
1022-- | 'searchByType', given a 'Search', 'IndexName', and 'MappingName', will perform that
1023-- search against a specific mapping within an index on an Elasticsearch server.
1025-- >>> let query = TermQuery (Term "user" "bitemyapp") Nothing
1026-- >>> let search = mkSearch (Just query) Nothing
1027-- >>> reply <- runBH' $ searchByType testIndex testMapping search
1028searchByType :: MonadBH m => IndexName -> MappingName -> Search
1029 -> m Reply
1030searchByType (IndexName indexName)
1031 (MappingName mappingName) = bindM2 dispatchSearch url . return
1032 where url = joinPath [indexName, mappingName, "_search"]
1034-- | For a given search, request a scroll for efficient streaming of
1035-- search results. Note that the search is put into 'SearchTypeScan'
1036-- mode and thus results will not be sorted. Combine this with
1037-- 'advanceScroll' to efficiently stream through the full result set
1038getInitialScroll ::
1039 (FromJSON a, MonadThrow m, MonadBH m) => IndexName ->
1040 MappingName ->
1041 Search ->
1042 m (Either EsError (SearchResult a))
1043getInitialScroll (IndexName indexName) (MappingName mappingName) search' = do
1044 let url = addQuery params <$> joinPath [indexName, mappingName, "_search"]
1045 params = [("scroll", Just "1m")]
1046 sorting = Just [DefaultSortSpec $ mkSort (FieldName "_doc") Descending]
1047 search = search' { sortBody = sorting }
1048 resp' <- bindM2 dispatchSearch url (return search)
1049 parseEsResponse resp'
1051-- | For a given search, request a scroll for efficient streaming of
1052-- search results. Combine this with 'advanceScroll' to efficiently
1053-- stream through the full result set. Note that this search respects
1054-- sorting and may be less efficient than 'getInitialScroll'.
1055getInitialSortedScroll ::
1056 (FromJSON a, MonadThrow m, MonadBH m) => IndexName ->
1057 MappingName ->
1058 Search ->
1059 m (Either EsError (SearchResult a))
1060getInitialSortedScroll (IndexName indexName) (MappingName mappingName) search = do
1061 let url = addQuery params <$> joinPath [indexName, mappingName, "_search"]
1062 params = [("scroll", Just "1m")]
1063 resp' <- bindM2 dispatchSearch url (return search)
1064 parseEsResponse resp'
1066scroll' :: (FromJSON a, MonadBH m, MonadThrow m) => Maybe ScrollId ->
1067 m ([Hit a], Maybe ScrollId)
1068scroll' Nothing = return ([], Nothing)
1069scroll' (Just sid) = do
1070 res <- advanceScroll sid 60
1071 case res of
1072 Right SearchResult {..} -> return (hits searchHits, scrollId)
1073 Left _ -> return ([], Nothing)
1075-- | Use the given scroll to fetch the next page of documents. If there are no
1076-- further pages, 'SearchResult.searchHits.hits' will be '[]'.
1077advanceScroll
1078 :: ( FromJSON a
1079 , MonadBH m
1080 , MonadThrow m
1082 => ScrollId
1083 -> NominalDiffTime
1084 -- ^ How long should the snapshot of data be kept around? This timeout is updated every time 'advanceScroll' is used, so don't feel the need to set it to the entire duration of your search processing. Note that durations < 1s will be rounded up. Also note that 'NominalDiffTime' is an instance of Num so literals like 60 will be interpreted as seconds. 60s is a reasonable default.
1085 -> m (Either EsError (SearchResult a))
1086advanceScroll (ScrollId sid) scroll = do
1087 url <- joinPath ["_search", "scroll"]
1088 resp <- post url (Just $ encode scrollObject)
1089 parseEsResponse resp
1090 where scrollTime = showText secs <> "s"
1091 secs :: Integer
1092 secs = round scroll
1094 scrollObject = object [ "scroll" .= scrollTime
1095 , "scroll_id" .= sid
1098simpleAccumulator ::
1099 (FromJSON a, MonadBH m, MonadThrow m) =>
1100 [Hit a] ->
1101 ([Hit a], Maybe ScrollId) ->
1102 m ([Hit a], Maybe ScrollId)
1103simpleAccumulator oldHits (newHits, Nothing) = return (oldHits ++ newHits, Nothing)
1104simpleAccumulator oldHits ([], _) = return (oldHits, Nothing)
1105simpleAccumulator oldHits (newHits, msid) = do
1106 (newHits', msid') <- scroll' msid
1107 simpleAccumulator (oldHits ++ newHits) (newHits', msid')
1109-- | 'scanSearch' uses the 'scroll' API of elastic,
1110-- for a given 'IndexName' and 'MappingName'. Note that this will
1111-- consume the entire search result set and will be doing O(n) list
1112-- appends so this may not be suitable for large result sets. In that
1113-- case, 'getInitialScroll' and 'advanceScroll' are good low level
1114-- tools. You should be able to hook them up trivially to conduit,
1115-- pipes, or your favorite streaming IO abstraction of choice. Note
1116-- that ordering on the search would destroy performance and thus is
1117-- ignored.
1118scanSearch :: (FromJSON a, MonadBH m, MonadThrow m) => IndexName
1119 -> MappingName
1120 -> Search
1121 -> m [Hit a]
1122scanSearch indexName mappingName search = do
1123 initialSearchResult <- getInitialScroll indexName mappingName search
1124 let (hits', josh) = case initialSearchResult of
1125 Right SearchResult {..} -> (hits searchHits, scrollId)
1126 Left _ -> ([], Nothing)
1127 (totalHits, _) <- simpleAccumulator [] (hits', josh)
1128 return totalHits
1130-- | 'mkSearch' is a helper function for defaulting additional fields of a 'Search'
1131-- to Nothing in case you only care about your 'Query' and 'Filter'. Use record update
1132-- syntax if you want to add things like aggregations or highlights while still using
1133-- this helper function.
1135-- >>> let query = TermQuery (Term "user" "bitemyapp") Nothing
1136-- >>> mkSearch (Just query) Nothing
1137-- Search {queryBody = Just (TermQuery (Term {termField = "user", termValue = "bitemyapp"}) Nothing), filterBody = Nothing, searchAfterKey = Nothing, sortBody = Nothing, aggBody = Nothing, highlight = Nothing, trackSortScores = False, from = From 0, size = Size 10, searchType = SearchTypeQueryThenFetch, fields = Nothing, source = Nothing}
1138mkSearch :: Maybe Query -> Maybe Filter -> Search
1139mkSearch query filter = Search query filter Nothing Nothing Nothing False (From 0) (Size 10) SearchTypeQueryThenFetch Nothing Nothing Nothing Nothing Nothing
1141-- | 'mkAggregateSearch' is a helper function that defaults everything in a 'Search' except for
1142-- the 'Query' and the 'Aggregation'.
1144-- >>> let terms = TermsAgg $ (mkTermsAggregation "user") { termCollectMode = Just BreadthFirst }
1145-- >>> terms
1146-- TermsAgg (TermsAggregation {term = Left "user", termInclude = Nothing, termExclude = Nothing, termOrder = Nothing, termMinDocCount = Nothing, termSize = Nothing, termShardSize = Nothing, termCollectMode = Just BreadthFirst, termExecutionHint = Nothing, termAggs = Nothing})
1147-- >>> let myAggregation = mkAggregateSearch Nothing $ mkAggregations "users" terms
1148mkAggregateSearch :: Maybe Query -> Aggregations -> Search
1149mkAggregateSearch query mkSearchAggs = Search query Nothing Nothing (Just mkSearchAggs) Nothing False (From 0) (Size 0) SearchTypeQueryThenFetch Nothing Nothing Nothing Nothing Nothing
1151-- | 'mkHighlightSearch' is a helper function that defaults everything in a 'Search' except for
1152-- the 'Query' and the 'Aggregation'.
1154-- >>> let query = QueryMatchQuery $ mkMatchQuery (FieldName "_all") (QueryString "haskell")
1155-- >>> let testHighlight = Highlights Nothing [FieldHighlight (FieldName "message") Nothing]
1156-- >>> let search = mkHighlightSearch (Just query) testHighlight
1157mkHighlightSearch :: Maybe Query -> Highlights -> Search
1158mkHighlightSearch query searchHighlights = Search query Nothing Nothing Nothing (Just searchHighlights) False (From 0) (Size 10) SearchTypeQueryThenFetch Nothing Nothing Nothing Nothing Nothing
1160-- | 'pageSearch' is a helper function that takes a search and assigns the from
1161-- and size fields for the search. The from parameter defines the offset
1162-- from the first result you want to fetch. The size parameter allows you to
1163-- configure the maximum amount of hits to be returned.
1165-- >>> let query = QueryMatchQuery $ mkMatchQuery (FieldName "_all") (QueryString "haskell")
1166-- >>> let search = mkSearch (Just query) Nothing
1167-- >>> search
1168-- Search {queryBody = Just (QueryMatchQuery (MatchQuery {matchQueryField = FieldName "_all", matchQueryQueryString = QueryString "haskell", matchQueryOperator = Or, matchQueryZeroTerms = ZeroTermsNone, matchQueryCutoffFrequency = Nothing, matchQueryMatchType = Nothing, matchQueryAnalyzer = Nothing, matchQueryMaxExpansions = Nothing, matchQueryLenient = Nothing, matchQueryBoost = Nothing})), filterBody = Nothing, sortBody = Nothing, aggBody = Nothing, highlight = Nothing, trackSortScores = False, from = From 0, size = Size 10, searchType = SearchTypeQueryThenFetch, fields = Nothing, source = Nothing}
1169-- >>> pageSearch (From 10) (Size 100) search
1170-- Search {queryBody = Just (QueryMatchQuery (MatchQuery {matchQueryField = FieldName "_all", matchQueryQueryString = QueryString "haskell", matchQueryOperator = Or, matchQueryZeroTerms = ZeroTermsNone, matchQueryCutoffFrequency = Nothing, matchQueryMatchType = Nothing, matchQueryAnalyzer = Nothing, matchQueryMaxExpansions = Nothing, matchQueryLenient = Nothing, matchQueryBoost = Nothing})), filterBody = Nothing, sortBody = Nothing, aggBody = Nothing, highlight = Nothing, trackSortScores = False, from = From 10, size = Size 100, searchType = SearchTypeQueryThenFetch, fields = Nothing, source = Nothing}
1171pageSearch :: From -- ^ The result offset
1172 -> Size -- ^ The number of results to return
1173 -> Search -- ^ The current seach
1174 -> Search -- ^ The paged search
1175pageSearch resultOffset pageSize search = search { from = resultOffset, size = pageSize }
1177parseUrl' :: MonadThrow m => Text -> m Request
1178parseUrl' t = parseRequest (URI.escapeURIString URI.isAllowedInURI (T.unpack t))
1180-- | Was there an optimistic concurrency control conflict when
1181-- indexing a document?
1182isVersionConflict :: Reply -> Bool
1183isVersionConflict = statusCheck (== 409)
1185isSuccess :: Reply -> Bool
1186isSuccess = statusCheck (inRange (200, 299))
1188isCreated :: Reply -> Bool
1189isCreated = statusCheck (== 201)
1191statusCheck :: (Int -> Bool) -> Reply -> Bool
1192statusCheck prd = prd . NHTS.statusCode . responseStatus
1194-- | This is a hook that can be set via the 'bhRequestHook' function
1195-- that will authenticate all requests using an HTTP Basic
1196-- Authentication header. Note that it is *strongly* recommended that
1197-- this option only be used over an SSL connection.
1199-- >> (mkBHEnv myServer myManager) { bhRequestHook = basicAuthHook (EsUsername "myuser") (EsPassword "mypass") }
1200basicAuthHook :: Monad m => EsUsername -> EsPassword -> Request -> m Request
1201basicAuthHook (EsUsername u) (EsPassword p) = return . applyBasicAuth u' p'
1202 where u' = T.encodeUtf8 u
1203 p' = T.encodeUtf8 p
1206boolQP :: Bool -> Text
1207boolQP True = "true"
1208boolQP False = "false"