valiant-conduit (empty) → 0.1.0.0

raw patch · 7 files changed

+354/−0 lines, 7 filesdep +basedep +bytestringdep +conduit

Dependencies added: base, bytestring, conduit, hspec, pg-wire, text, valiant, valiant-conduit

Files

CHANGELOG.md +14/−0
LICENSE +27/−0
README.md +39/−0
src/Valiant/Conduit.hs +68/−0
test/Main.hs +66/−0
test/TestSupport.hs +63/−0
valiant-conduit.cabal +77/−0

+ CHANGELOG.md view

@@ -0,0 +1,14 @@+# Changelog++The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),+and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).++## [0.1.0.0] - 2026-04-25++Initial release. Conduit streaming adapter for+[`valiant`](https://hackage.haskell.org/package/valiant). Exposes+`selectSource` (cursor-based, requires a transaction) and+`foldSource` (single-shot extended-protocol) as `ConduitT () r IO ()`+producers.++[0.1.0.0]: https://github.com/joshburgess/valiant/releases/tag/v0.1.0.0

+ LICENSE view

@@ -0,0 +1,27 @@+Copyright (c) 2026 Josh Burgess++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++1. Redistributions of source code must retain the above copyright notice,+   this list of conditions and the following disclaimer.++2. Redistributions in binary form must reproduce the above copyright notice,+   this list of conditions and the following disclaimer in the documentation+   and/or other materials provided with the distribution.++3. Neither the name of the copyright holder nor the names of its+   contributors may be used to endorse or promote products derived from this+   software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"+AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE+IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE+ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE+LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR+CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF+SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS+INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN+CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)+ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE+POSSIBILITY OF SUCH DAMAGE.

+ README.md view

@@ -0,0 +1,39 @@+# valiant-conduit++[Conduit](https://hackage.haskell.org/package/conduit) streaming+adapter for [`valiant`](https://hackage.haskell.org/package/valiant).++Produces `ConduitT () r IO ()` values from query results, so query+output composes with the rest of the conduit ecosystem.++## Quick start++```haskell+import Valiant+import Valiant.Conduit+import Conduit++countActiveUsers :: Pool -> IO Int+countActiveUsers pool =+  withTransaction pool $ \tx ->+    runConduit $+      selectSource (txConn tx) listAllUsers () 500+      .| filterC userActive+      .| lengthC+```++## Two streaming strategies++- `selectSource conn stmt params batchSize` — cursor-based. Must run+  inside a transaction. Fetches in batches of `batchSize` rows.+- `foldSource conn stmt params` — single-shot extended-protocol query.+  No transaction required.++Both functions accumulate one result set at a time before yielding,+so memory usage scales with the active batch (cursor) or the full+result set (`foldSource`). For truly incremental memory use across a+large cursor, drive `Valiant.withCursor` / `Valiant.fetchBatch`+yourself.++See the [valiant tutorial](https://github.com/joshburgess/valiant/blob/main/docs/TUTORIAL.md)+for `Statement` definitions and the `valiant prepare` workflow.

+ src/Valiant/Conduit.hs view

@@ -0,0 +1,68 @@+{-# OPTIONS_GHC -fno-full-laziness #-}++-- | Conduit streaming adapter for valiant.+--+-- Stream query results as @ConduitT@ sources, enabling integration with+-- the conduit ecosystem for composable stream processing.+--+-- Two streaming strategies are provided:+--+-- * 'selectSource' — cursor-based, requires a transaction, fetches in+--   configurable batches from the server.+-- * 'foldSource' — single-shot extended-protocol query, no transaction+--   required.+--+-- @+-- import Valiant+-- import Valiant.Conduit+-- import Conduit+--+-- withTransaction pool $ \\tx ->+--   runConduit $+--     selectSource (txConn tx) listAllUsers () 500+--     .| mapC userName+--     .| sinkList+-- @+module Valiant.Conduit+  ( -- * Cursor-based streaming (requires transaction)+    selectSource+    -- * Fold-based streaming (no transaction required)+  , foldSource+  ) where++import Control.Monad.IO.Class (liftIO)+import Data.Conduit (ConduitT, yield)+import Valiant (Connection, Statement, fetchAll, fetchAllCursor)++-- | Stream query results using a server-side cursor.+--+-- Must be called inside a transaction. Fetches rows in batches of the+-- given size, yielding decoded rows one at a time into the conduit.+--+-- Note: rows are accumulated in memory before yielding because cursor+-- access requires an exclusive wire session that completes before the+-- conduit can run. For incremental memory use, see 'foldSource' (which+-- returns a single result set) or use the lower-level cursor API+-- ('Valiant.withCursor' / 'Valiant.fetchBatch') directly.+selectSource+  :: Connection+  -> Statement p r+  -> p+  -> Int+  -- ^ Batch size (number of rows per FETCH)+  -> ConduitT () r IO ()+selectSource conn stmt params batchSize = do+  rows <- liftIO (fetchAllCursor conn stmt params batchSize)+  mapM_ yield rows++-- | Stream query results using single-shot extended-protocol execution.+--+-- Does not require a transaction.+foldSource+  :: Connection+  -> Statement p r+  -> p+  -> ConduitT () r IO ()+foldSource conn stmt params = do+  rows <- liftIO (fetchAll conn stmt params)+  mapM_ yield rows

+ test/Main.hs view

@@ -0,0 +1,66 @@+module Main where++import Data.Conduit (runConduit, (.|))+import Data.Conduit.List qualified as CL+import Data.Int (Int32)+import Data.Text (Text)+import Valiant+import Valiant.Conduit (foldSource, selectSource)+import Test.Hspec+import TestSupport++main :: IO ()+main = hspec $ do+  describe "Valiant.Conduit" $ do+    describe "selectSource (cursor-based)" $ do+      it "streams all rows from a cursor" $ do+        withTestConnection $ \conn -> withSchema conn $ do+          insertTestUsers conn+          withTransactionConn conn $ \tx -> do+            rows <- runConduit $+              selectSource (txConn tx) stmtListAll () 2+              .| CL.consume+            length rows `shouldBe` 5++      it "returns empty for no rows" $ do+        withTestConnection $ \conn -> withSchema conn $ do+          withTransactionConn conn $ \tx -> do+            rows <- runConduit $+              selectSource (txConn tx) stmtListAll () 10+              .| CL.consume+            rows `shouldBe` ([] :: [(Int32, Text)])++      it "respects batch size by returning correct total" $ do+        withTestConnection $ \conn -> withSchema conn $ do+          insertTestUsers conn+          withTransactionConn conn $ \tx -> do+            rows <- runConduit $+              selectSource (txConn tx) stmtListAll () 1+              .| CL.consume+            length rows `shouldBe` 5++    describe "foldSource (no transaction)" $ do+      it "streams all rows without a transaction" $ do+        withTestConnection $ \conn -> withSchema conn $ do+          insertTestUsers conn+          rows <- runConduit $+            foldSource conn stmtListAll ()+            .| CL.consume+          length rows `shouldBe` 5++      it "returns empty for no rows" $ do+        withTestConnection $ \conn -> withSchema conn $ do+          rows <- runConduit $+            foldSource conn stmtListAll ()+            .| CL.consume+          rows `shouldBe` ([] :: [(Int32, Text)])++      it "composes with conduit combinators" $ do+        withTestConnection $ \conn -> withSchema conn $ do+          insertTestUsers conn+          names <- runConduit $+            foldSource conn stmtListAll ()+            .| CL.map snd+            .| CL.consume+          length names `shouldBe` 5+          take 1 names `shouldBe` ["Alice"]

+ test/TestSupport.hs view

@@ -0,0 +1,63 @@+module TestSupport+  ( withTestConnection+  , withSchema+  , insertTestUsers+  , stmtListAll+  , stmtSelectOne+  ) where++import Data.ByteString (ByteString)+import Data.ByteString.Char8 qualified as BS8+import Data.Int (Int32)+import Data.Text (Text)+import Valiant+import System.Environment (lookupEnv)++requireDatabaseUrl :: IO ByteString+requireDatabaseUrl = do+  mUrl <- lookupEnv "DATABASE_URL"+  case mUrl of+    Just url -> pure (BS8.pack url)+    Nothing -> error "DATABASE_URL is not set."++withTestConnection :: (Connection -> IO a) -> IO a+withTestConnection action = do+  url <- requireDatabaseUrl+  conn <- connectString url+  result <- action conn+  close conn+  pure result++withSchema :: Connection -> IO a -> IO a+withSchema conn action = do+  _ <- simpleQuery conn "DROP TABLE IF EXISTS users_conduit CASCADE"+  _ <- simpleQuery conn+    "CREATE TABLE users_conduit (\+    \  id SERIAL PRIMARY KEY,\+    \  name TEXT NOT NULL,\+    \  email TEXT\+    \)"+  result <- action+  _ <- simpleQuery conn "DROP TABLE IF EXISTS users_conduit CASCADE"+  pure result++insertTestUsers :: Connection -> IO ()+insertTestUsers conn = do+  _ <- simpleQuery conn+    "INSERT INTO users_conduit (name, email) VALUES \+    \('Alice', 'alice@example.com'),\+    \('Bob', 'bob@example.com'),\+    \('Carol', NULL),\+    \('Dave', 'dave@example.com'),\+    \('Eve', 'eve@example.com')"+  pure ()++stmtListAll :: Statement () (Int32, Text)+stmtListAll = mkStatement+  "SELECT id, name FROM users_conduit ORDER BY id"+  [] ["id", "name"] "<test>"++stmtSelectOne :: Statement Int32 (Int32, Text, Maybe Text)+stmtSelectOne = mkStatement+  "SELECT id, name, email FROM users_conduit WHERE id = $1"+  [23] ["id", "name", "email"] "<test>"

+ valiant-conduit.cabal view

@@ -0,0 +1,77 @@+cabal-version:   3.0+name:            valiant-conduit+version:         0.1.0.0+synopsis:        Conduit streaming adapter for valiant+description:+  Stream PostgreSQL query results as Conduit sources. Provides cursor-based+  streaming via @selectSource@ and single-shot streaming via @foldSource@,+  both producing @ConduitT () r IO ()@ values compatible with the conduit+  ecosystem.+license:         BSD-3-Clause+license-file:    LICENSE+author:          Josh Burgess+maintainer:      joshburgess.webdev@gmail.com+category:        Database+homepage:        https://github.com/joshburgess/valiant+bug-reports:     https://github.com/joshburgess/valiant/issues+build-type:      Simple+extra-doc-files:+  README.md+  CHANGELOG.md+tested-with:     GHC ==9.10.3++source-repository head+  type:     git+  location: https://github.com/joshburgess/valiant+  subdir:   adapters/valiant-conduit++flag werror+  description: Enable -Werror for development builds.+  default:     False+  manual:      True++common warnings+  ghc-options: -Wall -Wcompat -Wno-unticked-promoted-constructors -funbox-strict-fields -fspecialise-aggressively+  if flag(werror)+    ghc-options: -Werror++library+  import:           warnings+  hs-source-dirs:   src+  default-language: GHC2021+  default-extensions:+    DerivingStrategies+    OverloadedStrings+    StrictData++  exposed-modules:+    Valiant.Conduit++  build-depends:+    , base            >=4.17 && <5+    , conduit         >=1.3  && <1.4+    , valiant         >=0.1  && <0.2++test-suite valiant-conduit-test+  import:           warnings+  type:             exitcode-stdio-1.0+  hs-source-dirs:   test+  main-is:          Main.hs+  default-language: GHC2021+  default-extensions:+    OverloadedStrings++  ghc-options: -rtsopts "-with-rtsopts=-K8K"++  other-modules:+    TestSupport++  build-depends:+    , base            >=4.17 && <5+    , bytestring      >=0.11 && <0.13+    , conduit         >=1.3  && <1.4+    , hspec           >=2.11 && <2.13+    , valiant+    , valiant-conduit+    , pg-wire+    , text            >=2.0  && <2.2