hscassandra (empty) → 0.0.4
raw patch · 6 files changed
+329/−0 lines, 6 filesdep +Thriftdep +basedep +bytestringsetup-changed
Dependencies added: Thrift, base, bytestring, cassandra-thrift, containers, mtl, network, old-time
Files
- README +0/−0
- Setup.hs +3/−0
- hscassandra.cabal +31/−0
- src/Database/Cassandra.hs +151/−0
- src/Database/Cassandra/Monad.hs +106/−0
- src/Database/Cassandra/Types.hs +38/−0
+ README view
+ Setup.hs view
@@ -0,0 +1,3 @@+import Distribution.Simple+main = defaultMain+
+ hscassandra.cabal view
@@ -0,0 +1,31 @@+name : hscassandra +version : 0.0.4+license : BSD3 +cabal-version : >= 1.6+author : Kirk Peterson +maintainer : kirk@glyphsoftware.com,+ necrobious@gmail.com+homepage : https://github.com/necrobious/hscassandra+stability : alpha+synopsis : cassandra database interface +description:+ A simple abstration over the Cassandra Thrift bindings designed to make working with Cassandra's thrift API easier, without removing the ability to use it directly.+ A Cassandra monad is presented for mannaging the connection to the Cassandra server, and the calculation of cassandra timestams (in microseconds).+category : Database +build-type : Simple+source-repository head+ type : git+ location : git://github.com/necrobious/hscassandra.git+library+ hs-source-dirs : src+ exposed-modules: Database.Cassandra,+ Database.Cassandra.Monad,+ Database.Cassandra.Types+ build-depends : base >= 4.2.0.0 && < 4.9, + containers,+ old-time,+ mtl >= 2.0,+ bytestring,+ network,+ Thrift >= 0.5.0.2,+ cassandra-thrift >= 0.7.2.1
+ src/Database/Cassandra.hs view
@@ -0,0 +1,151 @@+module Database.Cassandra+ ( Column(..)+ , (=:)+ , (=|)+ , insert+ , Filter(..)+ , get+ , multiget+ , range+ , columns++ , withCassandra+ , CassandraConfig(..)+ , initConfig+ , getConnection+ , getKeyspace+ , setKeyspace+ , getConsistencyLevel+ , setConsistencyLevel+ , getTime+ , getCassandra+ , Cassandra+ , CassandraT+ ) where++import qualified Database.Cassandra.Thrift.Cassandra_Types as Thrift+import qualified Database.Cassandra.Thrift.Cassandra_Client as Cas++import Database.Cassandra.Monad+import Database.Cassandra.Types++import Data.Map (Map)+import qualified Data.Map as Map++import Data.ByteString.Lazy(ByteString)+import qualified Data.ByteString.Lazy.Char8 as Lazy++import Control.Monad.Trans+import Data.Int(Int32)+++data Column = Column ColumnName ColumnValue+ | Super ColumnName [Column] + deriving (Show)++-- | Build up a column's insert values+(=:) :: (BS column, BS value) => column -> value -> Column+(=:) col val = Column (bs col) (bs val)++-- | Build up a super column's insert values+(=|) :: (BS supercolumn) => supercolumn -> [Column] -> Column +(=|) sup = Super (bs sup)++-- insert "Users" "necrobious@gmail.com" +-- [ "fn" =: "Kirk"+-- , "ln" =: "Peterson"+-- , "Address" =| [ "street1" =: "2020"+-- , "state" =: "Oregon"]+-- ] +insert :: (BS key) => ColumnFamily -> key -> [Column] -> Cassandra ()+insert column_family key columns = do + consistency <- getConsistencyLevel+ conn <- getConnection+ now <- getTime+ liftIO $ Cas.batch_mutate conn (Map.singleton (bs key) (Map.singleton column_family (mutations now))) consistency+ where+ mutations now = map (q now) columns+ q now c = Thrift.Mutation (Just $ f now c) Nothing+ f now (Super s cs) = Thrift.ColumnOrSuperColumn Nothing (Just (Thrift.SuperColumn (Just s) (Just $ map (m now) cs))) + f now (Column c v) = Thrift.ColumnOrSuperColumn (Just (col c v now)) Nothing+ col c v now = Thrift.Column (Just c) (Just v) (Just now) Nothing+ m now (Column c v) = col c v now++data Filter = AllColumns+ | ColNames [ByteString]+ | ColRange+ { rangeStart :: ByteString+ , rangeEnd :: ByteString+ , rangeReverse :: Bool+ , rangeLimit :: Int32+ }+-- | a smarter constructor for building a Range filter+range :: (BS column_name) => column_name -> column_name -> Bool -> Int32 -> Filter+range start finish = ColRange (bs start) (bs finish) ++-- | a smarter constructor for building a Columns filter +columns :: (BS column_name) => [column_name] -> Filter+columns = ColNames . (map bs) ++-- | for the given key, within the column family, retrieve all columns, unless filtered+get :: (BS key) => ColumnFamily -> key -> Filter -> Cassandra [Column]+get column_family key filters = do+ consistency <- getConsistencyLevel+ conn <- getConnection+ results <- liftIO $ Cas.get_slice conn (bs key) (column_parent column_family) (slice_predicate filters) consistency + return $ foldr rewrap [] results++get_count :: (BS key) => ColumnFamily -> key -> Filter -> Cassandra Int32+get_count column_family key filters = do+ consistency <- getConsistencyLevel+ conn <- getConnection+ liftIO $ Cas.get_count conn (bs key) (column_parent column_family) (slice_predicate filters) consistency + ++multiget :: (BS key) => ColumnFamily -> [key] -> Filter -> Cassandra (Map key [Column])+multiget column_family keys filters = do+ let byBs = keys2map keys+ consistency <- getConsistencyLevel+ conn <- getConnection+ results <- liftIO $ Cas.multiget_slice conn (Map.keys byBs) (column_parent column_family) (slice_predicate filters) consistency + return $ map2map byBs $ Map.foldrWithKey remap Map.empty results -- $ foldr rewrap [] results+++keys2map :: (BS key) => [key] -> Map ByteString key+keys2map keys = foldr (\ k m -> Map.insert (bs k) k m) Map.empty keys++map2map :: (BS key) => Map ByteString key -> Map ByteString a -> Map key a+map2map lookupMap resultsMap = Map.foldrWithKey foldOver Map.empty resultsMap + where+ foldOver resultKey val accMap =+ case Map.lookup resultKey lookupMap of+ Just lookupKey -> Map.insert lookupKey val accMap+ Nothing -> accMap+++remap :: (BS key) => key -> [Thrift.ColumnOrSuperColumn] -> Map key [Column] -> Map key [Column]+remap key cols acc = Map.insert key (foldr rewrap [] cols) acc ++rewrap :: Thrift.ColumnOrSuperColumn -> [Column] -> [Column]+rewrap (Thrift.ColumnOrSuperColumn (Just (Thrift.Column (Just n) (Just v) _ _)) Nothing) acc = + (Column n v) : acc+rewrap (Thrift.ColumnOrSuperColumn Nothing (Just (Thrift.SuperColumn (Just n) (Just cs)))) acc =+ (Super n (foldr c2c [] cs)) : acc +rewrap _ acc = acc++c2c :: Thrift.Column -> [Column] -> [Column]+c2c (Thrift.Column (Just n) (Just v) _ _) acc = (Column n v) : acc+c2c _ acc = acc+++column_parent :: ColumnFamily -> Thrift.ColumnParent+column_parent column_family = Thrift.ColumnParent (Just column_family) Nothing++slice_predicate :: Filter -> Thrift.SlicePredicate+slice_predicate AllColumns =+ Thrift.SlicePredicate Nothing (Just $ Thrift.SliceRange (Just Lazy.empty) (Just Lazy.empty) (Just False) (Just 100)) +slice_predicate (ColNames bs) = + Thrift.SlicePredicate (Just bs) Nothing+slice_predicate (ColRange rs re rr rl) = + Thrift.SlicePredicate Nothing (Just (Thrift.SliceRange (Just rs) (Just re) (Just rr) (Just rl)))+
+ src/Database/Cassandra/Monad.hs view
@@ -0,0 +1,106 @@+{-# LANGUAGE DeriveDataTypeable, GeneralizedNewtypeDeriving #-}+module Database.Cassandra.Monad+ ( withCassandra+ , CassandraConfig(..)+ , initConfig+ , getConnection+ , getKeyspace+ , setKeyspace+ , getConsistencyLevel+ , setConsistencyLevel+ , getTime+ , getCassandra+ , Cassandra+ , CassandraT+ ) where+import Database.Cassandra.Types+import Network -- (PortID, PortNumber)+import Thrift.Protocol.Binary+import Thrift.Transport.Handle+import Thrift.Transport.Framed+import Database.Cassandra.Thrift.Cassandra_Types+import Database.Cassandra.Thrift.Cassandra_Client hiding (get)+import Control.Exception (bracket)+import System.IO (hClose, Handle)+import Data.Int (Int64)+import Control.Monad+import Control.Monad.State+import System.Time+import Data.Map (Map) +import qualified Data.Map as M++withCassandra :: CassandraConfig -> Cassandra a -> IO a +withCassandra config callback = do+ bracket+ (hOpen (cassandraHostname config, PortNumber $ fromIntegral $ cassandraPort config))+ (\ h -> tFlush h >> tClose h)+ (\ handle -> do + framed <- openFramedTransport handle+ let binpro = BinaryProtocol framed + let conn = (binpro, binpro)+ login conn (authreq config)+ set_keyspace conn (cassandraKeyspace config)+ liftM fst $ runCassandraT callback (config{cassandraConnection=conn})+ )+ where+ authreq CassandraConfig{cassandraUsername=u, cassandraPassword=p} = + AuthenticationRequest{f_AuthenticationRequest_credentials=Just $ credmap u p}+ credmap username password = M.insert "password" password (M.insert "username" username M.empty)++data CassandraConfig = CassandraConfig+ { cassandraConnection :: (BinaryProtocol (FramedTransport Handle), BinaryProtocol (FramedTransport Handle)) + , cassandraKeyspace :: String+ , cassandraConsistencyLevel :: ConsistencyLevel+ , cassandraHostname :: Hostname+ , cassandraPort :: Port+ , cassandraUsername :: Username+ , cassandraPassword :: Password+ }++instance Show CassandraConfig where+ show c = "KS: "++ (cassandraKeyspace c) ++ ", CL: " ++ (show $ cassandraConsistencyLevel c) ++ ", host: " ++ (cassandraHostname c) ++ ", port: " ++ (show $ cassandraPort c) ++ ", user: " ++ (cassandraUsername c) ++ ", pass: " ++ (cassandraPassword c) ++initConfig :: CassandraConfig+initConfig = CassandraConfig+ { cassandraConnection = undefined+ , cassandraKeyspace = "system"+ , cassandraConsistencyLevel = ONE+ , cassandraHostname = "127.0.0.1"+ , cassandraPort = 9160+ , cassandraUsername = "default"+ , cassandraPassword = ""+ }++getConnection :: Cassandra (BinaryProtocol (FramedTransport Handle), BinaryProtocol (FramedTransport Handle))+getConnection = cassandraConnection `fmap` get -- CassandraT ask++getKeyspace :: Cassandra Keyspace+getKeyspace = cassandraKeyspace `fmap` get++getConsistencyLevel :: Cassandra ConsistencyLevel+getConsistencyLevel = cassandraConsistencyLevel `fmap` get++setConsistencyLevel :: ConsistencyLevel -> Cassandra ()+setConsistencyLevel consistency = + getCassandra >>= \ config -> put config{cassandraConsistencyLevel=consistency} ++setKeyspace :: Keyspace -> Cassandra ()+setKeyspace keyspace = do+ config <- getCassandra+ conn <- getConnection+ liftIO $ set_keyspace conn keyspace+ put config{cassandraKeyspace=keyspace}+ +-- cassandra is VERY sensitive to its timestamp values.. as a convention, timestamps are always in microseconds.+getTime :: Cassandra Int64+getTime = do+ TOD sec pico <- liftIO getClockTime+ return $ fromInteger $ (sec * 1000000) + (toInteger $ pico `div` 1000000)++getCassandra :: Cassandra CassandraConfig+getCassandra = get ++type Cassandra a = CassandraT a+type CassandraT a = StateT CassandraConfig IO a+runCassandraT = runStateT+
+ src/Database/Cassandra/Types.hs view
@@ -0,0 +1,38 @@+{-# LANGUAGE TypeSynonymInstances #-}++module Database.Cassandra.Types where++import Data.ByteString.Lazy(ByteString)+import Data.ByteString.Lazy.Char8(pack, fromChunks)+import qualified Data.ByteString as Strict++type Port = Int+type Hostname = String+type Username = String+type Password = String++-- type alias' for making Cassandra's Thrift API easier to understand +type ClusterName = String+type Keyspace = String+type Partitioner = String+type Snitch = String+type ColumnFamily = String+type ColumnName = ByteString +type ColumnValue = ByteString +type ThriftApiVersion = String+type SchemaId = String +++class (Ord a) => BS a where+ bs :: a -> ByteString++instance BS String where+ bs = pack ++instance BS ByteString where+ bs = id + +instance BS Strict.ByteString where+ bs bs' = fromChunks [bs'] ++