chore(users/Profpatsch): Update postgres module n stuff
Improvements from “upstream”, fresh served. Change-Id: I60e02835730f6a65739eaa729f3e3eed1a0693e6 Reviewed-on: https://cl.tvl.fyi/c/depot/+/9025 Autosubmit: Profpatsch <mail@profpatsch.de> Tested-by: BuildkiteCI Reviewed-by: Profpatsch <mail@profpatsch.de>
This commit is contained in:
parent
fa8288823b
commit
33fa42a1a3
7 changed files with 443 additions and 197 deletions
|
@ -7,8 +7,10 @@ pkgs.haskellPackages.mkDerivation {
|
|||
src = depot.users.Profpatsch.exactSource ./. [
|
||||
./my-prelude.cabal
|
||||
./src/Aeson.hs
|
||||
./src/AtLeast.hs
|
||||
./src/MyPrelude.hs
|
||||
./src/Test.hs
|
||||
./src/Seconds.hs
|
||||
./src/Tool.hs
|
||||
./src/ValidationParseT.hs
|
||||
./src/Postgres/Decoder.hs
|
||||
|
@ -23,7 +25,9 @@ pkgs.haskellPackages.mkDerivation {
|
|||
pkgs.haskellPackages.pa-error-tree
|
||||
pkgs.haskellPackages.pa-json
|
||||
pkgs.haskellPackages.pa-pretty
|
||||
pkgs.haskellPackages.pa-field-parser
|
||||
pkgs.haskellPackages.aeson-better-errors
|
||||
pkgs.haskellPackages.resource-pool
|
||||
pkgs.haskellPackages.error
|
||||
pkgs.haskellPackages.hspec
|
||||
pkgs.haskellPackages.hspec-expectations-pretty-diff
|
||||
|
|
|
@ -57,10 +57,12 @@ library
|
|||
exposed-modules:
|
||||
MyPrelude
|
||||
Aeson
|
||||
AtLeast
|
||||
Test
|
||||
Postgres.Decoder
|
||||
Postgres.MonadPostgres
|
||||
ValidationParseT
|
||||
Seconds
|
||||
Tool
|
||||
|
||||
-- Modules included in this executable, other than Main.
|
||||
|
@ -75,10 +77,15 @@ library
|
|||
, pa-error-tree
|
||||
, pa-json
|
||||
, pa-pretty
|
||||
, pa-field-parser
|
||||
, aeson
|
||||
, aeson-better-errors
|
||||
, bytestring
|
||||
, containers
|
||||
, resource-pool
|
||||
, resourcet
|
||||
, scientific
|
||||
, time
|
||||
, error
|
||||
, exceptions
|
||||
, filepath
|
||||
|
|
51
users/Profpatsch/my-prelude/src/AtLeast.hs
Normal file
51
users/Profpatsch/my-prelude/src/AtLeast.hs
Normal file
|
@ -0,0 +1,51 @@
|
|||
{-# LANGUAGE QuasiQuotes #-}
|
||||
|
||||
module AtLeast where
|
||||
|
||||
import Data.Aeson (FromJSON (parseJSON))
|
||||
import Data.Aeson.BetterErrors qualified as Json
|
||||
import FieldParser (FieldParser)
|
||||
import FieldParser qualified as Field
|
||||
import GHC.Records (HasField (..))
|
||||
import GHC.TypeLits (KnownNat, natVal)
|
||||
import PossehlAnalyticsPrelude
|
||||
( Natural,
|
||||
Proxy (Proxy),
|
||||
fmt,
|
||||
prettyError,
|
||||
(&),
|
||||
)
|
||||
|
||||
-- | A natural number that must be at least as big as the type literal.
|
||||
newtype AtLeast (min :: Natural) num = AtLeast num
|
||||
-- Just use the instances of the wrapped number type
|
||||
deriving newtype (Eq, Show)
|
||||
|
||||
-- | This is the “destructor” for `AtLeast`, because of the phantom type (@min@) it cannot be inferred automatically.
|
||||
instance HasField "unAtLeast" (AtLeast min num) num where
|
||||
getField (AtLeast num) = num
|
||||
|
||||
parseAtLeast ::
|
||||
forall min num.
|
||||
(KnownNat min, Integral num, Show num) =>
|
||||
FieldParser num (AtLeast min num)
|
||||
parseAtLeast =
|
||||
let minInt = natVal (Proxy @min)
|
||||
in Field.FieldParser $ \from ->
|
||||
if from >= (minInt & fromIntegral)
|
||||
then Right (AtLeast from)
|
||||
else Left [fmt|Must be at least {minInt & show} but was {from & show}|]
|
||||
|
||||
instance
|
||||
(KnownNat min, FromJSON num, Integral num, Bounded num, Show num) =>
|
||||
FromJSON (AtLeast min num)
|
||||
where
|
||||
parseJSON =
|
||||
Json.toAesonParser
|
||||
prettyError
|
||||
( do
|
||||
num <- Json.fromAesonParser @_ @num
|
||||
case Field.runFieldParser (parseAtLeast @min @num) num of
|
||||
Left err -> Json.throwCustomError err
|
||||
Right a -> pure a
|
||||
)
|
|
@ -5,6 +5,7 @@ import Data.Aeson qualified as Json
|
|||
import Data.Aeson.BetterErrors qualified as Json
|
||||
import Data.Error.Tree
|
||||
import Data.Typeable (Typeable)
|
||||
import Database.PostgreSQL.Simple (Binary (fromBinary))
|
||||
import Database.PostgreSQL.Simple.FromField qualified as PG
|
||||
import Database.PostgreSQL.Simple.FromRow qualified as PG
|
||||
import Json qualified
|
||||
|
@ -15,6 +16,14 @@ import PossehlAnalyticsPrelude
|
|||
newtype Decoder a = Decoder (PG.RowParser a)
|
||||
deriving newtype (Functor, Applicative, Alternative, Monad)
|
||||
|
||||
-- | Parse a `bytea` field, equivalent to @Binary ByteString@ but avoids the pitfall of having to use 'Binary'.
|
||||
bytea :: Decoder ByteString
|
||||
bytea = fromField @(Binary ByteString) <&> (.fromBinary)
|
||||
|
||||
-- | Parse a nullable `bytea` field, equivalent to @Binary ByteString@ but avoids the pitfall of having to use 'Binary'.
|
||||
byteaMay :: Decoder (Maybe ByteString)
|
||||
byteaMay = fromField @(Maybe (Binary ByteString)) <&> fmap (.fromBinary)
|
||||
|
||||
-- | Turn any type that implements 'PG.fromField' into a 'Decoder'. Use type applications to prevent accidental conversions:
|
||||
--
|
||||
-- @
|
||||
|
@ -56,3 +65,30 @@ json parser = Decoder $ PG.fieldWith $ \field bytes -> do
|
|||
field
|
||||
(err & Json.parseErrorTree "Cannot decode jsonb column" & prettyErrorTree & textToString)
|
||||
Right a -> pure a
|
||||
|
||||
-- | Parse fields out of a nullable json value returned from the database.
|
||||
--
|
||||
-- ATTN: The whole json record has to be transferred before it is parsed,
|
||||
-- so if you only need a tiny bit of it, use `->` and `->>` in your SQL statement
|
||||
-- and return only the fields you need from the query.
|
||||
--
|
||||
-- In that case pay attention to NULL though:
|
||||
--
|
||||
-- @
|
||||
-- SELECT '{"foo": {}}'::jsonb->>'foo' IS NULL
|
||||
-- → TRUE
|
||||
-- @
|
||||
--
|
||||
-- Also note: `->>` will coerce the json value to @text@, regardless of the content.
|
||||
-- So the JSON object @{"foo": {}}"@ would be returned as the text: @"{\"foo\": {}}"@.
|
||||
jsonMay :: Typeable a => Json.ParseT ErrorTree Identity a -> Decoder (Maybe a)
|
||||
jsonMay parser = Decoder $ PG.fieldWith $ \field bytes -> do
|
||||
val <- PG.fromField @(Maybe Json.Value) field bytes
|
||||
case Json.parseValue parser <$> val of
|
||||
Nothing -> pure Nothing
|
||||
Just (Left err) ->
|
||||
PG.returnError
|
||||
PG.ConversionFailed
|
||||
field
|
||||
(err & Json.parseErrorTree "Cannot decode jsonb column" & prettyErrorTree & textToString)
|
||||
Just (Right a) -> pure (Just a)
|
||||
|
|
|
@ -1,36 +1,44 @@
|
|||
{-# LANGUAGE AllowAmbiguousTypes #-}
|
||||
{-# LANGUAGE FunctionalDependencies #-}
|
||||
{-# LANGUAGE DeriveAnyClass #-}
|
||||
{-# LANGUAGE QuasiQuotes #-}
|
||||
{-# LANGUAGE TypeFamilyDependencies #-}
|
||||
{-# LANGUAGE UndecidableInstances #-}
|
||||
{-# LANGUAGE TemplateHaskell #-}
|
||||
{-# OPTIONS_GHC -Wno-orphans #-}
|
||||
|
||||
module Postgres.MonadPostgres where
|
||||
|
||||
import AtLeast (AtLeast)
|
||||
import Control.Exception
|
||||
import Control.Monad.Except
|
||||
import Control.Monad.Logger.CallStack
|
||||
import Control.Monad.Logger (MonadLogger, logDebug, logWarn)
|
||||
import Control.Monad.Reader (MonadReader (ask), ReaderT (..))
|
||||
import Control.Monad.Trans.Resource
|
||||
import Data.Aeson (FromJSON)
|
||||
import Data.Error.Tree
|
||||
import Data.Int (Int64)
|
||||
import Data.Kind (Type)
|
||||
import Data.List qualified as List
|
||||
import Data.Pool (Pool)
|
||||
import Data.Pool qualified as Pool
|
||||
import Data.Text qualified as Text
|
||||
import Data.Typeable (Typeable)
|
||||
import Database.PostgreSQL.Simple (Connection, FormatError, FromRow, Query, QueryError, ResultError, SqlError, ToRow)
|
||||
import Database.PostgreSQL.Simple qualified as PG
|
||||
import Database.PostgreSQL.Simple qualified as Postgres
|
||||
import Database.PostgreSQL.Simple.FromRow qualified as PG
|
||||
import Database.PostgreSQL.Simple.ToField (ToField)
|
||||
import Database.PostgreSQL.Simple.ToRow (ToRow (toRow))
|
||||
import Database.PostgreSQL.Simple.Types (fromQuery)
|
||||
import Database.PostgreSQL.Simple.Types (Query (..))
|
||||
import GHC.Records (HasField (..))
|
||||
import Label
|
||||
import PossehlAnalyticsPrelude
|
||||
import Postgres.Decoder
|
||||
import Postgres.Decoder qualified as Dec
|
||||
import Pretty (showPretty)
|
||||
import Seconds
|
||||
import System.Exit (ExitCode (..))
|
||||
import Tool
|
||||
import UnliftIO (MonadUnliftIO (withRunInIO))
|
||||
import UnliftIO.Process qualified as Process
|
||||
import UnliftIO.Resource qualified as Resource
|
||||
|
||||
-- | Postgres queries/commands that can be executed within a running transaction.
|
||||
--
|
||||
|
@ -38,12 +46,12 @@ import UnliftIO.Process qualified as Process
|
|||
-- and will behave the same unless othewise documented.
|
||||
class Monad m => MonadPostgres (m :: Type -> Type) where
|
||||
-- | Execute an INSERT, UPDATE, or other SQL query that is not expected to return results.
|
||||
--
|
||||
|
||||
-- Returns the number of rows affected.
|
||||
execute :: (ToRow params, Typeable params) => Query -> params -> Transaction m (Label "numberOfRowsAffected" Natural)
|
||||
|
||||
-- | Execute an INSERT, UPDATE, or other SQL query that is not expected to return results. Does not perform parameter substitution.
|
||||
--
|
||||
-- | Execute an INSERT, UPDATE, or other SQL query that is not expected to return results. Does not take parameters.
|
||||
|
||||
-- Returns the number of rows affected.
|
||||
execute_ :: Query -> Transaction m (Label "numberOfRowsAffected" Natural)
|
||||
|
||||
|
@ -170,19 +178,72 @@ newtype Transaction m a = Transaction {unTransaction :: (ReaderT Connection m a)
|
|||
runTransaction' :: Connection -> Transaction m a -> m a
|
||||
runTransaction' conn transaction = runReaderT transaction.unTransaction conn
|
||||
|
||||
-- | [Resource Pool](http://hackage.haskell.org/package/resource-pool-0.2.3.2/docs/Data-Pool.html) configuration.
|
||||
data PoolingInfo = PoolingInfo
|
||||
{ -- | Minimal amount of resources that are
|
||||
-- always available.
|
||||
numberOfStripes :: AtLeast 1 Int,
|
||||
-- | Time after which extra resources
|
||||
-- (above minimum) can stay in the pool
|
||||
-- without being used.
|
||||
unusedResourceOpenTime :: Seconds,
|
||||
-- | Max number of resources that can be
|
||||
-- in the Pool at any time
|
||||
maxOpenResourcesPerStripe :: AtLeast 1 Int
|
||||
}
|
||||
deriving stock (Generic, Eq, Show)
|
||||
deriving anyclass (FromJSON)
|
||||
|
||||
initMonadPostgres ::
|
||||
(Text -> IO ()) ->
|
||||
-- | Info describing the connection to the Postgres DB
|
||||
Postgres.ConnectInfo ->
|
||||
-- | Configuration info for pooling attributes
|
||||
PoolingInfo ->
|
||||
-- | Created Postgres connection pool
|
||||
ResourceT IO (Pool Postgres.Connection)
|
||||
initMonadPostgres logInfoFn connectInfo poolingInfo = do
|
||||
(_releaseKey, connPool) <-
|
||||
Resource.allocate
|
||||
(logInfoFn "Creating Postgres Connection Pool" >> createPGConnPool)
|
||||
(\pool -> logInfoFn "Destroying Postgres Connection Pool" >> destroyPGConnPool pool)
|
||||
pure connPool
|
||||
where
|
||||
-- \| Create a Postgres connection pool
|
||||
createPGConnPool ::
|
||||
IO (Pool Postgres.Connection)
|
||||
createPGConnPool =
|
||||
Pool.createPool
|
||||
poolCreateResource
|
||||
poolfreeResource
|
||||
poolingInfo.numberOfStripes.unAtLeast
|
||||
(poolingInfo.unusedResourceOpenTime & secondsToNominalDiffTime)
|
||||
(poolingInfo.maxOpenResourcesPerStripe.unAtLeast)
|
||||
where
|
||||
poolCreateResource = Postgres.connect connectInfo
|
||||
poolfreeResource = Postgres.close
|
||||
|
||||
-- \| Destroy a Postgres connection pool
|
||||
destroyPGConnPool ::
|
||||
-- \| Pool to be destroyed
|
||||
(Pool Postgres.Connection) ->
|
||||
IO ()
|
||||
destroyPGConnPool p = Pool.destroyAllResources p
|
||||
|
||||
-- | Catch any Postgres exception that gets thrown,
|
||||
-- print the query that was run and the query parameters,
|
||||
-- then rethrow inside an 'Error'.
|
||||
handlePGException ::
|
||||
forall a params m.
|
||||
(ToRow params, MonadUnliftIO m, MonadLogger m, MonadTools m) =>
|
||||
forall a params tools m.
|
||||
(ToRow params, MonadUnliftIO m, MonadLogger m, HasField "pgFormat" tools Tool) =>
|
||||
tools ->
|
||||
Text ->
|
||||
Query ->
|
||||
-- | Depending on whether we used `format` or `formatMany`.
|
||||
Either params [params] ->
|
||||
IO a ->
|
||||
Transaction m a
|
||||
handlePGException queryType query' params io = do
|
||||
handlePGException tools queryType query' params io = do
|
||||
withRunInIO $ \unliftIO ->
|
||||
io
|
||||
`catches` [ Handler $ unliftIO . logQueryException @SqlError,
|
||||
|
@ -197,8 +258,8 @@ handlePGException queryType query' params io = do
|
|||
logQueryException :: Exception e => e -> Transaction m a
|
||||
logQueryException exc = do
|
||||
formattedQuery <- case params of
|
||||
Left one -> pgFormatQuery' query' one
|
||||
Right many -> pgFormatQueryMany' query' many
|
||||
Left one -> pgFormatQuery' tools query' one
|
||||
Right many -> pgFormatQueryMany' tools query' many
|
||||
throwErr
|
||||
( singleError [fmt|Query Type: {queryType}|]
|
||||
:| [ nestedError "Exception" (exc & showPretty & newError & singleError),
|
||||
|
@ -208,27 +269,75 @@ handlePGException queryType query' params io = do
|
|||
logFormatException :: FormatError -> Transaction m a
|
||||
logFormatException fe = throwErr (fe & showPretty & newError & singleError & singleton)
|
||||
|
||||
pgExecute :: (ToRow params, MonadUnliftIO m, MonadLogger m, MonadTools m) => Query -> params -> Transaction m (Label "numberOfRowsAffected" Natural)
|
||||
pgExecute qry params = do
|
||||
-- | Perform a Postgres action within a transaction
|
||||
withPGTransaction ::
|
||||
-- | Postgres connection pool to be used for the action
|
||||
(Pool Postgres.Connection) ->
|
||||
-- | DB-action to be performed
|
||||
(Postgres.Connection -> IO a) ->
|
||||
-- | Result of the DB-action
|
||||
IO a
|
||||
withPGTransaction connPool f =
|
||||
Pool.withResource
|
||||
connPool
|
||||
(\conn -> Postgres.withTransaction conn (f conn))
|
||||
|
||||
runPGTransactionImpl :: MonadUnliftIO m => m (Pool Postgres.Connection) -> Transaction m a -> m a
|
||||
{-# INLINE runPGTransactionImpl #-}
|
||||
runPGTransactionImpl zoom (Transaction transaction) = do
|
||||
pool <- zoom
|
||||
withRunInIO $ \unliftIO ->
|
||||
withPGTransaction pool $ \conn -> do
|
||||
unliftIO $ runReaderT transaction conn
|
||||
|
||||
executeImpl ::
|
||||
(ToRow params, MonadUnliftIO m, MonadLogger m, HasField "pgFormat" tools Tool) =>
|
||||
m tools ->
|
||||
m DebugLogDatabaseQueries ->
|
||||
Query ->
|
||||
params ->
|
||||
Transaction m (Label "numberOfRowsAffected" Natural)
|
||||
{-# INLINE executeImpl #-}
|
||||
executeImpl zoomTools zoomDebugLogDatabaseQueries qry params = do
|
||||
tools <- lift @Transaction zoomTools
|
||||
logDatabaseQueries <- lift @Transaction zoomDebugLogDatabaseQueries
|
||||
logQueryIfEnabled tools logDatabaseQueries qry (HasSingleParam params)
|
||||
conn <- Transaction ask
|
||||
PG.execute conn qry params
|
||||
& handlePGException "execute" qry (Left params)
|
||||
>>= toNumberOfRowsAffected "pgExecute"
|
||||
& handlePGException tools "execute" qry (Left params)
|
||||
>>= toNumberOfRowsAffected "executeImpl"
|
||||
|
||||
pgExecute_ :: (MonadUnliftIO m, MonadLogger m, MonadTools m) => Query -> Transaction m (Label "numberOfRowsAffected" Natural)
|
||||
pgExecute_ qry = do
|
||||
executeImpl_ ::
|
||||
(MonadUnliftIO m, MonadLogger m, HasField "pgFormat" tools Tool) =>
|
||||
m tools ->
|
||||
m DebugLogDatabaseQueries ->
|
||||
Query ->
|
||||
Transaction m (Label "numberOfRowsAffected" Natural)
|
||||
{-# INLINE executeImpl_ #-}
|
||||
executeImpl_ zoomTools zoomDebugLogDatabaseQueries qry = do
|
||||
tools <- lift @Transaction zoomTools
|
||||
logDatabaseQueries <- lift @Transaction zoomDebugLogDatabaseQueries
|
||||
logQueryIfEnabled @() tools logDatabaseQueries qry HasNoParams
|
||||
conn <- Transaction ask
|
||||
PG.execute_ conn qry
|
||||
& handlePGException "execute_" qry (Left ())
|
||||
>>= toNumberOfRowsAffected "pgExecute_"
|
||||
& handlePGException tools "execute_" qry (Left ())
|
||||
>>= toNumberOfRowsAffected "executeImpl_"
|
||||
|
||||
pgExecuteMany :: (ToRow params, MonadUnliftIO m, MonadLogger m, MonadTools m) => Query -> [params] -> Transaction m (Label "numberOfRowsAffected" Natural)
|
||||
pgExecuteMany qry params =
|
||||
do
|
||||
executeManyImpl ::
|
||||
(ToRow params, MonadUnliftIO m, MonadLogger m, HasField "pgFormat" tools Tool) =>
|
||||
m tools ->
|
||||
m DebugLogDatabaseQueries ->
|
||||
Query ->
|
||||
[params] ->
|
||||
Transaction m (Label "numberOfRowsAffected" Natural)
|
||||
executeManyImpl zoomTools zoomDebugLogDatabaseQueries qry params = do
|
||||
tools <- lift @Transaction zoomTools
|
||||
logDatabaseQueries <- lift @Transaction zoomDebugLogDatabaseQueries
|
||||
logQueryIfEnabled tools logDatabaseQueries qry (HasMultiParams params)
|
||||
conn <- Transaction ask
|
||||
PG.executeMany conn qry params
|
||||
& handlePGException "executeMany" qry (Right params)
|
||||
>>= toNumberOfRowsAffected "pgExecuteMany"
|
||||
& handlePGException tools "executeMany" qry (Right params)
|
||||
>>= toNumberOfRowsAffected "executeManyImpl"
|
||||
|
||||
toNumberOfRowsAffected :: MonadIO m => Text -> Int64 -> m (Label "numberOfRowsAffected" Natural)
|
||||
toNumberOfRowsAffected functionName i64 =
|
||||
|
@ -240,23 +349,35 @@ toNumberOfRowsAffected functionName i64 =
|
|||
& liftIO
|
||||
<&> label @"numberOfRowsAffected"
|
||||
|
||||
pgExecuteManyReturningWith :: (ToRow params, MonadUnliftIO m, MonadLogger m, MonadTools m) => Query -> [params] -> Decoder r -> Transaction m [r]
|
||||
pgExecuteManyReturningWith qry params (Decoder fromRow) =
|
||||
do
|
||||
executeManyReturningWithImpl ::
|
||||
(ToRow params, MonadUnliftIO m, MonadLogger m, HasField "pgFormat" tools Tool) =>
|
||||
m tools ->
|
||||
m DebugLogDatabaseQueries ->
|
||||
Query ->
|
||||
[params] ->
|
||||
Decoder r ->
|
||||
Transaction m [r]
|
||||
{-# INLINE executeManyReturningWithImpl #-}
|
||||
executeManyReturningWithImpl zoomTools zoomDebugLogDatabaseQueries qry params (Decoder fromRow) = do
|
||||
tools <- lift @Transaction zoomTools
|
||||
logDatabaseQueries <- lift @Transaction zoomDebugLogDatabaseQueries
|
||||
logQueryIfEnabled tools logDatabaseQueries qry (HasMultiParams params)
|
||||
conn <- Transaction ask
|
||||
PG.returningWith fromRow conn qry params
|
||||
& handlePGException "executeManyReturning" qry (Right params)
|
||||
& handlePGException tools "executeManyReturning" qry (Right params)
|
||||
|
||||
pgFold ::
|
||||
(FromRow row, ToRow params, MonadUnliftIO m, MonadLogger m, MonadTools m) =>
|
||||
foldRowsImpl ::
|
||||
(FromRow row, ToRow params, MonadUnliftIO m, MonadLogger m, HasField "pgFormat" tools Tool) =>
|
||||
m tools ->
|
||||
Query ->
|
||||
params ->
|
||||
a ->
|
||||
(a -> row -> Transaction m a) ->
|
||||
Transaction m a
|
||||
pgFold qry params accumulator f = do
|
||||
{-# INLINE foldRowsImpl #-}
|
||||
foldRowsImpl zoomTools qry params accumulator f = do
|
||||
conn <- Transaction ask
|
||||
|
||||
tools <- lift @Transaction zoomTools
|
||||
withRunInIO
|
||||
( \runInIO ->
|
||||
do
|
||||
|
@ -266,10 +387,18 @@ pgFold qry params accumulator f = do
|
|||
params
|
||||
accumulator
|
||||
(\acc row -> runInIO $ f acc row)
|
||||
& handlePGException "fold" qry (Left params)
|
||||
& handlePGException tools "fold" qry (Left params)
|
||||
& runInIO
|
||||
)
|
||||
|
||||
pgFormatQueryNoParams' ::
|
||||
(MonadIO m, MonadLogger m, HasField "pgFormat" tools Tool) =>
|
||||
tools ->
|
||||
Query ->
|
||||
Transaction m Text
|
||||
pgFormatQueryNoParams' tools q =
|
||||
lift $ pgFormatQueryByteString tools q.fromQuery
|
||||
|
||||
pgFormatQuery :: (ToRow params, MonadIO m) => Query -> params -> Transaction m ByteString
|
||||
pgFormatQuery qry params = Transaction $ do
|
||||
conn <- ask
|
||||
|
@ -280,29 +409,42 @@ pgFormatQueryMany qry params = Transaction $ do
|
|||
conn <- ask
|
||||
liftIO $ PG.formatMany conn qry params
|
||||
|
||||
pgQueryWith :: (ToRow params, MonadUnliftIO m, MonadLogger m, MonadTools m) => Query -> params -> Decoder r -> Transaction m [r]
|
||||
pgQueryWith qry params (Decoder fromRow) = do
|
||||
queryWithImpl ::
|
||||
(ToRow params, MonadUnliftIO m, MonadLogger m, HasField "pgFormat" tools Tool) =>
|
||||
m tools ->
|
||||
m DebugLogDatabaseQueries ->
|
||||
Query ->
|
||||
params ->
|
||||
Decoder r ->
|
||||
Transaction m [r]
|
||||
{-# INLINE queryWithImpl #-}
|
||||
queryWithImpl zoomTools zoomDebugLogDatabaseQueries qry params (Decoder fromRow) = do
|
||||
tools <- lift @Transaction zoomTools
|
||||
logDatabaseQueries <- lift @Transaction zoomDebugLogDatabaseQueries
|
||||
logQueryIfEnabled tools logDatabaseQueries qry (HasSingleParam params)
|
||||
conn <- Transaction ask
|
||||
PG.queryWith fromRow conn qry params
|
||||
& handlePGException "query" qry (Left params)
|
||||
& handlePGException tools "query" qry (Left params)
|
||||
|
||||
pgQueryWith_ :: (MonadUnliftIO m, MonadLogger m, MonadTools m) => Query -> Decoder r -> Transaction m [r]
|
||||
pgQueryWith_ qry (Decoder fromRow) = do
|
||||
queryWithImpl_ :: (MonadUnliftIO m, MonadLogger m, HasField "pgFormat" tools Tool) => m tools -> Query -> Decoder r -> Transaction m [r]
|
||||
{-# INLINE queryWithImpl_ #-}
|
||||
queryWithImpl_ zoomTools qry (Decoder fromRow) = do
|
||||
tools <- lift @Transaction zoomTools
|
||||
conn <- Transaction ask
|
||||
liftIO (PG.queryWith_ fromRow conn qry)
|
||||
& handlePGException "query" qry (Left ())
|
||||
& handlePGException tools "query" qry (Left ())
|
||||
|
||||
pgQuery :: (ToRow params, FromRow r, MonadUnliftIO m, MonadLogger m, MonadTools m) => Query -> params -> Transaction m [r]
|
||||
pgQuery qry params = do
|
||||
pgQuery :: (ToRow params, FromRow r, MonadUnliftIO m, MonadLogger m, HasField "pgFormat" tools Tool) => tools -> Query -> params -> Transaction m [r]
|
||||
pgQuery tools qry params = do
|
||||
conn <- Transaction ask
|
||||
PG.query conn qry params
|
||||
& handlePGException "query" qry (Left params)
|
||||
& handlePGException tools "query" qry (Left params)
|
||||
|
||||
pgQuery_ :: (FromRow r, MonadUnliftIO m, MonadLogger m, MonadTools m) => Query -> Transaction m [r]
|
||||
pgQuery_ qry = do
|
||||
pgQuery_ :: (FromRow r, MonadUnliftIO m, MonadLogger m, HasField "pgFormat" tools Tool) => tools -> Query -> Transaction m [r]
|
||||
pgQuery_ tools qry = do
|
||||
conn <- Transaction ask
|
||||
PG.query_ conn qry
|
||||
& handlePGException "query_" qry (Left ())
|
||||
& handlePGException tools "query_" qry (Left ())
|
||||
|
||||
data SingleRowError = SingleRowError
|
||||
{ -- | How many columns were actually returned by the query
|
||||
|
@ -313,41 +455,23 @@ data SingleRowError = SingleRowError
|
|||
instance Exception SingleRowError where
|
||||
displayException (SingleRowError {..}) = [fmt|Single row expected from SQL query result, {numberOfRowsReturned} rows were returned instead."|]
|
||||
|
||||
pgFormatQueryNoParams' :: (MonadIO m, MonadLogger m, MonadTools m) => Query -> Transaction m Text
|
||||
pgFormatQueryNoParams' q =
|
||||
lift $ pgFormatQueryByteString q.fromQuery
|
||||
|
||||
pgFormatQuery' :: (MonadIO m, ToRow params, MonadLogger m, MonadTools m) => Query -> params -> Transaction m Text
|
||||
pgFormatQuery' q p =
|
||||
pgFormatQuery' :: (MonadIO m, ToRow params, MonadLogger m, HasField "pgFormat" tools Tool) => tools -> Query -> params -> Transaction m Text
|
||||
pgFormatQuery' tools q p =
|
||||
pgFormatQuery q p
|
||||
>>= lift . pgFormatQueryByteString
|
||||
>>= lift . pgFormatQueryByteString tools
|
||||
|
||||
pgFormatQueryMany' :: (MonadIO m, ToRow params, MonadLogger m, MonadTools m) => Query -> [params] -> Transaction m Text
|
||||
pgFormatQueryMany' q p =
|
||||
pgFormatQueryMany' :: (MonadIO m, ToRow params, MonadLogger m, HasField "pgFormat" tools Tool) => tools -> Query -> [params] -> Transaction m Text
|
||||
pgFormatQueryMany' tools q p =
|
||||
pgFormatQueryMany q p
|
||||
>>= lift . pgFormatQueryByteString
|
||||
>>= lift . pgFormatQueryByteString tools
|
||||
|
||||
-- | Tools required at runtime
|
||||
data Tools = Tools
|
||||
{ pgFormat :: Tool
|
||||
}
|
||||
deriving stock (Show)
|
||||
-- | Read the executable name "pg_format"
|
||||
postgresToolsParser :: ToolParserT IO (Label "pgFormat" Tool)
|
||||
postgresToolsParser = label @"pgFormat" <$> readTool "pg_format"
|
||||
|
||||
class Monad m => MonadTools m where
|
||||
getTools :: m Tools
|
||||
|
||||
initMonadTools :: Label "envvar" Text -> IO Tools
|
||||
initMonadTools var =
|
||||
Tool.readTools (label @"toolsEnvVar" var.envvar) toolParser
|
||||
where
|
||||
toolParser = do
|
||||
pgFormat <- readTool "pg_format"
|
||||
pure $ Tools {..}
|
||||
|
||||
pgFormatQueryByteString :: (MonadIO m, MonadLogger m, MonadTools m) => ByteString -> m Text
|
||||
pgFormatQueryByteString queryBytes = do
|
||||
pgFormatQueryByteString :: (MonadIO m, MonadLogger m, HasField "pgFormat" tools Tool) => tools -> ByteString -> m Text
|
||||
pgFormatQueryByteString tools queryBytes = do
|
||||
do
|
||||
tools <- getTools
|
||||
(exitCode, stdout, stderr) <-
|
||||
Process.readProcessWithExitCode
|
||||
tools.pgFormat.toolPath
|
||||
|
@ -356,8 +480,8 @@ pgFormatQueryByteString queryBytes = do
|
|||
case exitCode of
|
||||
ExitSuccess -> pure (stdout & stringToText)
|
||||
ExitFailure status -> do
|
||||
logWarn [fmt|pg_format failed with status {status} while formatting the query, using original query string. Is there a syntax error?|]
|
||||
logDebug
|
||||
$logWarn [fmt|pg_format failed with status {status} while formatting the query, using original query string. Is there a syntax error?|]
|
||||
$logDebug
|
||||
( prettyErrorTree
|
||||
( nestedMultiError
|
||||
"pg_format output"
|
||||
|
@ -366,9 +490,79 @@ pgFormatQueryByteString queryBytes = do
|
|||
)
|
||||
)
|
||||
)
|
||||
logDebug [fmt|pg_format stdout: stderr|]
|
||||
$logDebug [fmt|pg_format stdout: stderr|]
|
||||
pure (queryBytes & bytesToTextUtf8Lenient)
|
||||
|
||||
data DebugLogDatabaseQueries
|
||||
= -- | Do not log the database queries
|
||||
DontLogDatabaseQueries
|
||||
| -- | Log the database queries as debug output;
|
||||
LogDatabaseQueries
|
||||
| -- | Log the database queries as debug output and additionally the EXPLAIN output (from the query analyzer, not the actual values after execution cause that’s a bit harder to do)
|
||||
LogDatabaseQueriesAndExplain
|
||||
deriving stock (Show, Enum, Bounded)
|
||||
|
||||
data HasQueryParams param
|
||||
= HasNoParams
|
||||
| HasSingleParam param
|
||||
| HasMultiParams [param]
|
||||
|
||||
-- | Log the postgres query depending on the given setting
|
||||
logQueryIfEnabled ::
|
||||
( ToRow params,
|
||||
MonadUnliftIO m,
|
||||
MonadLogger m,
|
||||
HasField "pgFormat" tools Tool
|
||||
) =>
|
||||
tools ->
|
||||
DebugLogDatabaseQueries ->
|
||||
Query ->
|
||||
HasQueryParams params ->
|
||||
Transaction m ()
|
||||
logQueryIfEnabled tools logDatabaseQueries qry params = do
|
||||
-- In case we have query logging enabled, we want to do that
|
||||
let formattedQuery = case params of
|
||||
HasNoParams -> pgFormatQueryNoParams' tools qry
|
||||
HasSingleParam p -> pgFormatQuery' tools qry p
|
||||
HasMultiParams ps -> pgFormatQueryMany' tools qry ps
|
||||
let doLog errs =
|
||||
errs
|
||||
& nestedMultiError "Postgres query"
|
||||
& prettyErrorTree
|
||||
& $logDebug
|
||||
& lift
|
||||
let addQuery = do
|
||||
formattedQuery
|
||||
<&> newError
|
||||
<&> singleError
|
||||
let addExplain = do
|
||||
q <- formattedQuery
|
||||
queryWithImpl_
|
||||
(pure tools)
|
||||
( "EXPLAIN "
|
||||
<> (
|
||||
-- TODO: this is not nice, but the only way to get the `executeMany` form to work with this
|
||||
-- because we need the query with all elements already interpolated.
|
||||
Query (q & textToBytesUtf8)
|
||||
)
|
||||
)
|
||||
(Dec.fromField @Text)
|
||||
<&> Text.intercalate "\n"
|
||||
<&> newError
|
||||
<&> singleError
|
||||
|
||||
case logDatabaseQueries of
|
||||
DontLogDatabaseQueries -> pure ()
|
||||
LogDatabaseQueries -> do
|
||||
aq <- addQuery
|
||||
doLog (aq :| [])
|
||||
LogDatabaseQueriesAndExplain -> do
|
||||
aq <- addQuery
|
||||
-- XXX: stuff like `CREATE SCHEMA` cannot be EXPLAINed, so we should catch exceptions here
|
||||
-- and just ignore anything that errors (if it errors because of a problem with the query, it would have been caught by the query itself.
|
||||
ex <- addExplain
|
||||
doLog (nestedError "Query" aq :| [nestedError "Explain" ex])
|
||||
|
||||
instance (ToField t1) => ToRow (Label l1 t1) where
|
||||
toRow t2 = toRow $ PG.Only $ getField @l1 t2
|
||||
|
||||
|
|
55
users/Profpatsch/my-prelude/src/Seconds.hs
Normal file
55
users/Profpatsch/my-prelude/src/Seconds.hs
Normal file
|
@ -0,0 +1,55 @@
|
|||
module Seconds where
|
||||
|
||||
import Data.Aeson (FromJSON)
|
||||
import Data.Aeson qualified as Json
|
||||
import Data.Aeson.Types (FromJSON (parseJSON))
|
||||
import Data.Scientific
|
||||
import Data.Time (NominalDiffTime)
|
||||
import FieldParser
|
||||
import FieldParser qualified as Field
|
||||
import GHC.Natural (naturalToInteger)
|
||||
import PossehlAnalyticsPrelude
|
||||
|
||||
-- | A natural number of seconds.
|
||||
newtype Seconds = Seconds {unSeconds :: Natural}
|
||||
deriving stock (Eq, Show)
|
||||
|
||||
-- | Parse a decimal number as a number of seconds
|
||||
textToSeconds :: FieldParser Text Seconds
|
||||
textToSeconds = Seconds <$> Field.decimalNatural
|
||||
|
||||
scientificToSeconds :: FieldParser Scientific Seconds
|
||||
scientificToSeconds =
|
||||
( Field.boundedScientificIntegral @Int "Number of seconds"
|
||||
>>> Field.integralToNatural
|
||||
)
|
||||
& rmap Seconds
|
||||
|
||||
-- Microseconds, represented internally with a 64 bit Int
|
||||
newtype MicrosecondsInt = MicrosecondsInt {unMicrosecondsInt :: Int}
|
||||
deriving stock (Eq, Show)
|
||||
|
||||
-- | Try to fit a number of seconds into a MicrosecondsInt
|
||||
secondsToMicrosecondsInt :: FieldParser Seconds MicrosecondsInt
|
||||
secondsToMicrosecondsInt =
|
||||
lmap
|
||||
(\sec -> naturalToInteger sec.unSeconds * 1_000_000)
|
||||
(Field.bounded "Could not fit into an Int after multiplying with 1_000_000 (seconds to microseconds)")
|
||||
& rmap MicrosecondsInt
|
||||
|
||||
secondsToNominalDiffTime :: Seconds -> NominalDiffTime
|
||||
secondsToNominalDiffTime sec =
|
||||
sec.unSeconds
|
||||
& naturalToInteger
|
||||
& fromInteger @NominalDiffTime
|
||||
|
||||
instance FromJSON Seconds where
|
||||
parseJSON = Field.toParseJSON jsonNumberToSeconds
|
||||
|
||||
-- | Parse a json number as a number of seconds.
|
||||
jsonNumberToSeconds :: FieldParser' Error Json.Value Seconds
|
||||
jsonNumberToSeconds = Field.jsonNumber >>> scientificToSeconds
|
||||
|
||||
-- | Return the number of seconds in a week
|
||||
secondsInAWeek :: Seconds
|
||||
secondsInAWeek = Seconds (3600 * 24 * 7)
|
|
@ -23,7 +23,6 @@ import Database.PostgreSQL.Simple (Binary (Binary), Only (..))
|
|||
import Database.PostgreSQL.Simple qualified as Postgres
|
||||
import Database.PostgreSQL.Simple.SqlQQ (sql)
|
||||
import Database.PostgreSQL.Simple.Types (PGArray (PGArray))
|
||||
import Database.PostgreSQL.Simple.Types qualified as Postgres
|
||||
import Database.Postgres.Temp qualified as TmpPg
|
||||
import FieldParser (FieldParser' (..))
|
||||
import FieldParser qualified as Field
|
||||
|
@ -53,6 +52,7 @@ import Text.Blaze.Html (Html)
|
|||
import Text.Blaze.Html.Renderer.Pretty qualified as Html.Pretty
|
||||
import Text.Blaze.Html.Renderer.Utf8 qualified as Html
|
||||
import Text.Blaze.Html5 qualified as Html
|
||||
import Tool (Tool, readTool, readTools)
|
||||
import UnliftIO
|
||||
|
||||
htmlUi :: App ()
|
||||
|
@ -757,7 +757,7 @@ getTorrentFileById dat = do
|
|||
WHERE torrent_id = ?::integer
|
||||
|]
|
||||
(Only $ (dat.torrentId :: Int))
|
||||
(label @"torrentFile" <$> decBytea)
|
||||
(label @"torrentFile" <$> Dec.bytea)
|
||||
>>= ensureSingleRow
|
||||
|
||||
updateTransmissionTorrentHashById ::
|
||||
|
@ -778,9 +778,6 @@ updateTransmissionTorrentHashById dat = do
|
|||
dat.torrentId :: Int
|
||||
)
|
||||
|
||||
decBytea :: Dec.Decoder ByteString
|
||||
decBytea = Dec.fromField @(Binary ByteString) <&> (.fromBinary)
|
||||
|
||||
assertOneUpdated ::
|
||||
(HasField "numberOfRowsAffected" r Natural, MonadThrow m) =>
|
||||
Text ->
|
||||
|
@ -986,7 +983,7 @@ assertM f v = case f v of
|
|||
|
||||
runAppWith :: AppT IO a -> IO (Either TmpPg.StartError a)
|
||||
runAppWith appT = withDb $ \db -> do
|
||||
tools <- initMonadTools (label @"envvar" "WHATCD_RESOLVER_TOOLS")
|
||||
pgFormat <- readTools (label @"toolsEnvVar" "WHATCD_RESOLVER_TOOLS") (readTool "pg_format")
|
||||
let config = label @"logDatabaseQueries" LogDatabaseQueries
|
||||
pgConnPool <-
|
||||
Pool.createPool
|
||||
|
@ -1028,8 +1025,8 @@ withDb act = do
|
|||
act db
|
||||
|
||||
data Context = Context
|
||||
{ config :: Label "logDatabaseQueries" DatabaseLogging,
|
||||
tools :: Tools,
|
||||
{ config :: Label "logDatabaseQueries" DebugLogDatabaseQueries,
|
||||
pgFormat :: Tool,
|
||||
pgConnPool :: Pool Postgres.Connection,
|
||||
transmissionSessionId :: MVar ByteString
|
||||
}
|
||||
|
@ -1054,9 +1051,6 @@ orAppThrowTree = \case
|
|||
instance MonadIO m => MonadLogger (AppT m) where
|
||||
monadLoggerLog loc src lvl msg = liftIO $ Logger.defaultOutput IO.stderr loc src lvl (Logger.toLogStr msg)
|
||||
|
||||
instance Monad m => MonadTools (AppT m) where
|
||||
getTools = AppT $ asks (.tools)
|
||||
|
||||
class MonadTransmission m where
|
||||
getTransmissionId :: m (Maybe ByteString)
|
||||
setTransmissionId :: ByteString -> m ()
|
||||
|
@ -1068,32 +1062,13 @@ instance (MonadIO m) => MonadTransmission (AppT m) where
|
|||
putMVar var t
|
||||
|
||||
instance (MonadThrow m, MonadUnliftIO m) => MonadPostgres (AppT m) where
|
||||
execute qry params = do
|
||||
conf <- lift $ AppT (asks (.config))
|
||||
logQueryIfEnabled conf qry (HasSingleParam params)
|
||||
pgExecute qry params
|
||||
execute_ qry = do
|
||||
conf <- lift $ AppT (asks (.config))
|
||||
logQueryIfEnabled @(Only Text) conf qry HasNoParams
|
||||
pgExecute_ qry
|
||||
executeMany qry params = do
|
||||
conf <- lift $ AppT (asks (.config))
|
||||
logQueryIfEnabled conf qry (HasMultiParams params)
|
||||
pgExecuteMany qry params
|
||||
executeManyReturningWith qry params dec = do
|
||||
conf <- lift $ AppT (asks (.config))
|
||||
logQueryIfEnabled conf qry (HasMultiParams params)
|
||||
pgExecuteManyReturningWith qry params dec
|
||||
|
||||
queryWith qry params decoder = do
|
||||
conf <- lift $ AppT (asks (.config))
|
||||
logQueryIfEnabled conf qry (HasSingleParam params)
|
||||
pgQueryWith qry params decoder
|
||||
|
||||
-- TODO: log these queries as well with `logQueryIfEnabled`, but test out whether it works with query_ and foldRows first.
|
||||
queryWith_ = pgQueryWith_
|
||||
foldRows = pgFold
|
||||
|
||||
execute = executeImpl (AppT ask) (AppT $ asks (.config.logDatabaseQueries))
|
||||
execute_ = executeImpl_ (AppT ask) (AppT $ asks (.config.logDatabaseQueries))
|
||||
executeMany = executeManyImpl (AppT ask) (AppT $ asks (.config.logDatabaseQueries))
|
||||
executeManyReturningWith = executeManyReturningWithImpl (AppT ask) (AppT $ asks (.config.logDatabaseQueries))
|
||||
queryWith = queryWithImpl (AppT ask) (AppT $ asks (.config.logDatabaseQueries))
|
||||
queryWith_ = queryWithImpl_ (AppT ask)
|
||||
foldRows = foldRowsImpl (AppT ask)
|
||||
runTransaction = runPGTransaction
|
||||
|
||||
runPGTransaction :: MonadUnliftIO m => Transaction (AppT m) a -> AppT m a
|
||||
|
@ -1103,83 +1078,7 @@ runPGTransaction (Transaction transaction) = do
|
|||
withPGTransaction pool $ \conn -> do
|
||||
unliftIO $ runReaderT transaction conn
|
||||
|
||||
-- | Perform a Postgres action within a transaction
|
||||
withPGTransaction ::
|
||||
-- | Postgres connection pool to be used for the action
|
||||
Pool Postgres.Connection ->
|
||||
-- | DB-action to be performed
|
||||
(Postgres.Connection -> IO a) ->
|
||||
-- | Result of the DB-action
|
||||
IO a
|
||||
withPGTransaction connPool f =
|
||||
Pool.withResource
|
||||
connPool
|
||||
(\conn -> Postgres.withTransaction conn (f conn))
|
||||
|
||||
data HasQueryParams param
|
||||
= HasNoParams
|
||||
| HasSingleParam param
|
||||
| HasMultiParams [param]
|
||||
|
||||
-- | Log the postgres query depending on the setting of @config.debugInfo.logDatabaseQueries@.
|
||||
logQueryIfEnabled ::
|
||||
forall params config m.
|
||||
( Postgres.ToRow params,
|
||||
MonadUnliftIO m,
|
||||
MonadLogger m,
|
||||
MonadTools m,
|
||||
HasField "logDatabaseQueries" config DatabaseLogging
|
||||
) =>
|
||||
config ->
|
||||
Postgres.Query ->
|
||||
HasQueryParams params ->
|
||||
Transaction m ()
|
||||
logQueryIfEnabled config qry params = do
|
||||
-- In case we have query logging enabled, we want to do that
|
||||
let formattedQuery = case params of
|
||||
HasNoParams -> pgFormatQueryNoParams' qry
|
||||
HasSingleParam p -> pgFormatQuery' qry p
|
||||
HasMultiParams ps -> pgFormatQueryMany' qry ps
|
||||
|
||||
let doLog errs =
|
||||
errs
|
||||
& nestedMultiError "Postgres query"
|
||||
& prettyErrorTree
|
||||
& logDebug
|
||||
& lift
|
||||
let addQuery = do
|
||||
formattedQuery
|
||||
<&> newError
|
||||
<&> singleError
|
||||
let addExplain = do
|
||||
q <- formattedQuery
|
||||
pgQueryWith_
|
||||
( "EXPLAIN "
|
||||
<> (
|
||||
-- TODO: this is not nice, but the only way to get the `executeMany` form to work with this
|
||||
-- because we need the query with all elements already interpolated.
|
||||
Postgres.Query (q & textToBytesUtf8)
|
||||
)
|
||||
)
|
||||
(Dec.fromField @Text)
|
||||
<&> Text.intercalate "\n"
|
||||
<&> newError
|
||||
<&> singleError
|
||||
|
||||
case config.logDatabaseQueries of
|
||||
DontLogDatabaseQueries -> pure ()
|
||||
LogDatabaseQueries -> do
|
||||
aq <- addQuery
|
||||
doLog (aq :| [])
|
||||
LogDatabaseQueriesAndExplain -> do
|
||||
aq <- addQuery
|
||||
-- XXX: stuff like `CREATE SCHEMA` cannot be EXPLAINed, so we should catch exceptions here
|
||||
-- and just ignore anything that errors (if it errors because of a problem with the query, it would have been caught by the query itself.
|
||||
ex <- addExplain
|
||||
doLog (nestedError "Query" aq :| [nestedError "Explain" ex])
|
||||
|
||||
data DatabaseLogging
|
||||
= DontLogDatabaseQueries
|
||||
| LogDatabaseQueries
|
||||
| LogDatabaseQueriesAndExplain
|
||||
deriving stock (Show)
|
||||
|
|
Loading…
Reference in a new issue