Merge branch 'bulk-insert'

CHANGELOG.md

@@ -8,6 +8,7 @@ This project adheres to [Semantic Versioning](http://semver.org/).
 - Option to specify nulls first or last, eg `/people?order=age.desc.nullsfirst`
 - Filter nulls, `?col=is.null` and `?col=isnot.null`
 - Filter within jsonb, `?col->a->>b=eq.c`
+- Accept CSV in post body for bulk inserts
 
 ### Fixed
 - Allow NULL values in posts

postgrest.cabal

@@ -42,6 +42,7 @@ executable postgrest
                      , blaze-builder
                      , vector
                      , mtl
+                     , cassava
   Other-Modules:       App
                      , Auth
                      , Config
@@ -98,4 +99,6 @@ Test-Suite spec
                      , blaze-builder
                      , vector
                      , mtl
+                     , cassava
                      , process
+                     , heredoc

src/App.hs

@@ -2,26 +2,30 @@
 module App (app, sqlError, isSqlError) where
 
 import Control.Monad (join)
-import Control.Arrow ((***))
+import Control.Arrow ((***), second)
 import Control.Applicative
 
 import Data.Text hiding (map)
-import Data.Maybe (fromMaybe)
+import Data.Maybe (fromMaybe, mapMaybe)
 import Text.Regex.TDFA ((=~))
 import Data.Ord (comparing)
 import Data.Ranged.Ranges (emptyRange)
-import Data.HashMap.Strict (keys, elems, filterWithKey, toList)
+import qualified Data.HashMap.Strict as M
 import Data.String.Conversions (cs)
+import Data.CaseInsensitive (original)
 import Data.List (sortBy)
 import Data.Functor.Identity
 import qualified Data.Set as S
 import qualified Data.ByteString.Lazy as BL
+import qualified Blaze.ByteString.Builder as BB
+import qualified Data.Csv as CSV
 
 import Network.HTTP.Types.Status
 import Network.HTTP.Types.Header
 import Network.HTTP.Types.URI (parseSimpleQuery)
 import Network.HTTP.Base (urlEncodeVars)
 import Network.Wai
+import Network.Wai.Internal (Response(..))
 
 import Data.Aeson
 import Data.Monoid
@@ -98,26 +102,40 @@ app v1schema reqBody req =
             , (hLocation, "/postgrest/users?id=eq." <> cs (userId u))
             ] ""
 
-    ([table], "POST") ->
+    ([table], "POST") -> do
-      handleJsonObj reqBody $ \obj -> do
+      let qt = QualifiedTable schema (cs table)
-        let qt = QualifiedTable schema (cs table)
+          echoRequested = lookup "Prefer" hdrs == Just "return=representation"
-            query = insertInto qt (map cs $ keys obj) (elems obj)
+          parsed :: Either String (V.Vector Text, V.Vector (V.Vector Value))
-            echoRequested = lookup "Prefer" hdrs == Just "return=representation"
+          parsed = if lookup "Content-Type" hdrs == Just "text/csv"
-        row <- H.maybeEx query
+                    then do
-        let (Identity insertedJson) = fromMaybe (Identity "{}" :: Identity Text) row
+                      rows <- CSV.decode CSV.NoHeader reqBody
-            Just inserted = decode (cs insertedJson) :: Maybe Object
+                      if V.null rows then Left "CSV requires header"
-
+                        else Right (V.head rows, (V.map $ V.map $ parseCsvCell . cs) (V.tail rows))
-        primaryKeys <- map cs <$> primaryKeyColumns qt
+                    else eitherDecode reqBody >>= \val ->
-        let primaries = if Prelude.null primaryKeys
+                      case val of
-            then inserted
+                        Object obj -> Right .  second V.singleton .  V.unzip .  V.fromList $
-            else filterWithKey (const . (`elem` primaryKeys)) inserted
+                          M.toList obj
-        let params = urlEncodeVars
+                        _ -> Left "Expecting single JSON object or CSV rows"
-              $ map (\t -> (cs $ fst t, cs (paramFilter $ snd t)))
+      case parsed of
-              $ sortBy (comparing fst) $ toList primaries
+        Left err -> return $ responseLBS status400 [] $
-        return $ responseLBS status201
+          encode . object $ [("message", String $ "Failed to parse JSON payload. " <> cs err)]
-          [ jsonH
+        Right toBeInserted -> do
-          , (hLocation, "/" <> cs table <> "?" <> cs params)
+          rows :: [Identity Text] <- H.listEx $ uncurry (insertInto qt) toBeInserted
-          ] $ if echoRequested then cs insertedJson else ""
+          let inserted :: [Object] = mapMaybe (decode . cs . runIdentity) rows
+          primaryKeys <- primaryKeyColumns qt
+          let responses = flip map inserted $ \obj -> do
+                let primaries =
+                      if Prelude.null primaryKeys
+                        then obj
+                        else M.filterWithKey (const . (`elem` primaryKeys)) obj
+                let params = urlEncodeVars
+                      $ map (\t -> (cs $ fst t, cs (paramFilter $ snd t)))
+                      $ sortBy (comparing fst) $ M.toList primaries
+                responseLBS status201
+                  [ jsonH
+                  , (hLocation, "/" <> cs table <> "?" <> cs params)
+                  ] $ if echoRequested then encode obj else ""
+          return $ multipart status201 responses
 
     ([table], "PUT") ->
       handleJsonObj reqBody $ \obj -> do
@@ -129,10 +147,10 @@ app v1schema reqBody req =
                "You must speficy all and only primary keys as params"
           else do
             tableCols <- map (cs . colName) <$> columns qt
-            let cols = map cs $ keys obj
+            let cols = map cs $ M.keys obj
             if S.fromList tableCols == S.fromList cols
               then do
-                let vals = elems obj
+                let vals = M.elems obj
                 H.unitEx $ iffNotT
                         (whereT qq $ update qt cols vals)
                         (insertSelect qt cols vals)
@@ -148,7 +166,7 @@ app v1schema reqBody req =
         let qt = QualifiedTable schema (cs table)
         H.unitEx
           $ whereT qq
-          $ update qt (map cs $ keys obj) (elems obj)
+          $ update qt (map cs $ M.keys obj) (M.elems obj)
         return $ responseLBS status204 [ jsonH ] ""
 
     ([table], "DELETE") -> do
@@ -227,6 +245,26 @@ handleJsonObj reqBody handler = do
         jErr = encode . object $
           [("message", String "Expecting a JSON object")]
 
+parseCsvCell :: BL.ByteString -> Value
+parseCsvCell s = if s == "NULL" then Null else String $ cs s
+
+multipart :: Status -> [Response] -> Response
+multipart _ [] = responseLBS status204 [] ""
+multipart _ [r] = r
+multipart s rs =
+  responseLBS s [(hContentType, "multipart/mixed; boundary=\"postgrest_boundary\"")] $
+    BL.intercalate "\n--postgrest_boundary\n" (map renderResponseBody rs)
+
+  where
+    renderHeader :: Header -> BL.ByteString
+    renderHeader (k, v) = cs (original k) <> ": " <> cs v
+
+    renderResponseBody :: Response -> BL.ByteString
+    renderResponseBody (ResponseBuilder _ headers b) =
+      BL.intercalate "\n" (map renderHeader headers)
+        <> "\n\n" <> BB.toLazyByteString b
+    renderResponseBody _ = error
+      "Unable to create multipart response from non-ResponseBuilder"
 
 data TableOptions = TableOptions {
   tblOptcolumns :: [Column]

src/PgQuery.hs

@@ -22,6 +22,7 @@ import Control.Monad (join)
 import Data.String.Conversions (cs)
 import qualified Data.Aeson as JSON
 import qualified Data.List as L
+import qualified Data.Vector as V
 import Data.Scientific (isInteger, formatScientific, FPFormat(..))
 
 type PStmt = H.Stmt P.Postgres
@@ -108,16 +109,24 @@ returningStarT s = s { B.stmtTemplate = B.stmtTemplate s <> " RETURNING *" }
 deleteFrom :: QualifiedTable -> PStmt
 deleteFrom t = B.Stmt ("delete from " <> fromQt t) empty True
 
-insertInto :: QualifiedTable -> [T.Text] -> [JSON.Value] -> PStmt
+insertInto :: QualifiedTable
-insertInto t [] _ = B.Stmt
+              -> V.Vector T.Text
-  ("insert into " <> fromQt t <> " default values returning *") empty True
+              -> V.Vector (V.Vector JSON.Value)
-insertInto t cols vals = B.Stmt
+              -> PStmt
-  ("insert into " <> fromQt t <> " (" <>
+insertInto t cols vals
-    T.intercalate ", " (map pgFmtIdent cols) <>
+  | V.null cols = B.Stmt ("insert into " <> fromQt t <> " default values returning *") empty True
-    ") values ("
+  | otherwise   = B.Stmt
-    <> T.intercalate ", " (map insertableValue vals)
+    ("insert into " <> fromQt t <> " (" <>
-    <> ") returning row_to_json(" <> fromQt t <> ".*)")
+      T.intercalate ", " (V.toList $ V.map pgFmtIdent cols) <>
-  empty True
+      ") values "
+      <> T.intercalate ", "
+        (V.toList $ V.map (\v -> "("
+            <> T.intercalate ", " (V.toList $ V.map insertableValue v)
+            <> ")"
+          ) vals
+        )
+      <> " returning row_to_json(" <> fromQt t <> ".*)")
+    empty True
 
 insertSelect :: QualifiedTable -> [T.Text] -> [JSON.Value] -> PStmt
 insertSelect t [] _ = B.Stmt
@@ -261,9 +270,12 @@ unquoted (JSON.Number n) =
 unquoted (JSON.Bool b) = cs . show $ b
 unquoted _ = ""
 
+insertableText :: T.Text -> T.Text
+insertableText = (<> "::unknown") . pgFmtLit
+
 insertableValue :: JSON.Value -> T.Text
 insertableValue JSON.Null = "null"
-insertableValue v = ((<> "::unknown") . pgFmtLit . unquoted) v
+insertableValue v = insertableText $ unquoted v
 
 paramFilter :: JSON.Value -> T.Text
 paramFilter JSON.Null = "is.null"

test/Feature/InsertSpec.hs

@@ -9,6 +9,7 @@ import SpecHelper
 
 import qualified Data.Aeson as JSON
 import Data.Maybe (fromJust)
+import Text.Heredoc
 import Network.HTTP.Types.Header
 import Network.HTTP.Types
 import Control.Monad (replicateM_)
@@ -93,6 +94,50 @@ spec = afterAll_ resetDb $ around withApp $ do
           , matchHeaders = []
           }
 
+  describe "CSV insert" $ do
+
+    after_ (clearTable "menagerie") . context "disparate csv types" $
+      it "succeeds with multipart response" $ do
+        p <- request methodPost "/menagerie" [("Content-Type", "text/csv")]
+               [str|integer,double,varchar,boolean,date,money,enum
+                   |13,3.14159,testing!,false,1900-01-01,$3.99,foo
+                   |12,0.1,a string,true,1929-10-01,12,bar
+                   |]
+        liftIO $ do
+          simpleBody p `shouldBe` "Content-Type: application/json\nLocation: /menagerie?integer=eq.13\n\n\n--postgrest_boundary\nContent-Type: application/json\nLocation: /menagerie?integer=eq.12\n\n"
+          simpleStatus p `shouldBe` created201
+
+    after_ (clearTable "no_pk") . context "requesting full representation" $ do
+      it "returns full details of inserted record" $
+        request methodPost "/no_pk"
+                     [("Content-Type", "text/csv"), ("Prefer", "return=representation")]
+                     "a,b\nbar,baz"
+          `shouldRespondWith` ResponseMatcher {
+            matchBody    = Just [json| { "a":"bar", "b":"baz" } |]
+          , matchStatus  = 201
+          , matchHeaders = ["Content-Type" <:> "application/json",
+                            "Location" <:> "/no_pk?a=eq.bar&b=eq.baz"]
+          }
+
+      it "can post nulls" $
+        request methodPost "/no_pk"
+                     [("Content-Type", "text/csv"), ("Prefer", "return=representation")]
+                     "a,b\nNULL,foo"
+          `shouldRespondWith` ResponseMatcher {
+            matchBody    = Just [json| { "a":null, "b":"foo" } |]
+          , matchStatus  = 201
+          , matchHeaders = ["Content-Type" <:> "application/json",
+                            "Location" <:> "/no_pk?a=is.null&b=eq.foo"]
+          }
+
+    after_ (clearTable "no_pk") . context "with wrong number of columns" $ do
+      it "fails for too few" $ do
+        p <- request methodPost "/no_pk" [("Content-Type", "text/csv")] "a,b\nfoo,bar\nbaz"
+        liftIO $ simpleStatus p `shouldBe` badRequest400
+      it "fails for too many" $ do
+        p <- request methodPost "/no_pk" [("Content-Type", "text/csv")] "a,b\nfoo,bar\nbaz,bat,bad"
+        liftIO $ simpleStatus p `shouldBe` badRequest400
+
   describe "Putting record" $ do
 
     context "to unkonwn uri" $