{-# OPTIONS_GHC -fno-warn-orphans -fno-warn-partial-type-signatures #-}
{-# LANGUAGE PartialTypeSignatures #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE ConstraintKinds #-}
{-# LANGUAGE GeneralizedNewtypeDeriving #-}
{-# LANGUAGE DeriveFunctor #-}
{-# LANGUAGE StandaloneDeriving #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE MultiParamTypeClasses #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE CPP #-}
module Database.Beam.Postgres.Connection
( Pg(..), PgF(..)
, liftIOWithHandle
, runBeamPostgres, runBeamPostgresDebug
, pgRenderSyntax, runPgRowReader, getFields
, withPgDebug
, postgresUriSyntax ) where
import Control.Exception (SomeException(..), throwIO)
import Control.Monad.Base (MonadBase(..))
import Control.Monad.Free.Church
import Control.Monad.IO.Class
import Control.Monad.Trans.Control (MonadBaseControl(..))
import Database.Beam hiding (runDelete, runUpdate, runInsert, insert)
import Database.Beam.Backend.SQL.BeamExtensions
import Database.Beam.Backend.SQL.Row ( FromBackendRowF(..), FromBackendRowM(..)
, BeamRowReadError(..), ColumnParseError(..) )
import Database.Beam.Backend.URI
import Database.Beam.Schema.Tables
import Database.Beam.Postgres.Syntax
import Database.Beam.Postgres.Full
import Database.Beam.Postgres.Types
import qualified Database.PostgreSQL.LibPQ as Pg hiding
(Connection, escapeStringConn, escapeIdentifier, escapeByteaConn, exec)
import qualified Database.PostgreSQL.Simple as Pg
import qualified Database.PostgreSQL.Simple.FromField as Pg
import qualified Database.PostgreSQL.Simple.Internal as Pg
( Field(..), RowParser(..)
, escapeStringConn, escapeIdentifier, escapeByteaConn
, exec, throwResultError )
import qualified Database.PostgreSQL.Simple.Internal as PgI
import qualified Database.PostgreSQL.Simple.Ok as Pg
import qualified Database.PostgreSQL.Simple.Types as Pg (Query(..))
import Control.Monad.Reader
import Control.Monad.State
import qualified Control.Monad.Fail as Fail
import Data.ByteString (ByteString)
import Data.ByteString.Builder (toLazyByteString, byteString)
import qualified Data.ByteString.Lazy as BL
import Data.Maybe (listToMaybe, fromMaybe)
import Data.Proxy
import Data.String
import qualified Data.Text as T
import Data.Text.Encoding (decodeUtf8)
import Data.Typeable (cast)
#if !MIN_VERSION_base(4, 11, 0)
import Data.Semigroup
#endif
import Foreign.C.Types
import Network.URI (uriToString)
data PgStream a = PgStreamDone (Either BeamRowReadError a)
| PgStreamContinue (Maybe PgI.Row -> IO (PgStream a))
postgresUriSyntax :: c Postgres Pg.Connection Pg
-> BeamURIOpeners c
postgresUriSyntax :: forall (c :: * -> * -> (* -> *) -> *).
c Postgres Connection Pg -> BeamURIOpeners c
postgresUriSyntax =
(forall a. Connection -> Pg a -> IO a)
-> String
-> (URI -> IO (Connection, IO ()))
-> c Postgres Connection Pg
-> BeamURIOpeners c
forall hdl (m :: * -> *) (c :: * -> * -> (* -> *) -> *) be.
(forall a. hdl -> m a -> IO a)
-> String
-> (URI -> IO (hdl, IO ()))
-> c be hdl m
-> BeamURIOpeners c
mkUriOpener Connection -> Pg a -> IO a
forall a. Connection -> Pg a -> IO a
runBeamPostgres String
"postgresql:"
(\URI
uri -> do
let pgConnStr :: ByteString
pgConnStr = String -> ByteString
forall a. IsString a => String -> a
fromString ((String -> String) -> URI -> String -> String
uriToString String -> String
forall a. a -> a
id URI
uri String
"")
hdl <- ByteString -> IO Connection
Pg.connectPostgreSQL ByteString
pgConnStr
pure (hdl, Pg.close hdl))
pgRenderSyntax ::
Pg.Connection -> PgSyntax -> IO ByteString
pgRenderSyntax :: Connection -> PgSyntax -> IO ByteString
pgRenderSyntax Connection
conn (PgSyntax PgSyntaxM ()
mkQuery) =
Builder -> ByteString
renderBuilder (Builder -> ByteString) -> IO Builder -> IO ByteString
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> PgSyntaxM () -> forall r. (() -> r) -> (PgSyntaxF r -> r) -> r
forall (f :: * -> *) a.
F f a -> forall r. (a -> r) -> (f r -> r) -> r
runF PgSyntaxM ()
mkQuery () -> Builder -> IO Builder
forall {f :: * -> *} {p} {a}. Applicative f => p -> a -> f a
finish PgSyntaxF (Builder -> IO Builder) -> Builder -> IO Builder
step Builder
forall a. Monoid a => a
mempty
where
renderBuilder :: Builder -> ByteString
renderBuilder = LazyByteString -> ByteString
BL.toStrict (LazyByteString -> ByteString)
-> (Builder -> LazyByteString) -> Builder -> ByteString
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Builder -> LazyByteString
toLazyByteString
step :: PgSyntaxF (Builder -> IO Builder) -> Builder -> IO Builder
step (EmitBuilder Builder
b Builder -> IO Builder
next) Builder
a = Builder -> IO Builder
next (Builder
a Builder -> Builder -> Builder
forall a. Semigroup a => a -> a -> a
<> Builder
b)
step (EmitByteString ByteString
b Builder -> IO Builder
next) Builder
a = Builder -> IO Builder
next (Builder
a Builder -> Builder -> Builder
forall a. Semigroup a => a -> a -> a
<> ByteString -> Builder
byteString ByteString
b)
step (EscapeString ByteString
b Builder -> IO Builder
next) Builder
a = do
res <- String -> IO (Either ByteString ByteString) -> IO ByteString
forall {m :: * -> *} {a} {b}.
(MonadFail m, Show a) =>
String -> m (Either a b) -> m b
wrapError String
"EscapeString" (Connection -> ByteString -> IO (Either ByteString ByteString)
Pg.escapeStringConn Connection
conn ByteString
b)
next (a <> byteString res)
step (EscapeBytea ByteString
b Builder -> IO Builder
next) Builder
a = do
res <- String -> IO (Either ByteString ByteString) -> IO ByteString
forall {m :: * -> *} {a} {b}.
(MonadFail m, Show a) =>
String -> m (Either a b) -> m b
wrapError String
"EscapeBytea" (Connection -> ByteString -> IO (Either ByteString ByteString)
Pg.escapeByteaConn Connection
conn ByteString
b)
next (a <> byteString res)
step (EscapeIdentifier ByteString
b Builder -> IO Builder
next) Builder
a = do
res <- String -> IO (Either ByteString ByteString) -> IO ByteString
forall {m :: * -> *} {a} {b}.
(MonadFail m, Show a) =>
String -> m (Either a b) -> m b
wrapError String
"EscapeIdentifier" (Connection -> ByteString -> IO (Either ByteString ByteString)
Pg.escapeIdentifier Connection
conn ByteString
b)
next (a <> byteString res)
finish :: p -> a -> f a
finish p
_ = a -> f a
forall a. a -> f a
forall (f :: * -> *) a. Applicative f => a -> f a
pure
wrapError :: String -> m (Either a b) -> m b
wrapError String
step' m (Either a b)
go = do
res <- m (Either a b)
go
case res of
Right b
res' -> b -> m b
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure b
res'
Left a
res' -> String -> m b
forall a. String -> m a
forall (m :: * -> *) a. MonadFail m => String -> m a
fail (String
step' String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
": " String -> String -> String
forall a. Semigroup a => a -> a -> a
<> a -> String
forall a. Show a => a -> String
show a
res')
getFields :: Pg.Result -> IO [Pg.Field]
getFields :: Result -> IO [Field]
getFields Result
res = do
Pg.Col colCount <- Result -> IO Column
Pg.nfields Result
res
let getField CInt
col =
Result -> Column -> Oid -> Field
Pg.Field Result
res (CInt -> Column
Pg.Col CInt
col) (Oid -> Field) -> IO Oid -> IO Field
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Result -> Column -> IO Oid
Pg.ftype Result
res (CInt -> Column
Pg.Col CInt
col)
mapM getField [0..colCount - 1]
runPgRowReader ::
Pg.Connection -> Pg.Row -> Pg.Result -> [Pg.Field] -> FromBackendRowM Postgres a -> IO (Either BeamRowReadError a)
runPgRowReader :: forall a.
Connection
-> Row
-> Result
-> [Field]
-> FromBackendRowM Postgres a
-> IO (Either BeamRowReadError a)
runPgRowReader Connection
conn Row
rowIdx Result
res [Field]
fields (FromBackendRowM F (FromBackendRowF Postgres) a
readRow) =
Result -> IO Column
Pg.nfields Result
res IO Column
-> (Column -> IO (Either BeamRowReadError a))
-> IO (Either BeamRowReadError a)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \(Pg.Col CInt
colCount) ->
F (FromBackendRowF Postgres) a
-> forall r. (a -> r) -> (FromBackendRowF Postgres r -> r) -> r
forall (f :: * -> *) a.
F f a -> forall r. (a -> r) -> (f r -> r) -> r
runF F (FromBackendRowF Postgres) a
readRow a -> CInt -> CInt -> [Field] -> IO (Either BeamRowReadError a)
forall {f :: * -> *} {b} {p} {p} {p} {a}.
Applicative f =>
b -> p -> p -> p -> f (Either a b)
finish FromBackendRowF
Postgres
(CInt -> CInt -> [Field] -> IO (Either BeamRowReadError a))
-> CInt -> CInt -> [Field] -> IO (Either BeamRowReadError a)
forall x.
FromBackendRowF
Postgres
(CInt -> CInt -> [Field] -> IO (Either BeamRowReadError x))
-> CInt -> CInt -> [Field] -> IO (Either BeamRowReadError x)
step CInt
0 CInt
colCount [Field]
fields
where
step :: forall x. FromBackendRowF Postgres (CInt -> CInt -> [PgI.Field] -> IO (Either BeamRowReadError x))
-> CInt -> CInt -> [PgI.Field] -> IO (Either BeamRowReadError x)
step :: forall x.
FromBackendRowF
Postgres
(CInt -> CInt -> [Field] -> IO (Either BeamRowReadError x))
-> CInt -> CInt -> [Field] -> IO (Either BeamRowReadError x)
step (ParseOneField a -> CInt -> CInt -> [Field] -> IO (Either BeamRowReadError x)
_) CInt
curCol CInt
colCount [] = Either BeamRowReadError x -> IO (Either BeamRowReadError x)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (BeamRowReadError -> Either BeamRowReadError x
forall a b. a -> Either a b
Left (Maybe Int -> ColumnParseError -> BeamRowReadError
BeamRowReadError (Int -> Maybe Int
forall a. a -> Maybe a
Just (CInt -> Int
forall a b. (Integral a, Num b) => a -> b
fromIntegral CInt
curCol)) (Int -> ColumnParseError
ColumnNotEnoughColumns (CInt -> Int
forall a b. (Integral a, Num b) => a -> b
fromIntegral CInt
colCount))))
step (ParseOneField a -> CInt -> CInt -> [Field] -> IO (Either BeamRowReadError x)
_) CInt
curCol CInt
colCount [Field]
_
| CInt
curCol CInt -> CInt -> Bool
forall a. Ord a => a -> a -> Bool
>= CInt
colCount = Either BeamRowReadError x -> IO (Either BeamRowReadError x)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (BeamRowReadError -> Either BeamRowReadError x
forall a b. a -> Either a b
Left (Maybe Int -> ColumnParseError -> BeamRowReadError
BeamRowReadError (Int -> Maybe Int
forall a. a -> Maybe a
Just (CInt -> Int
forall a b. (Integral a, Num b) => a -> b
fromIntegral CInt
curCol)) (Int -> ColumnParseError
ColumnNotEnoughColumns (CInt -> Int
forall a b. (Integral a, Num b) => a -> b
fromIntegral CInt
colCount))))
step (ParseOneField (a -> CInt -> CInt -> [Field] -> IO (Either BeamRowReadError x)
next' :: next -> _)) CInt
curCol CInt
colCount (Field
field:[Field]
remainingFields) =
do fieldValue <- Result -> Row -> Column -> IO (Maybe ByteString)
Pg.getvalue' Result
res Row
rowIdx (CInt -> Column
Pg.Col CInt
curCol)
res' <- Pg.runConversion (Pg.fromField field fieldValue) conn
case res' of
Pg.Errors [SomeException]
errs ->
let err :: ColumnParseError
err = ColumnParseError -> Maybe ColumnParseError -> ColumnParseError
forall a. a -> Maybe a -> a
fromMaybe (String -> ColumnParseError
ColumnErrorInternal String
"Column parse failed with unknown exception") (Maybe ColumnParseError -> ColumnParseError)
-> Maybe ColumnParseError -> ColumnParseError
forall a b. (a -> b) -> a -> b
$
[ColumnParseError] -> Maybe ColumnParseError
forall a. [a] -> Maybe a
listToMaybe ([ColumnParseError] -> Maybe ColumnParseError)
-> [ColumnParseError] -> Maybe ColumnParseError
forall a b. (a -> b) -> a -> b
$
do SomeException e <- [SomeException]
errs
Just pgErr <- pure (cast e)
case pgErr of
Pg.ConversionFailed { errSQLType :: ResultError -> String
Pg.errSQLType = String
sql
, errHaskellType :: ResultError -> String
Pg.errHaskellType = String
hs
, errMessage :: ResultError -> String
Pg.errMessage = String
msg
, errSQLField :: ResultError -> String
Pg.errSQLField = String
field } ->
ColumnParseError -> [ColumnParseError]
forall a. a -> [a]
forall (f :: * -> *) a. Applicative f => a -> f a
pure (String -> String -> String -> ColumnParseError
ColumnTypeMismatch String
hs String
sql (String
"Conversion failed for field'" String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
field String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
"': " String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
msg))
Pg.Incompatible { errSQLType :: ResultError -> String
Pg.errSQLType = String
sql
, errHaskellType :: ResultError -> String
Pg.errHaskellType = String
hs
, errMessage :: ResultError -> String
Pg.errMessage = String
msg
, errSQLField :: ResultError -> String
Pg.errSQLField = String
field } ->
ColumnParseError -> [ColumnParseError]
forall a. a -> [a]
forall (f :: * -> *) a. Applicative f => a -> f a
pure (String -> String -> String -> ColumnParseError
ColumnTypeMismatch String
hs String
sql (String
"Incompatible field: '" String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
field String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
"': " String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String
msg))
Pg.UnexpectedNull {} ->
ColumnParseError -> [ColumnParseError]
forall a. a -> [a]
forall (f :: * -> *) a. Applicative f => a -> f a
pure ColumnParseError
ColumnUnexpectedNull
in Either BeamRowReadError x -> IO (Either BeamRowReadError x)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (BeamRowReadError -> Either BeamRowReadError x
forall a b. a -> Either a b
Left (Maybe Int -> ColumnParseError -> BeamRowReadError
BeamRowReadError (Int -> Maybe Int
forall a. a -> Maybe a
Just (CInt -> Int
forall a b. (Integral a, Num b) => a -> b
fromIntegral CInt
curCol)) ColumnParseError
err))
Pg.Ok a
x -> a -> CInt -> CInt -> [Field] -> IO (Either BeamRowReadError x)
next' a
x (CInt
curCol CInt -> CInt -> CInt
forall a. Num a => a -> a -> a
+ CInt
1) CInt
colCount [Field]
remainingFields
step (Alt (FromBackendRowM F (FromBackendRowF Postgres) a
a) (FromBackendRowM F (FromBackendRowF Postgres) a
b) a -> CInt -> CInt -> [Field] -> IO (Either BeamRowReadError x)
next) CInt
curCol CInt
colCount [Field]
cols =
do aRes <- F (FromBackendRowF Postgres) a
-> forall r. (a -> r) -> (FromBackendRowF Postgres r -> r) -> r
forall (f :: * -> *) a.
F f a -> forall r. (a -> r) -> (f r -> r) -> r
runF F (FromBackendRowF Postgres) a
a (\a
x CInt
curCol' CInt
colCount' [Field]
cols' -> Either BeamRowReadError (IO (Either BeamRowReadError x))
-> IO (Either BeamRowReadError (IO (Either BeamRowReadError x)))
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (IO (Either BeamRowReadError x)
-> Either BeamRowReadError (IO (Either BeamRowReadError x))
forall a b. b -> Either a b
Right (a -> CInt -> CInt -> [Field] -> IO (Either BeamRowReadError x)
next a
x CInt
curCol' CInt
colCount' [Field]
cols'))) FromBackendRowF
Postgres
(CInt
-> CInt
-> [Field]
-> IO (Either BeamRowReadError (IO (Either BeamRowReadError x))))
-> CInt
-> CInt
-> [Field]
-> IO (Either BeamRowReadError (IO (Either BeamRowReadError x)))
forall x.
FromBackendRowF
Postgres
(CInt -> CInt -> [Field] -> IO (Either BeamRowReadError x))
-> CInt -> CInt -> [Field] -> IO (Either BeamRowReadError x)
step CInt
curCol CInt
colCount [Field]
cols
case aRes of
Right IO (Either BeamRowReadError x)
next' -> IO (Either BeamRowReadError x)
next'
Left BeamRowReadError
aErr -> do
bRes <- F (FromBackendRowF Postgres) a
-> forall r. (a -> r) -> (FromBackendRowF Postgres r -> r) -> r
forall (f :: * -> *) a.
F f a -> forall r. (a -> r) -> (f r -> r) -> r
runF F (FromBackendRowF Postgres) a
b (\a
x CInt
curCol' CInt
colCount' [Field]
cols' -> Either BeamRowReadError (IO (Either BeamRowReadError x))
-> IO (Either BeamRowReadError (IO (Either BeamRowReadError x)))
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (IO (Either BeamRowReadError x)
-> Either BeamRowReadError (IO (Either BeamRowReadError x))
forall a b. b -> Either a b
Right (a -> CInt -> CInt -> [Field] -> IO (Either BeamRowReadError x)
next a
x CInt
curCol' CInt
colCount' [Field]
cols'))) FromBackendRowF
Postgres
(CInt
-> CInt
-> [Field]
-> IO (Either BeamRowReadError (IO (Either BeamRowReadError x))))
-> CInt
-> CInt
-> [Field]
-> IO (Either BeamRowReadError (IO (Either BeamRowReadError x)))
forall x.
FromBackendRowF
Postgres
(CInt -> CInt -> [Field] -> IO (Either BeamRowReadError x))
-> CInt -> CInt -> [Field] -> IO (Either BeamRowReadError x)
step CInt
curCol CInt
colCount [Field]
cols
case bRes of
Right IO (Either BeamRowReadError x)
next' -> IO (Either BeamRowReadError x)
next'
Left {} -> Either BeamRowReadError x -> IO (Either BeamRowReadError x)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (BeamRowReadError -> Either BeamRowReadError x
forall a b. a -> Either a b
Left BeamRowReadError
aErr)
step (FailParseWith BeamRowReadError
err) CInt
_ CInt
_ [Field]
_ =
Either BeamRowReadError x -> IO (Either BeamRowReadError x)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (BeamRowReadError -> Either BeamRowReadError x
forall a b. a -> Either a b
Left BeamRowReadError
err)
finish :: b -> p -> p -> p -> f (Either a b)
finish b
x p
_ p
_ p
_ = Either a b -> f (Either a b)
forall a. a -> f a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (b -> Either a b
forall a b. b -> Either a b
Right b
x)
withPgDebug :: (String -> IO ()) -> Pg.Connection -> Pg a -> IO (Either BeamRowReadError a)
withPgDebug :: forall a.
(String -> IO ())
-> Connection -> Pg a -> IO (Either BeamRowReadError a)
withPgDebug String -> IO ()
dbg Connection
conn (Pg F PgF a
action) =
let finish :: b -> f (Either a b)
finish b
x = Either a b -> f (Either a b)
forall a. a -> f a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (b -> Either a b
forall a b. b -> Either a b
Right b
x)
step :: PgF (IO (Either BeamRowReadError a))
-> IO (Either BeamRowReadError a)
step (PgLiftIO IO a
io a -> IO (Either BeamRowReadError a)
next) = IO a
io IO a
-> (a -> IO (Either BeamRowReadError a))
-> IO (Either BeamRowReadError a)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= a -> IO (Either BeamRowReadError a)
next
step (PgLiftWithHandle (String -> IO ()) -> Connection -> IO a
withConn a -> IO (Either BeamRowReadError a)
next) = (String -> IO ()) -> Connection -> IO a
withConn String -> IO ()
dbg Connection
conn IO a
-> (a -> IO (Either BeamRowReadError a))
-> IO (Either BeamRowReadError a)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= a -> IO (Either BeamRowReadError a)
next
step (PgFetchNext Maybe x -> IO (Either BeamRowReadError a)
next) = Maybe x -> IO (Either BeamRowReadError a)
next Maybe x
forall a. Maybe a
Nothing
step (PgRunReturning FetchMode
CursorBatching
(PgCommandSyntax PgCommandType
PgCommandTypeQuery PgSyntax
syntax)
(Pg (Maybe x) -> Pg a
mkProcess :: Pg (Maybe x) -> Pg a')
a -> IO (Either BeamRowReadError a)
next) =
do query <- Connection -> PgSyntax -> IO ByteString
pgRenderSyntax Connection
conn PgSyntax
syntax
let Pg process = mkProcess (Pg (liftF (PgFetchNext id)))
dbg (T.unpack (decodeUtf8 query))
action' <- runF process finishProcess stepProcess Nothing
case action' of
PgStreamDone (Right a
x) -> Connection -> Query -> IO Int64
Pg.execute_ Connection
conn (ByteString -> Query
Pg.Query ByteString
query) IO Int64
-> IO (Either BeamRowReadError a) -> IO (Either BeamRowReadError a)
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> a -> IO (Either BeamRowReadError a)
next a
x
PgStreamDone (Left BeamRowReadError
err) -> Either BeamRowReadError a -> IO (Either BeamRowReadError a)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (BeamRowReadError -> Either BeamRowReadError a
forall a b. a -> Either a b
Left BeamRowReadError
err)
PgStreamContinue Maybe Row -> IO (PgStream a)
nextStream ->
let finishUp :: PgStream a -> IO (Either BeamRowReadError a)
finishUp (PgStreamDone (Right a
x)) = a -> IO (Either BeamRowReadError a)
next a
x
finishUp (PgStreamDone (Left BeamRowReadError
err)) = Either BeamRowReadError a -> IO (Either BeamRowReadError a)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (BeamRowReadError -> Either BeamRowReadError a
forall a b. a -> Either a b
Left BeamRowReadError
err)
finishUp (PgStreamContinue Maybe Row -> IO (PgStream a)
next') = Maybe Row -> IO (PgStream a)
next' Maybe Row
forall a. Maybe a
Nothing IO (PgStream a)
-> (PgStream a -> IO (Either BeamRowReadError a))
-> IO (Either BeamRowReadError a)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= PgStream a -> IO (Either BeamRowReadError a)
finishUp
columnCount :: Column
columnCount = Int -> Column
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Int -> Column) -> Int -> Column
forall a b. (a -> b) -> a -> b
$ Proxy Postgres -> Proxy x -> Int
forall be a. FromBackendRow be a => Proxy be -> Proxy a -> Int
valuesNeeded (forall t. Proxy t
forall {k} (t :: k). Proxy t
Proxy @Postgres) (forall t. Proxy t
forall {k} (t :: k). Proxy t
Proxy @x)
in RowParser Row
-> Connection
-> Query
-> PgStream a
-> (PgStream a -> Row -> IO (PgStream a))
-> IO (PgStream a)
forall r a.
RowParser r -> Connection -> Query -> a -> (a -> r -> IO a) -> IO a
Pg.foldWith_ (ReaderT Row (StateT Column Conversion) Row -> RowParser Row
forall a. ReaderT Row (StateT Column Conversion) a -> RowParser a
Pg.RP (Column -> ReaderT Row (StateT Column Conversion) ()
forall s (m :: * -> *). MonadState s m => s -> m ()
put Column
columnCount ReaderT Row (StateT Column Conversion) ()
-> ReaderT Row (StateT Column Conversion) Row
-> ReaderT Row (StateT Column Conversion) Row
forall a b.
ReaderT Row (StateT Column Conversion) a
-> ReaderT Row (StateT Column Conversion) b
-> ReaderT Row (StateT Column Conversion) b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> ReaderT Row (StateT Column Conversion) Row
forall r (m :: * -> *). MonadReader r m => m r
ask)) Connection
conn (ByteString -> Query
Pg.Query ByteString
query) ((Maybe Row -> IO (PgStream a)) -> PgStream a
forall a. (Maybe Row -> IO (PgStream a)) -> PgStream a
PgStreamContinue Maybe Row -> IO (PgStream a)
nextStream) PgStream a -> Row -> IO (PgStream a)
forall a. PgStream a -> Row -> IO (PgStream a)
runConsumer IO (PgStream a)
-> (PgStream a -> IO (Either BeamRowReadError a))
-> IO (Either BeamRowReadError a)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= PgStream a -> IO (Either BeamRowReadError a)
finishUp
step (PgRunReturning FetchMode
AtOnce
(PgCommandSyntax PgCommandType
PgCommandTypeQuery PgSyntax
syntax)
(Pg (Maybe x) -> Pg a
mkProcess :: Pg (Maybe x) -> Pg a')
a -> IO (Either BeamRowReadError a)
next) =
ByteString
-> PgSyntax
-> (Pg (Maybe x) -> Pg a)
-> (a -> IO (Either BeamRowReadError a))
-> IO (Either BeamRowReadError a)
forall x a' {a}.
FromBackendRow Postgres x =>
ByteString
-> PgSyntax
-> (Pg (Maybe x) -> Pg a')
-> (a' -> IO (Either BeamRowReadError a))
-> IO (Either BeamRowReadError a)
renderExecReturningList ByteString
"No tuples returned to Postgres query" PgSyntax
syntax Pg (Maybe x) -> Pg a
mkProcess a -> IO (Either BeamRowReadError a)
next
step (PgRunReturning FetchMode
_ (PgCommandSyntax PgCommandType
PgCommandTypeDataUpdateReturning PgSyntax
syntax) Pg (Maybe x) -> Pg a
mkProcess a -> IO (Either BeamRowReadError a)
next) =
ByteString
-> PgSyntax
-> (Pg (Maybe x) -> Pg a)
-> (a -> IO (Either BeamRowReadError a))
-> IO (Either BeamRowReadError a)
forall x a' {a}.
FromBackendRow Postgres x =>
ByteString
-> PgSyntax
-> (Pg (Maybe x) -> Pg a')
-> (a' -> IO (Either BeamRowReadError a))
-> IO (Either BeamRowReadError a)
renderExecReturningList ByteString
"No tuples returned to Postgres update/insert returning" PgSyntax
syntax Pg (Maybe x) -> Pg a
mkProcess a -> IO (Either BeamRowReadError a)
next
step (PgRunReturning FetchMode
_ (PgCommandSyntax PgCommandType
_ PgSyntax
syntax) Pg (Maybe x) -> Pg a
mkProcess a -> IO (Either BeamRowReadError a)
next) =
do query <- Connection -> PgSyntax -> IO ByteString
pgRenderSyntax Connection
conn PgSyntax
syntax
dbg (T.unpack (decodeUtf8 query))
_ <- Pg.execute_ conn (Pg.Query query)
let Pg process = mkProcess (Pg (liftF (PgFetchNext id)))
runF process next stepReturningNone
renderExecReturningList :: (FromBackendRow Postgres x) => _ -> PgSyntax -> (Pg (Maybe x) -> Pg a') -> _ -> _
renderExecReturningList :: ByteString
-> PgSyntax
-> (Pg (Maybe x) -> Pg a')
-> (a' -> IO (Either BeamRowReadError a))
-> IO (Either BeamRowReadError a)
renderExecReturningList ByteString
errMsg PgSyntax
syntax Pg (Maybe x) -> Pg a'
mkProcess a' -> IO (Either BeamRowReadError a)
next =
do query <- Connection -> PgSyntax -> IO ByteString
pgRenderSyntax Connection
conn PgSyntax
syntax
dbg (T.unpack (decodeUtf8 query))
res <- Pg.exec conn query
sts <- Pg.resultStatus res
case sts of
ExecStatus
Pg.TuplesOk -> do
let Pg F PgF a'
process = Pg (Maybe x) -> Pg a'
mkProcess (F PgF (Maybe x) -> Pg (Maybe x)
forall a. F PgF a -> Pg a
Pg (PgF (Maybe x) -> F PgF (Maybe x)
forall (f :: * -> *) (m :: * -> *) a.
(Functor f, MonadFree f m) =>
f a -> m a
liftF ((Maybe x -> Maybe x) -> PgF (Maybe x)
forall a next.
FromBackendRow Postgres a =>
(Maybe a -> next) -> PgF next
PgFetchNext Maybe x -> Maybe x
forall a. a -> a
id)))
F PgF a' -> forall r. (a' -> r) -> (PgF r -> r) -> r
forall (f :: * -> *) a.
F f a -> forall r. (a -> r) -> (f r -> r) -> r
runF F PgF a'
process (\a'
x CInt
_ -> Result -> IO ()
Pg.unsafeFreeResult Result
res IO ()
-> IO (Either BeamRowReadError a) -> IO (Either BeamRowReadError a)
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> a' -> IO (Either BeamRowReadError a)
next a'
x) (Result
-> PgF (CInt -> IO (Either BeamRowReadError a))
-> CInt
-> IO (Either BeamRowReadError a)
forall a.
Result
-> PgF (CInt -> IO (Either BeamRowReadError a))
-> CInt
-> IO (Either BeamRowReadError a)
stepReturningList Result
res) CInt
0
ExecStatus
_ -> ByteString
-> Result -> ExecStatus -> IO (Either BeamRowReadError a)
forall a. ByteString -> Result -> ExecStatus -> IO a
Pg.throwResultError ByteString
errMsg Result
res ExecStatus
sts
stepReturningNone :: forall a. PgF (IO (Either BeamRowReadError a)) -> IO (Either BeamRowReadError a)
stepReturningNone :: forall a.
PgF (IO (Either BeamRowReadError a))
-> IO (Either BeamRowReadError a)
stepReturningNone (PgLiftIO IO a
action' a -> IO (Either BeamRowReadError a)
next) = IO a
action' IO a
-> (a -> IO (Either BeamRowReadError a))
-> IO (Either BeamRowReadError a)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= a -> IO (Either BeamRowReadError a)
next
stepReturningNone (PgLiftWithHandle (String -> IO ()) -> Connection -> IO a
withConn a -> IO (Either BeamRowReadError a)
next) = (String -> IO ()) -> Connection -> IO a
withConn String -> IO ()
dbg Connection
conn IO a
-> (a -> IO (Either BeamRowReadError a))
-> IO (Either BeamRowReadError a)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= a -> IO (Either BeamRowReadError a)
next
stepReturningNone (PgFetchNext Maybe x -> IO (Either BeamRowReadError a)
next) = Maybe x -> IO (Either BeamRowReadError a)
next Maybe x
forall a. Maybe a
Nothing
stepReturningNone (PgRunReturning {}) = Either BeamRowReadError a -> IO (Either BeamRowReadError a)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (BeamRowReadError -> Either BeamRowReadError a
forall a b. a -> Either a b
Left (Maybe Int -> ColumnParseError -> BeamRowReadError
BeamRowReadError Maybe Int
forall a. Maybe a
Nothing (String -> ColumnParseError
ColumnErrorInternal String
"Nested queries not allowed")))
stepReturningList :: forall a. Pg.Result -> PgF (CInt -> IO (Either BeamRowReadError a)) -> CInt -> IO (Either BeamRowReadError a)
stepReturningList :: forall a.
Result
-> PgF (CInt -> IO (Either BeamRowReadError a))
-> CInt
-> IO (Either BeamRowReadError a)
stepReturningList Result
_ (PgLiftIO IO a
action' a -> CInt -> IO (Either BeamRowReadError a)
next) CInt
rowIdx = IO a
action' IO a
-> (a -> IO (Either BeamRowReadError a))
-> IO (Either BeamRowReadError a)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \a
x -> a -> CInt -> IO (Either BeamRowReadError a)
next a
x CInt
rowIdx
stepReturningList Result
res (PgFetchNext Maybe x -> CInt -> IO (Either BeamRowReadError a)
next) CInt
rowIdx =
do fields <- Result -> IO [Field]
getFields Result
res
Pg.Row rowCount <- Pg.ntuples res
if rowIdx >= rowCount
then next Nothing rowIdx
else runPgRowReader conn (Pg.Row rowIdx) res fields fromBackendRow >>= \case
Left BeamRowReadError
err -> Either BeamRowReadError a -> IO (Either BeamRowReadError a)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (BeamRowReadError -> Either BeamRowReadError a
forall a b. a -> Either a b
Left BeamRowReadError
err)
Right x
r -> Maybe x -> CInt -> IO (Either BeamRowReadError a)
next (x -> Maybe x
forall a. a -> Maybe a
Just x
r) (CInt
rowIdx CInt -> CInt -> CInt
forall a. Num a => a -> a -> a
+ CInt
1)
stepReturningList Result
_ (PgRunReturning {}) CInt
_ = Either BeamRowReadError a -> IO (Either BeamRowReadError a)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (BeamRowReadError -> Either BeamRowReadError a
forall a b. a -> Either a b
Left (Maybe Int -> ColumnParseError -> BeamRowReadError
BeamRowReadError Maybe Int
forall a. Maybe a
Nothing (String -> ColumnParseError
ColumnErrorInternal String
"Nested queries not allowed")))
stepReturningList Result
_ (PgLiftWithHandle {}) CInt
_ = Either BeamRowReadError a -> IO (Either BeamRowReadError a)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (BeamRowReadError -> Either BeamRowReadError a
forall a b. a -> Either a b
Left (Maybe Int -> ColumnParseError -> BeamRowReadError
BeamRowReadError Maybe Int
forall a. Maybe a
Nothing (String -> ColumnParseError
ColumnErrorInternal String
"Nested queries not allowed")))
finishProcess :: forall a. a -> Maybe PgI.Row -> IO (PgStream a)
finishProcess :: forall a. a -> Maybe Row -> IO (PgStream a)
finishProcess a
x Maybe Row
_ = PgStream a -> IO (PgStream a)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Either BeamRowReadError a -> PgStream a
forall a. Either BeamRowReadError a -> PgStream a
PgStreamDone (a -> Either BeamRowReadError a
forall a b. b -> Either a b
Right a
x))
stepProcess :: forall a. PgF (Maybe PgI.Row -> IO (PgStream a)) -> Maybe PgI.Row -> IO (PgStream a)
stepProcess :: forall a.
PgF (Maybe Row -> IO (PgStream a)) -> Maybe Row -> IO (PgStream a)
stepProcess (PgLiftIO IO a
action' a -> Maybe Row -> IO (PgStream a)
next) Maybe Row
row = IO a
action' IO a -> (a -> IO (PgStream a)) -> IO (PgStream a)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= (a -> Maybe Row -> IO (PgStream a))
-> Maybe Row -> a -> IO (PgStream a)
forall a b c. (a -> b -> c) -> b -> a -> c
flip a -> Maybe Row -> IO (PgStream a)
next Maybe Row
row
stepProcess (PgFetchNext Maybe x -> Maybe Row -> IO (PgStream a)
next) Maybe Row
Nothing =
PgStream a -> IO (PgStream a)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (PgStream a -> IO (PgStream a))
-> ((Maybe Row -> IO (PgStream a)) -> PgStream a)
-> (Maybe Row -> IO (PgStream a))
-> IO (PgStream a)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Maybe Row -> IO (PgStream a)) -> PgStream a
forall a. (Maybe Row -> IO (PgStream a)) -> PgStream a
PgStreamContinue ((Maybe Row -> IO (PgStream a)) -> IO (PgStream a))
-> (Maybe Row -> IO (PgStream a)) -> IO (PgStream a)
forall a b. (a -> b) -> a -> b
$ \Maybe Row
res ->
case Maybe Row
res of
Maybe Row
Nothing -> Maybe x -> Maybe Row -> IO (PgStream a)
next Maybe x
forall a. Maybe a
Nothing Maybe Row
forall a. Maybe a
Nothing
Just (PgI.Row Row
rowIdx Result
res') ->
Result -> IO [Field]
getFields Result
res' IO [Field] -> ([Field] -> IO (PgStream a)) -> IO (PgStream a)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \[Field]
fields ->
Connection
-> Row
-> Result
-> [Field]
-> FromBackendRowM Postgres x
-> IO (Either BeamRowReadError x)
forall a.
Connection
-> Row
-> Result
-> [Field]
-> FromBackendRowM Postgres a
-> IO (Either BeamRowReadError a)
runPgRowReader Connection
conn Row
rowIdx Result
res' [Field]
fields FromBackendRowM Postgres x
forall be a. FromBackendRow be a => FromBackendRowM be a
fromBackendRow IO (Either BeamRowReadError x)
-> (Either BeamRowReadError x -> IO (PgStream a))
-> IO (PgStream a)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
Left BeamRowReadError
err -> PgStream a -> IO (PgStream a)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Either BeamRowReadError a -> PgStream a
forall a. Either BeamRowReadError a -> PgStream a
PgStreamDone (BeamRowReadError -> Either BeamRowReadError a
forall a b. a -> Either a b
Left BeamRowReadError
err))
Right x
r -> Maybe x -> Maybe Row -> IO (PgStream a)
next (x -> Maybe x
forall a. a -> Maybe a
Just x
r) Maybe Row
forall a. Maybe a
Nothing
stepProcess (PgFetchNext Maybe x -> Maybe Row -> IO (PgStream a)
next) (Just (PgI.Row Row
rowIdx Result
res)) =
Result -> IO [Field]
getFields Result
res IO [Field] -> ([Field] -> IO (PgStream a)) -> IO (PgStream a)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \[Field]
fields ->
Connection
-> Row
-> Result
-> [Field]
-> FromBackendRowM Postgres x
-> IO (Either BeamRowReadError x)
forall a.
Connection
-> Row
-> Result
-> [Field]
-> FromBackendRowM Postgres a
-> IO (Either BeamRowReadError a)
runPgRowReader Connection
conn Row
rowIdx Result
res [Field]
fields FromBackendRowM Postgres x
forall be a. FromBackendRow be a => FromBackendRowM be a
fromBackendRow IO (Either BeamRowReadError x)
-> (Either BeamRowReadError x -> IO (PgStream a))
-> IO (PgStream a)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
Left BeamRowReadError
err -> PgStream a -> IO (PgStream a)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Either BeamRowReadError a -> PgStream a
forall a. Either BeamRowReadError a -> PgStream a
PgStreamDone (BeamRowReadError -> Either BeamRowReadError a
forall a b. a -> Either a b
Left BeamRowReadError
err))
Right x
r -> PgStream a -> IO (PgStream a)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ((Maybe Row -> IO (PgStream a)) -> PgStream a
forall a. (Maybe Row -> IO (PgStream a)) -> PgStream a
PgStreamContinue (Maybe x -> Maybe Row -> IO (PgStream a)
next (x -> Maybe x
forall a. a -> Maybe a
Just x
r)))
stepProcess (PgRunReturning {}) Maybe Row
_ = PgStream a -> IO (PgStream a)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Either BeamRowReadError a -> PgStream a
forall a. Either BeamRowReadError a -> PgStream a
PgStreamDone (BeamRowReadError -> Either BeamRowReadError a
forall a b. a -> Either a b
Left (Maybe Int -> ColumnParseError -> BeamRowReadError
BeamRowReadError Maybe Int
forall a. Maybe a
Nothing (String -> ColumnParseError
ColumnErrorInternal String
"Nested queries not allowed"))))
stepProcess (PgLiftWithHandle (String -> IO ()) -> Connection -> IO a
_ a -> Maybe Row -> IO (PgStream a)
_) Maybe Row
_ = PgStream a -> IO (PgStream a)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Either BeamRowReadError a -> PgStream a
forall a. Either BeamRowReadError a -> PgStream a
PgStreamDone (BeamRowReadError -> Either BeamRowReadError a
forall a b. a -> Either a b
Left (Maybe Int -> ColumnParseError -> BeamRowReadError
BeamRowReadError Maybe Int
forall a. Maybe a
Nothing (String -> ColumnParseError
ColumnErrorInternal String
"Nested queries not allowed"))))
runConsumer :: forall a. PgStream a -> PgI.Row -> IO (PgStream a)
runConsumer :: forall a. PgStream a -> Row -> IO (PgStream a)
runConsumer s :: PgStream a
s@(PgStreamDone {}) Row
_ = PgStream a -> IO (PgStream a)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure PgStream a
s
runConsumer (PgStreamContinue Maybe Row -> IO (PgStream a)
next) Row
row = Maybe Row -> IO (PgStream a)
next (Row -> Maybe Row
forall a. a -> Maybe a
Just Row
row)
in F PgF a -> forall r. (a -> r) -> (PgF r -> r) -> r
forall (f :: * -> *) a.
F f a -> forall r. (a -> r) -> (f r -> r) -> r
runF F PgF a
action a -> IO (Either BeamRowReadError a)
forall {f :: * -> *} {b} {a}. Applicative f => b -> f (Either a b)
finish PgF (IO (Either BeamRowReadError a))
-> IO (Either BeamRowReadError a)
step
data PgF next where
PgLiftIO :: IO a -> (a -> next) -> PgF next
PgRunReturning ::
FromBackendRow Postgres x =>
FetchMode -> PgCommandSyntax -> (Pg (Maybe x) -> Pg a) -> (a -> next) -> PgF next
PgFetchNext ::
FromBackendRow Postgres x =>
(Maybe x -> next) -> PgF next
PgLiftWithHandle :: ((String -> IO ()) -> Pg.Connection -> IO a) -> (a -> next) -> PgF next
instance Functor PgF where
fmap :: forall a b. (a -> b) -> PgF a -> PgF b
fmap a -> b
f = \case
PgLiftIO IO a
io a -> a
n -> IO a -> (a -> b) -> PgF b
forall a next. IO a -> (a -> next) -> PgF next
PgLiftIO IO a
io ((a -> b) -> PgF b) -> (a -> b) -> PgF b
forall a b. (a -> b) -> a -> b
$ a -> b
f (a -> b) -> (a -> a) -> a -> b
forall b c a. (b -> c) -> (a -> b) -> a -> c
. a -> a
n
PgRunReturning FetchMode
mode PgCommandSyntax
cmd Pg (Maybe x) -> Pg a
consume a -> a
n -> FetchMode
-> PgCommandSyntax -> (Pg (Maybe x) -> Pg a) -> (a -> b) -> PgF b
forall a a next.
FromBackendRow Postgres a =>
FetchMode
-> PgCommandSyntax
-> (Pg (Maybe a) -> Pg a)
-> (a -> next)
-> PgF next
PgRunReturning FetchMode
mode PgCommandSyntax
cmd Pg (Maybe x) -> Pg a
consume ((a -> b) -> PgF b) -> (a -> b) -> PgF b
forall a b. (a -> b) -> a -> b
$ a -> b
f (a -> b) -> (a -> a) -> a -> b
forall b c a. (b -> c) -> (a -> b) -> a -> c
. a -> a
n
PgFetchNext Maybe x -> a
n -> (Maybe x -> b) -> PgF b
forall a next.
FromBackendRow Postgres a =>
(Maybe a -> next) -> PgF next
PgFetchNext ((Maybe x -> b) -> PgF b) -> (Maybe x -> b) -> PgF b
forall a b. (a -> b) -> a -> b
$ a -> b
f (a -> b) -> (Maybe x -> a) -> Maybe x -> b
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Maybe x -> a
n
PgLiftWithHandle (String -> IO ()) -> Connection -> IO a
withConn a -> a
n -> ((String -> IO ()) -> Connection -> IO a) -> (a -> b) -> PgF b
forall a next.
((String -> IO ()) -> Connection -> IO a)
-> (a -> next) -> PgF next
PgLiftWithHandle (String -> IO ()) -> Connection -> IO a
withConn ((a -> b) -> PgF b) -> (a -> b) -> PgF b
forall a b. (a -> b) -> a -> b
$ a -> b
f (a -> b) -> (a -> a) -> a -> b
forall b c a. (b -> c) -> (a -> b) -> a -> c
. a -> a
n
data FetchMode
= CursorBatching
| AtOnce
newtype Pg a = Pg { forall a. Pg a -> F PgF a
runPg :: F PgF a }
deriving (Applicative Pg
Applicative Pg =>
(forall a b. Pg a -> (a -> Pg b) -> Pg b)
-> (forall a b. Pg a -> Pg b -> Pg b)
-> (forall a. a -> Pg a)
-> Monad Pg
forall a. a -> Pg a
forall a b. Pg a -> Pg b -> Pg b
forall a b. Pg a -> (a -> Pg b) -> Pg b
forall (m :: * -> *).
Applicative m =>
(forall a b. m a -> (a -> m b) -> m b)
-> (forall a b. m a -> m b -> m b)
-> (forall a. a -> m a)
-> Monad m
$c>>= :: forall a b. Pg a -> (a -> Pg b) -> Pg b
>>= :: forall a b. Pg a -> (a -> Pg b) -> Pg b
$c>> :: forall a b. Pg a -> Pg b -> Pg b
>> :: forall a b. Pg a -> Pg b -> Pg b
$creturn :: forall a. a -> Pg a
return :: forall a. a -> Pg a
Monad, Functor Pg
Functor Pg =>
(forall a. a -> Pg a)
-> (forall a b. Pg (a -> b) -> Pg a -> Pg b)
-> (forall a b c. (a -> b -> c) -> Pg a -> Pg b -> Pg c)
-> (forall a b. Pg a -> Pg b -> Pg b)
-> (forall a b. Pg a -> Pg b -> Pg a)
-> Applicative Pg
forall a. a -> Pg a
forall a b. Pg a -> Pg b -> Pg a
forall a b. Pg a -> Pg b -> Pg b
forall a b. Pg (a -> b) -> Pg a -> Pg b
forall a b c. (a -> b -> c) -> Pg a -> Pg b -> Pg c
forall (f :: * -> *).
Functor f =>
(forall a. a -> f a)
-> (forall a b. f (a -> b) -> f a -> f b)
-> (forall a b c. (a -> b -> c) -> f a -> f b -> f c)
-> (forall a b. f a -> f b -> f b)
-> (forall a b. f a -> f b -> f a)
-> Applicative f
$cpure :: forall a. a -> Pg a
pure :: forall a. a -> Pg a
$c<*> :: forall a b. Pg (a -> b) -> Pg a -> Pg b
<*> :: forall a b. Pg (a -> b) -> Pg a -> Pg b
$cliftA2 :: forall a b c. (a -> b -> c) -> Pg a -> Pg b -> Pg c
liftA2 :: forall a b c. (a -> b -> c) -> Pg a -> Pg b -> Pg c
$c*> :: forall a b. Pg a -> Pg b -> Pg b
*> :: forall a b. Pg a -> Pg b -> Pg b
$c<* :: forall a b. Pg a -> Pg b -> Pg a
<* :: forall a b. Pg a -> Pg b -> Pg a
Applicative, (forall a b. (a -> b) -> Pg a -> Pg b)
-> (forall a b. a -> Pg b -> Pg a) -> Functor Pg
forall a b. a -> Pg b -> Pg a
forall a b. (a -> b) -> Pg a -> Pg b
forall (f :: * -> *).
(forall a b. (a -> b) -> f a -> f b)
-> (forall a b. a -> f b -> f a) -> Functor f
$cfmap :: forall a b. (a -> b) -> Pg a -> Pg b
fmap :: forall a b. (a -> b) -> Pg a -> Pg b
$c<$ :: forall a b. a -> Pg b -> Pg a
<$ :: forall a b. a -> Pg b -> Pg a
Functor, MonadFree PgF)
instance Fail.MonadFail Pg where
fail :: forall a. String -> Pg a
fail String
e = IO a -> Pg a
forall a. IO a -> Pg a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (String -> IO a
forall a. String -> IO a
forall (m :: * -> *) a. MonadFail m => String -> m a
Fail.fail (String -> IO a) -> String -> IO a
forall a b. (a -> b) -> a -> b
$ String
"Internal Error with: " String -> String -> String
forall a. Semigroup a => a -> a -> a
<> String -> String
forall a. Show a => a -> String
show String
e)
instance MonadIO Pg where
liftIO :: forall a. IO a -> Pg a
liftIO IO a
x = PgF a -> Pg a
forall (f :: * -> *) (m :: * -> *) a.
(Functor f, MonadFree f m) =>
f a -> m a
liftF (IO a -> (a -> a) -> PgF a
forall a next. IO a -> (a -> next) -> PgF next
PgLiftIO IO a
x a -> a
forall a. a -> a
id)
instance MonadBase IO Pg where
liftBase :: forall a. IO a -> Pg a
liftBase = IO α -> Pg α
forall a. IO a -> Pg a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO
instance MonadBaseControl IO Pg where
type StM Pg a = a
liftBaseWith :: forall a. (RunInBase Pg IO -> IO a) -> Pg a
liftBaseWith RunInBase Pg IO -> IO a
action =
PgF a -> Pg a
forall (f :: * -> *) (m :: * -> *) a.
(Functor f, MonadFree f m) =>
f a -> m a
liftF (((String -> IO ()) -> Connection -> IO a) -> (a -> a) -> PgF a
forall a next.
((String -> IO ()) -> Connection -> IO a)
-> (a -> next) -> PgF next
PgLiftWithHandle (\String -> IO ()
dbg Connection
conn -> RunInBase Pg IO -> IO a
action ((String -> IO ()) -> Connection -> Pg a -> IO a
forall a. (String -> IO ()) -> Connection -> Pg a -> IO a
runBeamPostgresDebug String -> IO ()
dbg Connection
conn)) a -> a
forall a. a -> a
id)
restoreM :: forall a. StM Pg a -> Pg a
restoreM = a -> Pg a
StM Pg a -> Pg a
forall a. a -> Pg a
forall (f :: * -> *) a. Applicative f => a -> f a
pure
liftIOWithHandle :: (Pg.Connection -> IO a) -> Pg a
liftIOWithHandle :: forall a. (Connection -> IO a) -> Pg a
liftIOWithHandle Connection -> IO a
f = PgF a -> Pg a
forall (f :: * -> *) (m :: * -> *) a.
(Functor f, MonadFree f m) =>
f a -> m a
liftF (((String -> IO ()) -> Connection -> IO a) -> (a -> a) -> PgF a
forall a next.
((String -> IO ()) -> Connection -> IO a)
-> (a -> next) -> PgF next
PgLiftWithHandle (\String -> IO ()
_ -> Connection -> IO a
f) a -> a
forall a. a -> a
id)
runBeamPostgresDebug :: (String -> IO ()) -> Pg.Connection -> Pg a -> IO a
runBeamPostgresDebug :: forall a. (String -> IO ()) -> Connection -> Pg a -> IO a
runBeamPostgresDebug String -> IO ()
dbg Connection
conn Pg a
action =
(String -> IO ())
-> Connection -> Pg a -> IO (Either BeamRowReadError a)
forall a.
(String -> IO ())
-> Connection -> Pg a -> IO (Either BeamRowReadError a)
withPgDebug String -> IO ()
dbg Connection
conn Pg a
action IO (Either BeamRowReadError a)
-> (Either BeamRowReadError a -> IO a) -> IO a
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= (BeamRowReadError -> IO a)
-> (a -> IO a) -> Either BeamRowReadError a -> IO a
forall a c b. (a -> c) -> (b -> c) -> Either a b -> c
either BeamRowReadError -> IO a
forall e a. (HasCallStack, Exception e) => e -> IO a
throwIO a -> IO a
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure
runBeamPostgres :: Pg.Connection -> Pg a -> IO a
runBeamPostgres :: forall a. Connection -> Pg a -> IO a
runBeamPostgres = (String -> IO ()) -> Connection -> Pg a -> IO a
forall a. (String -> IO ()) -> Connection -> Pg a -> IO a
runBeamPostgresDebug (\String
_ -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())
instance MonadBeam Postgres Pg where
runReturningMany :: forall x a.
FromBackendRow Postgres x =>
BeamSqlBackendSyntax Postgres -> (Pg (Maybe x) -> Pg a) -> Pg a
runReturningMany BeamSqlBackendSyntax Postgres
cmd Pg (Maybe x) -> Pg a
consume =
PgF a -> Pg a
forall (f :: * -> *) (m :: * -> *) a.
(Functor f, MonadFree f m) =>
f a -> m a
liftF (FetchMode
-> PgCommandSyntax -> (Pg (Maybe x) -> Pg a) -> (a -> a) -> PgF a
forall a a next.
FromBackendRow Postgres a =>
FetchMode
-> PgCommandSyntax
-> (Pg (Maybe a) -> Pg a)
-> (a -> next)
-> PgF next
PgRunReturning FetchMode
CursorBatching BeamSqlBackendSyntax Postgres
PgCommandSyntax
cmd Pg (Maybe x) -> Pg a
consume a -> a
forall a. a -> a
id)
runReturningOne :: forall x.
FromBackendRow Postgres x =>
BeamSqlBackendSyntax Postgres -> Pg (Maybe x)
runReturningOne BeamSqlBackendSyntax Postgres
cmd =
PgF (Maybe x) -> Pg (Maybe x)
forall (f :: * -> *) (m :: * -> *) a.
(Functor f, MonadFree f m) =>
f a -> m a
liftF (FetchMode
-> PgCommandSyntax
-> (Pg (Maybe x) -> Pg (Maybe x))
-> (Maybe x -> Maybe x)
-> PgF (Maybe x)
forall a a next.
FromBackendRow Postgres a =>
FetchMode
-> PgCommandSyntax
-> (Pg (Maybe a) -> Pg a)
-> (a -> next)
-> PgF next
PgRunReturning FetchMode
AtOnce BeamSqlBackendSyntax Postgres
PgCommandSyntax
cmd Pg (Maybe x) -> Pg (Maybe x)
forall {m :: * -> *} {a}. Monad m => m (Maybe a) -> m (Maybe a)
consume Maybe x -> Maybe x
forall a. a -> a
id)
where
consume :: m (Maybe a) -> m (Maybe a)
consume m (Maybe a)
next = do
a <- m (Maybe a)
next
case a of
Maybe a
Nothing -> Maybe a -> m (Maybe a)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Maybe a
forall a. Maybe a
Nothing
Just a
x -> do
a' <- m (Maybe a)
next
case a' of
Maybe a
Nothing -> Maybe a -> m (Maybe a)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (a -> Maybe a
forall a. a -> Maybe a
Just a
x)
Just a
_ -> Maybe a -> m (Maybe a)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Maybe a
forall a. Maybe a
Nothing
runReturningFirst :: forall x.
FromBackendRow Postgres x =>
BeamSqlBackendSyntax Postgres -> Pg (Maybe x)
runReturningFirst BeamSqlBackendSyntax Postgres
cmd =
PgF (Maybe x) -> Pg (Maybe x)
forall (f :: * -> *) (m :: * -> *) a.
(Functor f, MonadFree f m) =>
f a -> m a
liftF (FetchMode
-> PgCommandSyntax
-> (Pg (Maybe x) -> Pg (Maybe x))
-> (Maybe x -> Maybe x)
-> PgF (Maybe x)
forall a a next.
FromBackendRow Postgres a =>
FetchMode
-> PgCommandSyntax
-> (Pg (Maybe a) -> Pg a)
-> (a -> next)
-> PgF next
PgRunReturning FetchMode
AtOnce BeamSqlBackendSyntax Postgres
PgCommandSyntax
cmd Pg (Maybe x) -> Pg (Maybe x)
forall a. a -> a
id Maybe x -> Maybe x
forall a. a -> a
id)
runReturningList :: forall x.
FromBackendRow Postgres x =>
BeamSqlBackendSyntax Postgres -> Pg [x]
runReturningList BeamSqlBackendSyntax Postgres
cmd =
PgF [x] -> Pg [x]
forall (f :: * -> *) (m :: * -> *) a.
(Functor f, MonadFree f m) =>
f a -> m a
liftF (FetchMode
-> PgCommandSyntax
-> (Pg (Maybe x) -> Pg [x])
-> ([x] -> [x])
-> PgF [x]
forall a a next.
FromBackendRow Postgres a =>
FetchMode
-> PgCommandSyntax
-> (Pg (Maybe a) -> Pg a)
-> (a -> next)
-> PgF next
PgRunReturning FetchMode
AtOnce BeamSqlBackendSyntax Postgres
PgCommandSyntax
cmd Pg (Maybe x) -> Pg [x]
forall {m :: * -> *} {a}. Monad m => m (Maybe a) -> m [a]
consume [x] -> [x]
forall a. a -> a
id)
where
consume :: m (Maybe a) -> m [a]
consume m (Maybe a)
next =
let collectM :: ([a] -> [a]) -> m [a]
collectM [a] -> [a]
acc = do
a <- m (Maybe a)
next
case a of
Maybe a
Nothing -> [a] -> m [a]
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ([a] -> [a]
acc [])
Just a
x -> ([a] -> [a]) -> m [a]
collectM ([a] -> [a]
acc ([a] -> [a]) -> ([a] -> [a]) -> [a] -> [a]
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (a
xa -> [a] -> [a]
forall a. a -> [a] -> [a]
:))
in ([a] -> [a]) -> m [a]
collectM [a] -> [a]
forall a. a -> a
id
instance MonadBeamInsertReturning Postgres Pg where
runInsertReturningList :: forall (table :: (* -> *) -> *).
(Beamable table, Projectible Postgres (table (QExpr Postgres ())),
FromBackendRow Postgres (table Identity)) =>
SqlInsert Postgres table -> Pg [table Identity]
runInsertReturningList SqlInsert Postgres table
i = do
let insertReturningCmd' :: PgReturningType
SqlInsert (QExprToIdentity (table (QExpr Postgres ())))
insertReturningCmd' = SqlInsert Postgres table
i SqlInsert Postgres table
-> (table (QExpr Postgres PostgresInaccessible)
-> table (QExpr Postgres ()))
-> PgReturningType
SqlInsert (QExprToIdentity (table (QExpr Postgres ())))
forall (cmd :: * -> ((* -> *) -> *) -> *) (tbl :: (* -> *) -> *) a.
(PgReturning cmd, Beamable tbl, Projectible Postgres a) =>
cmd Postgres tbl
-> (tbl (QExpr Postgres PostgresInaccessible) -> a)
-> PgReturningType cmd (QExprToIdentity a)
forall (tbl :: (* -> *) -> *) a.
(Beamable tbl, Projectible Postgres a) =>
SqlInsert Postgres tbl
-> (tbl (QExpr Postgres PostgresInaccessible) -> a)
-> PgReturningType SqlInsert (QExprToIdentity a)
`returning`
(forall a.
Columnar' (QExpr Postgres PostgresInaccessible) a
-> Columnar' (QExpr Postgres ()) a)
-> table (QExpr Postgres PostgresInaccessible)
-> table (QExpr Postgres ())
forall (table :: (* -> *) -> *) (f :: * -> *) (g :: * -> *).
Beamable table =>
(forall a. Columnar' f a -> Columnar' g a) -> table f -> table g
changeBeamRep (\(Columnar' (QExpr Text -> BeamSqlBackendExpressionSyntax Postgres
s) :: Columnar' (QExpr Postgres PostgresInaccessible) ty) ->
Columnar (QExpr Postgres ()) a -> Columnar' (QExpr Postgres ()) a
forall (f :: * -> *) a. Columnar f a -> Columnar' f a
Columnar' ((Text -> BeamSqlBackendExpressionSyntax Postgres)
-> QGenExpr QValueContext Postgres () a
forall context be s t.
(Text -> BeamSqlBackendExpressionSyntax be)
-> QGenExpr context be s t
QExpr Text -> BeamSqlBackendExpressionSyntax Postgres
s) :: Columnar' (QExpr Postgres ()) ty)
case PgReturningType
SqlInsert (QExprToIdentity (table (QExpr Postgres ())))
insertReturningCmd' of
PgReturningType
SqlInsert (QExprToIdentity (table (QExpr Postgres ())))
PgInsertReturning (QExprToIdentity (table (QExpr Postgres ())))
PgInsertReturningEmpty ->
[table Identity] -> Pg [table Identity]
forall a. a -> Pg a
forall (f :: * -> *) a. Applicative f => a -> f a
pure []
PgInsertReturning PgSyntax
insertReturningCmd ->
BeamSqlBackendSyntax Postgres -> Pg [table Identity]
forall x.
FromBackendRow Postgres x =>
BeamSqlBackendSyntax Postgres -> Pg [x]
forall be (m :: * -> *) x.
(MonadBeam be m, FromBackendRow be x) =>
BeamSqlBackendSyntax be -> m [x]
runReturningList (PgCommandType -> PgSyntax -> PgCommandSyntax
PgCommandSyntax PgCommandType
PgCommandTypeDataUpdateReturning PgSyntax
insertReturningCmd)
instance MonadBeamUpdateReturning Postgres Pg where
runUpdateReturningList :: forall (table :: (* -> *) -> *).
(Beamable table, Projectible Postgres (table (QExpr Postgres ())),
FromBackendRow Postgres (table Identity)) =>
SqlUpdate Postgres table -> Pg [table Identity]
runUpdateReturningList SqlUpdate Postgres table
u = do
let updateReturningCmd' :: PgReturningType
SqlUpdate (QExprToIdentity (table (QExpr Postgres ())))
updateReturningCmd' = SqlUpdate Postgres table
u SqlUpdate Postgres table
-> (table (QExpr Postgres PostgresInaccessible)
-> table (QExpr Postgres ()))
-> PgReturningType
SqlUpdate (QExprToIdentity (table (QExpr Postgres ())))
forall (cmd :: * -> ((* -> *) -> *) -> *) (tbl :: (* -> *) -> *) a.
(PgReturning cmd, Beamable tbl, Projectible Postgres a) =>
cmd Postgres tbl
-> (tbl (QExpr Postgres PostgresInaccessible) -> a)
-> PgReturningType cmd (QExprToIdentity a)
forall (tbl :: (* -> *) -> *) a.
(Beamable tbl, Projectible Postgres a) =>
SqlUpdate Postgres tbl
-> (tbl (QExpr Postgres PostgresInaccessible) -> a)
-> PgReturningType SqlUpdate (QExprToIdentity a)
`returning`
(forall a.
Columnar' (QExpr Postgres PostgresInaccessible) a
-> Columnar' (QExpr Postgres ()) a)
-> table (QExpr Postgres PostgresInaccessible)
-> table (QExpr Postgres ())
forall (table :: (* -> *) -> *) (f :: * -> *) (g :: * -> *).
Beamable table =>
(forall a. Columnar' f a -> Columnar' g a) -> table f -> table g
changeBeamRep (\(Columnar' (QExpr Text -> BeamSqlBackendExpressionSyntax Postgres
s) :: Columnar' (QExpr Postgres PostgresInaccessible) ty) ->
Columnar (QExpr Postgres ()) a -> Columnar' (QExpr Postgres ()) a
forall (f :: * -> *) a. Columnar f a -> Columnar' f a
Columnar' ((Text -> BeamSqlBackendExpressionSyntax Postgres)
-> QGenExpr QValueContext Postgres () a
forall context be s t.
(Text -> BeamSqlBackendExpressionSyntax be)
-> QGenExpr context be s t
QExpr Text -> BeamSqlBackendExpressionSyntax Postgres
s) :: Columnar' (QExpr Postgres ()) ty)
case PgReturningType
SqlUpdate (QExprToIdentity (table (QExpr Postgres ())))
updateReturningCmd' of
PgReturningType
SqlUpdate (QExprToIdentity (table (QExpr Postgres ())))
PgUpdateReturning (QExprToIdentity (table (QExpr Postgres ())))
PgUpdateReturningEmpty ->
[table Identity] -> Pg [table Identity]
forall a. a -> Pg a
forall (f :: * -> *) a. Applicative f => a -> f a
pure []
PgUpdateReturning PgSyntax
updateReturningCmd ->
BeamSqlBackendSyntax Postgres -> Pg [table Identity]
forall x.
FromBackendRow Postgres x =>
BeamSqlBackendSyntax Postgres -> Pg [x]
forall be (m :: * -> *) x.
(MonadBeam be m, FromBackendRow be x) =>
BeamSqlBackendSyntax be -> m [x]
runReturningList (PgCommandType -> PgSyntax -> PgCommandSyntax
PgCommandSyntax PgCommandType
PgCommandTypeDataUpdateReturning PgSyntax
updateReturningCmd)
instance MonadBeamDeleteReturning Postgres Pg where
runDeleteReturningList :: forall (table :: (* -> *) -> *).
(Beamable table, Projectible Postgres (table (QExpr Postgres ())),
FromBackendRow Postgres (table Identity)) =>
SqlDelete Postgres table -> Pg [table Identity]
runDeleteReturningList SqlDelete Postgres table
d = do
let PgDeleteReturning PgSyntax
deleteReturningCmd = SqlDelete Postgres table
d SqlDelete Postgres table
-> (table (QExpr Postgres PostgresInaccessible)
-> table (QExpr Postgres ()))
-> PgReturningType
SqlDelete (QExprToIdentity (table (QExpr Postgres ())))
forall (cmd :: * -> ((* -> *) -> *) -> *) (tbl :: (* -> *) -> *) a.
(PgReturning cmd, Beamable tbl, Projectible Postgres a) =>
cmd Postgres tbl
-> (tbl (QExpr Postgres PostgresInaccessible) -> a)
-> PgReturningType cmd (QExprToIdentity a)
forall (tbl :: (* -> *) -> *) a.
(Beamable tbl, Projectible Postgres a) =>
SqlDelete Postgres tbl
-> (tbl (QExpr Postgres PostgresInaccessible) -> a)
-> PgReturningType SqlDelete (QExprToIdentity a)
`returning`
(forall a.
Columnar' (QExpr Postgres PostgresInaccessible) a
-> Columnar' (QExpr Postgres ()) a)
-> table (QExpr Postgres PostgresInaccessible)
-> table (QExpr Postgres ())
forall (table :: (* -> *) -> *) (f :: * -> *) (g :: * -> *).
Beamable table =>
(forall a. Columnar' f a -> Columnar' g a) -> table f -> table g
changeBeamRep (\(Columnar' (QExpr Text -> BeamSqlBackendExpressionSyntax Postgres
s) :: Columnar' (QExpr Postgres PostgresInaccessible) ty) ->
Columnar (QExpr Postgres ()) a -> Columnar' (QExpr Postgres ()) a
forall (f :: * -> *) a. Columnar f a -> Columnar' f a
Columnar' ((Text -> BeamSqlBackendExpressionSyntax Postgres)
-> QGenExpr QValueContext Postgres () a
forall context be s t.
(Text -> BeamSqlBackendExpressionSyntax be)
-> QGenExpr context be s t
QExpr Text -> BeamSqlBackendExpressionSyntax Postgres
s) :: Columnar' (QExpr Postgres ()) ty)
BeamSqlBackendSyntax Postgres -> Pg [table Identity]
forall x.
FromBackendRow Postgres x =>
BeamSqlBackendSyntax Postgres -> Pg [x]
forall be (m :: * -> *) x.
(MonadBeam be m, FromBackendRow be x) =>
BeamSqlBackendSyntax be -> m [x]
runReturningList (PgCommandType -> PgSyntax -> PgCommandSyntax
PgCommandSyntax PgCommandType
PgCommandTypeDataUpdateReturning PgSyntax
deleteReturningCmd)