git.iko.soy/typeable/bloodhound
on 86-sub-aggregation-parsing help sign in
typeable/bloodhound/src/Database/Bloodhound/Client.hs
842 lines 40 kB history raw
1{-# LANGUAGE OverloadedStrings #-}
2{-# LANGUAGE RecordWildCards #-}
3{-# LANGUAGE TupleSections #-}
5-------------------------------------------------------------------------------
6-- |
7-- Module : Database.Bloodhound.Client
8-- Copyright : (C) 2014 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 , createIndex
27 , deleteIndex
28 , updateIndexSettings
29 , getIndexSettings
30 , optimizeIndex
31 , indexExists
32 , openIndex
33 , closeIndex
34 , updateIndexAliases
35 , getIndexAliases
36 , putTemplate
37 , templateExists
38 , deleteTemplate
39 , putMapping
40 , deleteMapping
41 , indexDocument
42 , updateDocument
43 , getDocument
44 , documentExists
45 , deleteDocument
46 , searchAll
47 , searchByIndex
48 , searchByType
49 , scanSearch
50 , getInitialScroll
51 , advanceScroll
52 , refreshIndex
53 , mkSearch
54 , mkAggregateSearch
55 , mkHighlightSearch
56 , bulk
57 , pageSearch
58 , mkShardCount
59 , mkReplicaCount
60 , getStatus
61 , encodeBulkOperations
62 , encodeBulkOperation
63 -- * Authentication
64 , basicAuthHook
65 -- * Reply-handling tools
66 , isVersionConflict
67 , isSuccess
68 , isCreated
69 , parseEsResponse
70 )
71 where
73import qualified Blaze.ByteString.Builder as BB
74import Control.Applicative
75import Control.Monad
76import Control.Monad.Catch
77import Control.Monad.IO.Class
78import Data.Aeson
79import Data.ByteString.Lazy.Builder
80import qualified Data.ByteString.Lazy.Char8 as L
81import Data.Foldable (toList)
82import qualified Data.HashMap.Strict as HM
83import Data.Ix
84import qualified Data.List as LS (filter, foldl')
85import Data.List.NonEmpty (NonEmpty (..))
86import Data.Maybe (catMaybes, fromMaybe, isJust)
87import Data.Monoid
88import Data.Text (Text)
89import qualified Data.Text as T
90import qualified Data.Text.Encoding as T
91import Data.Time.Clock
92import qualified Data.Vector as V
93import Network.HTTP.Client
94import qualified Network.HTTP.Types.Method as NHTM
95import qualified Network.HTTP.Types.Status as NHTS
96import qualified Network.HTTP.Types.URI as NHTU
97import qualified Network.URI as URI
98import Prelude hiding (filter, head)
100import Database.Bloodhound.Types
102-- $setup
103-- >>> :set -XOverloadedStrings
104-- >>> :set -XDeriveGeneric
105-- >>> import Database.Bloodhound
106-- >>> import Test.DocTest.Prop (assert)
107-- >>> let testServer = (Server "http://localhost:9200")
108-- >>> let runBH' = withBH defaultManagerSettings testServer
109-- >>> let testIndex = IndexName "twitter"
110-- >>> let testMapping = MappingName "tweet"
111-- >>> let defaultIndexSettings = IndexSettings (ShardCount 1) (ReplicaCount 0)
112-- >>> data TweetMapping = TweetMapping deriving (Eq, Show)
113-- >>> _ <- runBH' $ deleteIndex testIndex >> deleteMapping testIndex testMapping
114-- >>> import GHC.Generics
115-- >>> import Data.Time.Calendar (Day (..))
116-- >>> import Data.Time.Clock (UTCTime (..), secondsToDiffTime)
117-- >>> :{
118--instance ToJSON TweetMapping where
119-- toJSON TweetMapping =
120-- object ["tweet" .=
121-- object ["properties" .=
122-- object ["location" .=
123-- object ["type" .= ("geo_point" :: Text)]]]]
124--data Location = Location { lat :: Double
125-- , lon :: Double } deriving (Eq, Generic, Show)
126--data Tweet = Tweet { user :: Text
127-- , postDate :: UTCTime
128-- , message :: Text
129-- , age :: Int
130-- , location :: Location } deriving (Eq, Generic, Show)
131--exampleTweet = Tweet { user = "bitemyapp"
132-- , postDate = UTCTime
133-- (ModifiedJulianDay 55000)
134-- (secondsToDiffTime 10)
135-- , message = "Use haskell!"
136-- , age = 10000
137-- , location = Location 40.12 (-71.34) }
138--instance ToJSON Tweet where
139-- toJSON = genericToJSON defaultOptions
140--instance FromJSON Tweet where
141-- parseJSON = genericParseJSON defaultOptions
142--instance ToJSON Location where
143-- toJSON = genericToJSON defaultOptions
144--instance FromJSON Location where
145-- parseJSON = genericParseJSON defaultOptions
146--data BulkTest = BulkTest { name :: Text } deriving (Eq, Generic, Show)
147--instance FromJSON BulkTest where
148-- parseJSON = genericParseJSON defaultOptions
149--instance ToJSON BulkTest where
150-- toJSON = genericToJSON defaultOptions
151-- :}
153-- | 'mkShardCount' is a straight-forward smart constructor for 'ShardCount'
154-- which rejects 'Int' values below 1 and above 1000.
155--
156-- >>> mkShardCount 10
157-- Just (ShardCount 10)
158mkShardCount :: Int -> Maybe ShardCount
159mkShardCount n
160 | n < 1 = Nothing
161 | n > 1000 = Nothing
162 | otherwise = Just (ShardCount n)
164-- | 'mkReplicaCount' is a straight-forward smart constructor for 'ReplicaCount'
165-- which rejects 'Int' values below 1 and above 1000.
166--
167-- >>> mkReplicaCount 10
168-- Just (ReplicaCount 10)
169mkReplicaCount :: Int -> Maybe ReplicaCount
170mkReplicaCount n
171 | n < 1 = Nothing
172 | n > 1000 = Nothing -- ...
173 | otherwise = Just (ReplicaCount n)
175emptyBody :: L.ByteString
176emptyBody = L.pack ""
178dispatch :: MonadBH m => Method -> Text -> Maybe L.ByteString
179 -> m Reply
180dispatch dMethod url body = do
181 initReq <- liftIO $ parseUrl' url
182 reqHook <- bhRequestHook <$> getBHEnv
183 let reqBody = RequestBodyLBS $ fromMaybe emptyBody body
184 req <- liftIO $ reqHook $ initReq { method = dMethod
185 , requestBody = reqBody
186 , checkStatus = \_ _ _ -> Nothing}
187 mgr <- bhManager <$> getBHEnv
188 liftIO $ httpLbs req mgr
190joinPath' :: [Text] -> Text
191joinPath' = T.intercalate "/"
193joinPath :: MonadBH m => [Text] -> m Text
194joinPath ps = do
195 Server s <- bhServer <$> getBHEnv
196 return $ joinPath' (s:ps)
198appendSearchTypeParam :: Text -> SearchType -> Text
199appendSearchTypeParam originalUrl st = addQuery params originalUrl
200 where stText = "search_type"
201 params
202 | st == SearchTypeDfsQueryThenFetch = [(stText, Just "dfs_query_then_fetch")]
203 | st == SearchTypeCount = [(stText, Just "count")]
204 | st == SearchTypeScan = [(stText, Just "scan"), ("scroll", Just "1m")]
205 | st == SearchTypeQueryAndFetch = [(stText, Just "query_and_fetch")]
206 | st == SearchTypeDfsQueryAndFetch = [(stText, Just "dfs_query_and_fetch")]
207 -- used to catch 'SearchTypeQueryThenFetch', which is also the default
208 | otherwise = [(stText, Just "query_then_fetch")]
210-- | Severely dumbed down query renderer. Assumes your data doesn't
211-- need any encoding
212addQuery :: [(Text, Maybe Text)] -> Text -> Text
213addQuery q u = u <> rendered
214 where
215 rendered =
216 T.decodeUtf8 $ BB.toByteString $ NHTU.renderQueryText prependQuestionMark q
217 prependQuestionMark = True
219bindM2 :: (Applicative m, Monad m) => (a -> b -> m c) -> m a -> m b -> m c
220bindM2 f ma mb = join (f <$> ma <*> mb)
222-- | Convenience function that sets up a manager and BHEnv and runs
223-- the given set of bloodhound operations. Connections will be
224-- pipelined automatically in accordance with the given manager
225-- settings in IO. If you've got your own monad transformer stack, you
226-- should use 'runBH' directly.
227withBH :: ManagerSettings -> Server -> BH IO a -> IO a
228withBH ms s f = do
229 mgr <- newManager ms
230 let env = mkBHEnv s mgr
231 runBH env f
233-- Shortcut functions for HTTP methods
234delete :: MonadBH m => Text -> m Reply
235delete = flip (dispatch NHTM.methodDelete) Nothing
236get :: MonadBH m => Text -> m Reply
237get = flip (dispatch NHTM.methodGet) Nothing
238head :: MonadBH m => Text -> m Reply
239head = flip (dispatch NHTM.methodHead) Nothing
240put :: MonadBH m => Text -> Maybe L.ByteString -> m Reply
241put = dispatch NHTM.methodPut
242post :: MonadBH m => Text -> Maybe L.ByteString -> m Reply
243post = dispatch NHTM.methodPost
245-- indexDocument s ix name doc = put (root </> s </> ix </> name </> doc) (Just encode doc)
246-- http://hackage.haskell.org/package/http-client-lens-0.1.0/docs/Network-HTTP-Client-Lens.html
247-- https://github.com/supki/libjenkins/blob/master/src/Jenkins/Rest/Internal.hs
249-- | 'getStatus' fetches the 'Status' of a 'Server'
250--
251-- >>> serverStatus <- runBH' getStatus
252-- >>> fmap status (serverStatus)
253-- Just 200
254getStatus :: MonadBH m => m (Maybe Status)
255getStatus = do
256 response <- get =<< url
257 return $ decode (responseBody response)
258 where url = joinPath []
260-- | 'createIndex' will create an index given a 'Server', 'IndexSettings', and an 'IndexName'.
261--
262-- >>> response <- runBH' $ createIndex defaultIndexSettings (IndexName "didimakeanindex")
263-- >>> respIsTwoHunna response
264-- True
265-- >>> runBH' $ indexExists (IndexName "didimakeanindex")
266-- True
267createIndex :: MonadBH m => IndexSettings -> IndexName -> m Reply
268createIndex indexSettings (IndexName indexName) =
269 bindM2 put url (return body)
270 where url = joinPath [indexName]
271 body = Just $ encode indexSettings
274-- | 'deleteIndex' will delete an index given a 'Server', and an 'IndexName'.
275--
276-- >>> _ <- runBH' $ createIndex defaultIndexSettings (IndexName "didimakeanindex")
277-- >>> response <- runBH' $ deleteIndex (IndexName "didimakeanindex")
278-- >>> respIsTwoHunna response
279-- True
280-- >>> runBH' $ indexExists testIndex
281-- False
282deleteIndex :: MonadBH m => IndexName -> m Reply
283deleteIndex (IndexName indexName) =
284 delete =<< joinPath [indexName]
286-- | 'updateIndexSettings' will apply a non-empty list of setting updates to an index
287--
288-- >>> _ <- runBH' $ createIndex defaultIndexSettings (IndexName "unconfiguredindex")
289-- >>> response <- runBH' $ updateIndexSettings (BlocksWrite False :| []) (IndexName "unconfiguredindex")
290-- >>> respIsTwoHunna response
291-- True
292updateIndexSettings :: MonadBH m => NonEmpty UpdatableIndexSetting -> IndexName -> m Reply
293updateIndexSettings updates (IndexName indexName) =
294 bindM2 put url (return body)
295 where url = joinPath [indexName, "_settings"]
296 body = Just (encode jsonBody)
297 jsonBody = Object (deepMerge [u | Object u <- toJSON <$> toList updates])
300getIndexSettings :: (MonadBH m, MonadThrow m) => IndexName
301 -> m (Either EsError IndexSettingsSummary)
302getIndexSettings (IndexName indexName) = do
303 parseEsResponse =<< get =<< url
304 where url = joinPath [indexName, "_settings"]
307-- | 'optimizeIndex' will optimize a single index, list of indexes or
308-- all indexes. Note that this call will block until finishing but
309-- will continue even if the request times out. Concurrent requests to
310-- optimize an index while another is performing will block until the
311-- previous one finishes. For more information see
312-- <https://www.elastic.co/guide/en/elasticsearch/reference/1.7/indices-optimize.html>. Nothing
313-- worthwhile comes back in the reply body, so matching on the status
314-- should suffice.
315--
316-- 'optimizeIndex' with a maxNumSegments of 1 and onlyExpungeDeletes
317-- to True is the main way to release disk space back to the OS being
318-- held by deleted documents.
319--
320-- Note that this API was deprecated in ElasticSearch 2.1 for the
321-- almost completely identical forcemerge API. Adding support to that
322-- API would be trivial but due to the significant breaking changes,
323-- this library cannot currently be used with >= 2.0, so that feature was omitted.
324--
325-- >>> let ixn = IndexName "unoptimizedindex"
326-- >>> _ <- runBH' $ deleteIndex ixn >> createIndex defaultIndexSettings ixn
327-- >>> response <- runBH' $ optimizeIndex (IndexList (ixn :| [])) (defaultIndexOptimizationSettings { maxNumSegments = Just 1, onlyExpungeDeletes = True })
328-- >>> respIsTwoHunna response
329-- True
330optimizeIndex :: MonadBH m => IndexSelection -> IndexOptimizationSettings -> m Reply
331optimizeIndex ixs IndexOptimizationSettings {..} =
332 bindM2 post url (return body)
333 where url = addQuery params <$> joinPath [indexName, "_optimize"]
334 params = catMaybes [ ("max_num_segments",) . Just . showText <$> maxNumSegments
335 , Just ("only_expunge_deletes", Just (boolQP onlyExpungeDeletes))
336 , Just ("flush", Just (boolQP flushAfterOptimize))
337 ]
338 indexName = indexSelectionName ixs
339 boolQP True = "true"
340 boolQP False = "false"
341 body = Nothing
344-------------------------------------------------------------------------------
345indexSelectionName :: IndexSelection -> Text
346indexSelectionName (IndexList names) = T.intercalate "," [n | IndexName n <- toList names]
347indexSelectionName AllIndexes = "_all"
349deepMerge :: [Object] -> Object
350deepMerge = LS.foldl' go mempty
351 where go acc = LS.foldl' go' acc . HM.toList
352 go' acc (k, v) = HM.insertWith merge k v acc
353 merge (Object a) (Object b) = Object (deepMerge [a, b])
354 merge _ b = b
357statusCodeIs :: (Int, Int) -> Reply -> Bool
358statusCodeIs r resp = inRange r $ NHTS.statusCode (responseStatus resp)
360respIsTwoHunna :: Reply -> Bool
361respIsTwoHunna = statusCodeIs (200, 299)
363existentialQuery :: MonadBH m => Text -> m (Reply, Bool)
364existentialQuery url = do
365 reply <- head url
366 return (reply, respIsTwoHunna reply)
369-- | Tries to parse a response body as the expected type @a@ and
370-- failing that tries to parse it as an EsError. All well-formed, JSON
371-- responses from elasticsearch should fall into these two
372-- categories. If they don't, a 'EsProtocolException' will be
373-- thrown. If you encounter this, please report the full body it
374-- reports along with your ElasticSearch verison.
375parseEsResponse :: (MonadThrow m, FromJSON a) => Reply
376 -> m (Either EsError a)
377parseEsResponse reply
378 | respIsTwoHunna reply = case eitherDecode body of
379 Right a -> return (Right a)
380 Left _ -> tryParseError
381 | otherwise = tryParseError
382 where body = responseBody reply
383 tryParseError = case eitherDecode body of
384 Right e -> return (Left e)
385 -- this case should not be possible
386 Left _ -> explode
387 explode = throwM (EsProtocolException body)
389-- | 'indexExists' enables you to check if an index exists. Returns 'Bool'
390-- in IO
391--
392-- >>> exists <- runBH' $ indexExists testIndex
393indexExists :: MonadBH m => IndexName -> m Bool
394indexExists (IndexName indexName) = do
395 (_, exists) <- existentialQuery =<< joinPath [indexName]
396 return exists
398-- | 'refreshIndex' will force a refresh on an index. You must
399-- do this if you want to read what you wrote.
400--
401-- >>> _ <- runBH' $ createIndex defaultIndexSettings testIndex
402-- >>> _ <- runBH' $ refreshIndex testIndex
403refreshIndex :: MonadBH m => IndexName -> m Reply
404refreshIndex (IndexName indexName) =
405 bindM2 post url (return Nothing)
406 where url = joinPath [indexName, "_refresh"]
408stringifyOCIndex :: OpenCloseIndex -> Text
409stringifyOCIndex oci = case oci of
410 OpenIndex -> "_open"
411 CloseIndex -> "_close"
413openOrCloseIndexes :: MonadBH m => OpenCloseIndex -> IndexName -> m Reply
414openOrCloseIndexes oci (IndexName indexName) =
415 bindM2 post url (return Nothing)
416 where ociString = stringifyOCIndex oci
417 url = joinPath [indexName, ociString]
419-- | 'openIndex' opens an index given a 'Server' and an 'IndexName'. Explained in further detail at
420-- <http://www.elasticsearch.org/guide/en/elasticsearch/reference/current/indices-open-close.html>
421--
422-- >>> reply <- runBH' $ openIndex testIndex
423openIndex :: MonadBH m => IndexName -> m Reply
424openIndex = openOrCloseIndexes OpenIndex
426-- | 'closeIndex' closes an index given a 'Server' and an 'IndexName'. Explained in further detail at
427-- <http://www.elasticsearch.org/guide/en/elasticsearch/reference/current/indices-open-close.html>
428--
429-- >>> reply <- runBH' $ closeIndex testIndex
430closeIndex :: MonadBH m => IndexName -> m Reply
431closeIndex = openOrCloseIndexes CloseIndex
434-- | 'updateIndexAliases' updates the server's index alias
435-- table. Operations are atomic. Explained in further detail at
436-- <https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-aliases.html>
437--
438-- >>> let src = IndexName "a-real-index"
439-- >>> let aliasName = IndexName "an-alias"
440-- >>> let iAlias = IndexAlias src (IndexAliasName aliasName)
441-- >>> let aliasCreate = IndexAliasCreate Nothing Nothing
442-- >>> _ <- runBH' $ deleteIndex src
443-- >>> respIsTwoHunna <$> runBH' (createIndex defaultIndexSettings src)
444-- True
445-- >>> runBH' $ indexExists src
446-- True
447-- >>> respIsTwoHunna <$> runBH' (updateIndexAliases (AddAlias iAlias aliasCreate :| []))
448-- True
449-- >>> runBH' $ indexExists aliasName
450-- True
451updateIndexAliases :: MonadBH m => NonEmpty IndexAliasAction -> m Reply
452updateIndexAliases actions = bindM2 post url (return body)
453 where url = joinPath ["_aliases"]
454 body = Just (encode bodyJSON)
455 bodyJSON = object [ "actions" .= toList actions]
457-- | Get all aliases configured on the server.
458getIndexAliases :: (MonadBH m, MonadThrow m)
459 => m (Either EsError IndexAliasesSummary)
460getIndexAliases = parseEsResponse =<< get =<< url
461 where url = joinPath ["_aliases"]
463-- | 'putTemplate' creates a template given an 'IndexTemplate' and a 'TemplateName'.
464-- Explained in further detail at
465-- <https://www.elastic.co/guide/en/elasticsearch/reference/1.7/indices-templates.html>
466--
467-- >>> let idxTpl = IndexTemplate (TemplatePattern "tweet-*") (Just (IndexSettings (ShardCount 1) (ReplicaCount 1))) [toJSON TweetMapping]
468-- >>> resp <- runBH' $ putTemplate idxTpl (TemplateName "tweet-tpl")
469putTemplate :: MonadBH m => IndexTemplate -> TemplateName -> m Reply
470putTemplate indexTemplate (TemplateName templateName) =
471 bindM2 put url (return body)
472 where url = joinPath ["_template", templateName]
473 body = Just $ encode indexTemplate
475-- | 'templateExists' checks to see if a template exists.
476--
477-- >>> exists <- runBH' $ templateExists (TemplateName "tweet-tpl")
478templateExists :: MonadBH m => TemplateName -> m Bool
479templateExists (TemplateName templateName) = do
480 (_, exists) <- existentialQuery =<< joinPath ["_template", templateName]
481 return exists
483-- | 'deleteTemplate' is an HTTP DELETE and deletes a template.
484--
485-- >>> let idxTpl = IndexTemplate (TemplatePattern "tweet-*") (Just (IndexSettings (ShardCount 1) (ReplicaCount 1))) [toJSON TweetMapping]
486-- >>> _ <- runBH' $ putTemplate idxTpl (TemplateName "tweet-tpl")
487-- >>> resp <- runBH' $ deleteTemplate (TemplateName "tweet-tpl")
488deleteTemplate :: MonadBH m => TemplateName -> m Reply
489deleteTemplate (TemplateName templateName) =
490 delete =<< joinPath ["_template", templateName]
492-- | 'putMapping' is an HTTP PUT and has upsert semantics. Mappings are schemas
493-- for documents in indexes.
494--
495-- >>> _ <- runBH' $ createIndex defaultIndexSettings testIndex
496-- >>> resp <- runBH' $ putMapping testIndex testMapping TweetMapping
497-- >>> print resp
498-- Response {responseStatus = Status {statusCode = 200, statusMessage = "OK"}, responseVersion = HTTP/1.1, responseHeaders = [("Content-Type","application/json; charset=UTF-8"),("Content-Length","21")], responseBody = "{\"acknowledged\":true}", responseCookieJar = CJ {expose = []}, responseClose' = ResponseClose}
499putMapping :: (MonadBH m, ToJSON a) => IndexName
500 -> MappingName -> a -> m Reply
501putMapping (IndexName indexName) (MappingName mappingName) mapping =
502 bindM2 put url (return body)
503 where url = joinPath [indexName, "_mapping", mappingName]
504 -- "_mapping" and mappingName above were originally transposed
505 -- erroneously. The correct API call is: "/INDEX/_mapping/MAPPING_NAME"
506 body = Just $ encode mapping
508-- | 'deleteMapping' is an HTTP DELETE and deletes a mapping for a given index.
509-- Mappings are schemas for documents in indexes.
510--
511-- >>> _ <- runBH' $ createIndex defaultIndexSettings testIndex
512-- >>> _ <- runBH' $ putMapping testIndex testMapping TweetMapping
513-- >>> resp <- runBH' $ deleteMapping testIndex testMapping
514-- >>> print resp
515-- Response {responseStatus = Status {statusCode = 200, statusMessage = "OK"}, responseVersion = HTTP/1.1, responseHeaders = [("Content-Type","application/json; charset=UTF-8"),("Content-Length","21")], responseBody = "{\"acknowledged\":true}", responseCookieJar = CJ {expose = []}, responseClose' = ResponseClose}
516deleteMapping :: MonadBH m => IndexName -> MappingName -> m Reply
517deleteMapping (IndexName indexName)
518 (MappingName mappingName) =
519 -- "_mapping" and mappingName below were originally transposed
520 -- erroneously. The correct API call is: "/INDEX/_mapping/MAPPING_NAME"
521 delete =<< joinPath [indexName, "_mapping", mappingName]
523versionCtlParams :: IndexDocumentSettings -> [(Text, Maybe Text)]
524versionCtlParams cfg =
525 case idsVersionControl cfg of
526 NoVersionControl -> []
527 InternalVersion v -> versionParams v "internal"
528 ExternalGT (ExternalDocVersion v) -> versionParams v "external_gt"
529 ExternalGTE (ExternalDocVersion v) -> versionParams v "external_gte"
530 ForceVersion (ExternalDocVersion v) -> versionParams v "force"
531 where
532 vt = showText . docVersionNumber
533 versionParams v t = [ ("version", Just $ vt v)
534 , ("version_type", Just t)
535 ]
537-- | 'indexDocument' is the primary way to save a single document in
538-- Elasticsearch. The document itself is simply something we can
539-- convert into a JSON 'Value'. The 'DocId' will function as the
540-- primary key for the document.
541--
542-- >>> resp <- runBH' $ indexDocument testIndex testMapping defaultIndexDocumentSettings exampleTweet (DocId "1")
543-- >>> print resp
544-- Response {responseStatus = Status {statusCode = 201, statusMessage = "Created"}, responseVersion = HTTP/1.1, responseHeaders = [("Content-Type","application/json; charset=UTF-8"),("Content-Length","74")], responseBody = "{\"_index\":\"twitter\",\"_type\":\"tweet\",\"_id\":\"1\",\"_version\":1,\"created\":true}", responseCookieJar = CJ {expose = []}, responseClose' = ResponseClose}
545indexDocument :: (ToJSON doc, MonadBH m) => IndexName -> MappingName
546 -> IndexDocumentSettings -> doc -> DocId -> m Reply
547indexDocument (IndexName indexName)
548 (MappingName mappingName) cfg document (DocId docId) =
549 bindM2 put url (return body)
550 where url = addQuery params <$> joinPath [indexName, mappingName, docId]
551 parentParams = case idsParent cfg of
552 Nothing -> []
553 Just (DocumentParent (DocId p)) -> [ ("parent", Just p) ]
554 params = versionCtlParams cfg ++ parentParams
555 body = Just (encode document)
557-- | 'updateDocument' provides a way to perform an partial update of a
558-- an already indexed document.
559updateDocument :: (ToJSON patch, MonadBH m) => IndexName -> MappingName
560 -> IndexDocumentSettings -> patch -> DocId -> m Reply
561updateDocument (IndexName indexName)
562 (MappingName mappingName) cfg patch (DocId docId) =
563 bindM2 post url (return body)
564 where url = addQuery (versionCtlParams cfg) <$>
565 joinPath [indexName, mappingName, docId, "_update"]
566 body = Just (encode $ object ["doc" .= toJSON patch])
568-- | 'deleteDocument' is the primary way to delete a single document.
569--
570-- >>> _ <- runBH' $ deleteDocument testIndex testMapping (DocId "1")
571deleteDocument :: MonadBH m => IndexName -> MappingName
572 -> DocId -> m Reply
573deleteDocument (IndexName indexName)
574 (MappingName mappingName) (DocId docId) =
575 delete =<< joinPath [indexName, mappingName, docId]
577-- | 'bulk' uses
578-- <http://www.elasticsearch.org/guide/en/elasticsearch/reference/current/docs-bulk.html Elasticsearch's bulk API>
579-- to perform bulk operations. The 'BulkOperation' data type encodes the
580-- index/update/delete/create operations. You pass a 'V.Vector' of 'BulkOperation's
581-- and a 'Server' to 'bulk' in order to send those operations up to your Elasticsearch
582-- server to be performed. I changed from [BulkOperation] to a Vector due to memory overhead.
583--
584-- >>> let stream = V.fromList [BulkIndex testIndex testMapping (DocId "2") (toJSON (BulkTest "blah"))]
585-- >>> _ <- runBH' $ bulk stream
586-- >>> _ <- runBH' $ refreshIndex testIndex
587bulk :: MonadBH m => V.Vector BulkOperation -> m Reply
588bulk bulkOps = bindM2 post url (return body)
589 where url = joinPath ["_bulk"]
590 body = Just $ encodeBulkOperations bulkOps
592-- | 'encodeBulkOperations' is a convenience function for dumping a vector of 'BulkOperation'
593-- into an 'L.ByteString'
594--
595-- >>> let bulkOps = V.fromList [BulkIndex testIndex testMapping (DocId "2") (toJSON (BulkTest "blah"))]
596-- >>> encodeBulkOperations bulkOps
597-- "\n{\"index\":{\"_type\":\"tweet\",\"_id\":\"2\",\"_index\":\"twitter\"}}\n{\"name\":\"blah\"}\n"
598encodeBulkOperations :: V.Vector BulkOperation -> L.ByteString
599encodeBulkOperations stream = collapsed where
600 blobs = fmap encodeBulkOperation stream
601 mashedTaters = mash (mempty :: Builder) blobs
602 collapsed = toLazyByteString $ mappend mashedTaters (byteString "\n")
604mash :: Builder -> V.Vector L.ByteString -> Builder
605mash = V.foldl' (\b x -> b `mappend` (byteString "\n") `mappend` (lazyByteString x))
607mkBulkStreamValue :: Text -> Text -> Text -> Text -> Value
608mkBulkStreamValue operation indexName mappingName docId =
609 object [operation .=
610 object [ "_index" .= indexName
611 , "_type" .= mappingName
612 , "_id" .= docId]]
614-- | 'encodeBulkOperation' is a convenience function for dumping a single 'BulkOperation'
615-- into an 'L.ByteString'
616--
617-- >>> let bulkOp = BulkIndex testIndex testMapping (DocId "2") (toJSON (BulkTest "blah"))
618-- >>> encodeBulkOperation bulkOp
619-- "{\"index\":{\"_type\":\"tweet\",\"_id\":\"2\",\"_index\":\"twitter\"}}\n{\"name\":\"blah\"}"
620encodeBulkOperation :: BulkOperation -> L.ByteString
621encodeBulkOperation (BulkIndex (IndexName indexName)
622 (MappingName mappingName)
623 (DocId docId) value) = blob
624 where metadata = mkBulkStreamValue "index" indexName mappingName docId
625 blob = encode metadata `mappend` "\n" `mappend` encode value
627encodeBulkOperation (BulkCreate (IndexName indexName)
628 (MappingName mappingName)
629 (DocId docId) value) = blob
630 where metadata = mkBulkStreamValue "create" indexName mappingName docId
631 blob = encode metadata `mappend` "\n" `mappend` encode value
633encodeBulkOperation (BulkDelete (IndexName indexName)
634 (MappingName mappingName)
635 (DocId docId)) = blob
636 where metadata = mkBulkStreamValue "delete" indexName mappingName docId
637 blob = encode metadata
639encodeBulkOperation (BulkUpdate (IndexName indexName)
640 (MappingName mappingName)
641 (DocId docId) value) = blob
642 where metadata = mkBulkStreamValue "update" indexName mappingName docId
643 doc = object ["doc" .= value]
644 blob = encode metadata `mappend` "\n" `mappend` encode doc
646-- | 'getDocument' is a straight-forward way to fetch a single document from
647-- Elasticsearch using a 'Server', 'IndexName', 'MappingName', and a 'DocId'.
648-- The 'DocId' is the primary key for your Elasticsearch document.
649--
650-- >>> yourDoc <- runBH' $ getDocument testIndex testMapping (DocId "1")
651getDocument :: MonadBH m => IndexName -> MappingName
652 -> DocId -> m Reply
653getDocument (IndexName indexName)
654 (MappingName mappingName) (DocId docId) =
655 get =<< joinPath [indexName, mappingName, docId]
657-- | 'documentExists' enables you to check if a document exists. Returns 'Bool'
658-- in IO
659--
660-- >>> exists <- runBH' $ documentExists testIndex testMapping Nothing (DocId "1")
661documentExists :: MonadBH m => IndexName -> MappingName
662 -> Maybe DocumentParent -> DocId -> m Bool
663documentExists (IndexName indexName) (MappingName mappingName)
664 parent (DocId docId) = do
665 (_, exists) <- existentialQuery =<< url
666 return exists
667 where url = addQuery params <$> joinPath [indexName, mappingName, docId]
668 parentParam = fmap (\(DocumentParent (DocId p)) -> p) parent
669 params = LS.filter (\(_, v) -> isJust v) [("parent", parentParam)]
671dispatchSearch :: MonadBH m => Text -> Search -> m Reply
672dispatchSearch url search = post url' (Just (encode search))
673 where url' = appendSearchTypeParam url (searchType search)
675-- | 'searchAll', given a 'Search', will perform that search against all indexes
676-- on an Elasticsearch server. Try to avoid doing this if it can be helped.
677--
678-- >>> let query = TermQuery (Term "user" "bitemyapp") Nothing
679-- >>> let search = mkSearch (Just query) Nothing
680-- >>> reply <- runBH' $ searchAll search
681searchAll :: MonadBH m => Search -> m Reply
682searchAll = bindM2 dispatchSearch url . return
683 where url = joinPath ["_search"]
685-- | 'searchByIndex', given a 'Search' and an 'IndexName', will perform that search
686-- against all mappings within an index on an Elasticsearch server.
687--
688-- >>> let query = TermQuery (Term "user" "bitemyapp") Nothing
689-- >>> let search = mkSearch (Just query) Nothing
690-- >>> reply <- runBH' $ searchByIndex testIndex search
691searchByIndex :: MonadBH m => IndexName -> Search -> m Reply
692searchByIndex (IndexName indexName) = bindM2 dispatchSearch url . return
693 where url = joinPath [indexName, "_search"]
695-- | 'searchByType', given a 'Search', 'IndexName', and 'MappingName', will perform that
696-- search against a specific mapping within an index on an Elasticsearch server.
697--
698-- >>> let query = TermQuery (Term "user" "bitemyapp") Nothing
699-- >>> let search = mkSearch (Just query) Nothing
700-- >>> reply <- runBH' $ searchByType testIndex testMapping search
701searchByType :: MonadBH m => IndexName -> MappingName -> Search
702 -> m Reply
703searchByType (IndexName indexName)
704 (MappingName mappingName) = bindM2 dispatchSearch url . return
705 where url = joinPath [indexName, mappingName, "_search"]
707-- | For a given scearch, request a scroll for efficient streaming of
708-- search results. Note that the search is put into 'SearchTypeScan'
709-- mode and thus results will not be sorted. Combine this with
710-- 'advanceScroll' to efficiently stream through the full result set
711getInitialScroll :: MonadBH m => IndexName -> MappingName -> Search -> m (Maybe ScrollId)
712getInitialScroll (IndexName indexName) (MappingName mappingName) search = do
713 let url = joinPath [indexName, mappingName, "_search"]
714 search' = search { searchType = SearchTypeScan }
715 resp' <- bindM2 dispatchSearch url (return search')
716 let msr = decode' $ responseBody resp' :: Maybe (SearchResult ())
717 msid = maybe Nothing scrollId msr
718 return msid
720scroll' :: (FromJSON a, MonadBH m, MonadThrow m) => Maybe ScrollId -> m ([Hit a], Maybe ScrollId)
721scroll' Nothing = return ([], Nothing)
722scroll' (Just sid) = do
723 res <- advanceScroll sid 60
724 case res of
725 Right SearchResult {..} -> return (hits searchHits, scrollId)
726 Left _ -> return ([], Nothing)
728-- | Use the given scroll to fetch the next page of documents. If there are no
729-- further pages, 'SearchResult.searchHits.hits' will be '[]'.
730advanceScroll
731 :: ( FromJSON a
732 , MonadBH m
733 , MonadThrow m
734 )
735 => ScrollId
736 -> NominalDiffTime
737 -- ^ 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.
738 -> m (Either EsError (SearchResult a))
739advanceScroll (ScrollId sid) scroll = do
740 url <- joinPath ["_search/scroll?scroll=" <> scrollTime]
741 parseEsResponse =<< post url (Just . L.fromStrict $ T.encodeUtf8 sid)
742 where scrollTime = showText secs <> "s"
743 secs :: Integer
744 secs = round scroll
746simpleAccumulator :: (FromJSON a, MonadBH m, MonadThrow m) => [Hit a] -> ([Hit a], Maybe ScrollId) -> m ([Hit a], Maybe ScrollId)
747simpleAccumulator oldHits (newHits, Nothing) = return (oldHits ++ newHits, Nothing)
748simpleAccumulator oldHits ([], _) = return (oldHits, Nothing)
749simpleAccumulator oldHits (newHits, msid) = do
750 (newHits', msid') <- scroll' msid
751 simpleAccumulator (oldHits ++ newHits) (newHits', msid')
753-- | 'scanSearch' uses the 'scan&scroll' API of elastic,
754-- for a given 'IndexName' and 'MappingName'. Note that this will
755-- consume the entire search result set and will be doing O(n) list
756-- appends so this may not be suitable for large result sets. In that
757-- case, 'getInitialScroll' and 'advanceScroll' are good low level
758-- tools. You should be able to hook them up trivially to conduit,
759-- pipes, or your favorite streaming IO abstraction of choice. Note
760-- that ordering on the search would destroy performance and thus is
761-- ignored.
762scanSearch :: (FromJSON a, MonadBH m, MonadThrow m) => IndexName -> MappingName -> Search -> m [Hit a]
763scanSearch indexName mappingName search = do
764 msid <- getInitialScroll indexName mappingName search
765 (hits, msid') <- scroll' msid
766 (totalHits, _) <- simpleAccumulator [] (hits, msid')
767 return totalHits
769-- | 'mkSearch' is a helper function for defaulting additional fields of a 'Search'
770-- to Nothing in case you only care about your 'Query' and 'Filter'. Use record update
771-- syntax if you want to add things like aggregations or highlights while still using
772-- this helper function.
773--
774-- >>> let query = TermQuery (Term "user" "bitemyapp") Nothing
775-- >>> mkSearch (Just query) Nothing
776-- Search {queryBody = Just (TermQuery (Term {termField = "user", termValue = "bitemyapp"}) Nothing), filterBody = Nothing, sortBody = Nothing, aggBody = Nothing, highlight = Nothing, trackSortScores = False, from = From 0, size = Size 10, searchType = SearchTypeQueryThenFetch, fields = Nothing, source = Nothing}
777mkSearch :: Maybe Query -> Maybe Filter -> Search
778mkSearch query filter = Search query filter Nothing Nothing Nothing False (From 0) (Size 10) SearchTypeQueryThenFetch Nothing Nothing
780-- | 'mkAggregateSearch' is a helper function that defaults everything in a 'Search' except for
781-- the 'Query' and the 'Aggregation'.
782--
783-- >>> let terms = TermsAgg $ (mkTermsAggregation "user") { termCollectMode = Just BreadthFirst }
784-- >>> terms
785-- TermsAgg (TermsAggregation {term = Left "user", termInclude = Nothing, termExclude = Nothing, termOrder = Nothing, termMinDocCount = Nothing, termSize = Nothing, termShardSize = Nothing, termCollectMode = Just BreadthFirst, termExecutionHint = Nothing, termAggs = Nothing})
786-- >>> let myAggregation = mkAggregateSearch Nothing $ mkAggregations "users" terms
787mkAggregateSearch :: Maybe Query -> Aggregations -> Search
788mkAggregateSearch query mkSearchAggs = Search query Nothing Nothing (Just mkSearchAggs) Nothing False (From 0) (Size 0) SearchTypeQueryThenFetch Nothing Nothing
790-- | 'mkHighlightSearch' is a helper function that defaults everything in a 'Search' except for
791-- the 'Query' and the 'Aggregation'.
792--
793-- >>> let query = QueryMatchQuery $ mkMatchQuery (FieldName "_all") (QueryString "haskell")
794-- >>> let testHighlight = Highlights Nothing [FieldHighlight (FieldName "message") Nothing]
795-- >>> let search = mkHighlightSearch (Just query) testHighlight
796mkHighlightSearch :: Maybe Query -> Highlights -> Search
797mkHighlightSearch query searchHighlights = Search query Nothing Nothing Nothing (Just searchHighlights) False (From 0) (Size 10) SearchTypeQueryThenFetch Nothing Nothing
799-- | 'pageSearch' is a helper function that takes a search and assigns the from
800-- and size fields for the search. The from parameter defines the offset
801-- from the first result you want to fetch. The size parameter allows you to
802-- configure the maximum amount of hits to be returned.
803--
804-- >>> let query = QueryMatchQuery $ mkMatchQuery (FieldName "_all") (QueryString "haskell")
805-- >>> let search = mkSearch (Just query) Nothing
806-- >>> search
807-- 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}
808-- >>> pageSearch (From 10) (Size 100) search
809-- 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}
810pageSearch :: From -- ^ The result offset
811 -> Size -- ^ The number of results to return
812 -> Search -- ^ The current seach
813 -> Search -- ^ The paged search
814pageSearch resultOffset pageSize search = search { from = resultOffset, size = pageSize }
816parseUrl' :: MonadThrow m => Text -> m Request
817parseUrl' t = parseUrl (URI.escapeURIString URI.isAllowedInURI (T.unpack t))
819-- | Was there an optimistic concurrency control conflict when
820-- indexing a document?
821isVersionConflict :: Reply -> Bool
822isVersionConflict = statusCheck (== 409)
824isSuccess :: Reply -> Bool
825isSuccess = statusCheck (inRange (200, 299))
827isCreated :: Reply -> Bool
828isCreated = statusCheck (== 201)
830statusCheck :: (Int -> Bool) -> Reply -> Bool
831statusCheck prd = prd . NHTS.statusCode . responseStatus
833-- | This is a hook that can be set via the 'bhRequestHook' function
834-- that will authenticate all requests using an HTTP Basic
835-- Authentication header. Note that it is *strongly* recommended that
836-- this option only be used over an SSL connection.
837--
838-- >> (mkBHEnv myServer myManager) { bhRequestHook = basicAuthHook (EsUsername "myuser") (EsPassword "mypass") }
839basicAuthHook :: Monad m => EsUsername -> EsPassword -> Request -> m Request
840basicAuthHook (EsUsername u) (EsPassword p) = return . applyBasicAuth u' p'
841 where u' = T.encodeUtf8 u
842 p' = T.encodeUtf8 p