async_postgres/pg_pool_cluster

Search:
Group by:

Types

PgPoolCluster = ref object
Connection pool cluster with explicit read/write routing.
  • read* methods route to the replica pool (read-only queries).
  • write* methods route to the primary pool (writes, SELECT FOR UPDATE, etc.).

For transactions, use cluster.primaryPool.withTransaction directly.

ReplicaFallback = enum
  fallbackNone,             ## Error when replica is unavailable
  fallbackPrimary            ## Fall back to primary when replica is unavailable

Procs

proc close(cluster: PgPoolCluster; timeout = ZeroDuration): Future[void] {.
    ...stackTrace: false, raises: [Exception, CatchableError],
    tags: [RootEffect, TimeEffect], forbids: [].}
Close both primary and replica pools. 
proc fallbackTimeout(cluster: PgPoolCluster): Duration {....raises: [], tags: [],
    forbids: [].}
The configured fallback timeout for read operations. When fallbackPrimary is set and the replica acquire fails, this limits how long the fallback primary acquire may wait. ZeroDuration means the primary pool's own acquireTimeout is used as-is. 
proc isClosed(cluster: PgPoolCluster): bool {....raises: [], tags: [], forbids: [].}
Whether the cluster has been closed. 
proc newPoolCluster(primaryConfig: PoolConfig; replicaConfig: PoolConfig;
                    fallback = fallbackNone; fallbackTimeout = ZeroDuration): Future[
    PgPoolCluster] {....stackTrace: false,
                     raises: [Exception, ValueError, CatchableError],
                     tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
Create a new pool cluster with separate primary and replica pools. If connConfig.targetSessionAttrs is tsaAny (the default), it is automatically set to tsaReadWrite for primary and tsaPreferStandby for replica. 
proc primaryPool(cluster: PgPoolCluster): PgPool {....raises: [], tags: [],
    forbids: [].}
The primary (read-write) pool. 
proc readQuery(cluster: PgPoolCluster; sql: string; params: seq[PgParam] = @[];
               resultFormat: ResultFormat = rfAuto;
               timeout: Duration = ZeroDuration): Future[QueryResult] {.
    ...stackTrace: false, raises: [Exception, ValueError, CatchableError,
                                PgPoolError, PgQueryError, PgConnectionError,
                                SslError, KeyError, ProtocolError,
                                PgTimeoutError, AsyncTimeoutError],
    tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
proc readQueryColumn(cluster: PgPoolCluster; sql: string;
                     params: seq[PgParam] = @[];
                     timeout: Duration = ZeroDuration): Future[seq[string]] {.
    ...stackTrace: false, raises: [Exception, ValueError, CatchableError,
                                PgPoolError, PgQueryError, PgConnectionError,
                                SslError, KeyError, ProtocolError,
                                PgTimeoutError, AsyncTimeoutError, PgTypeError],
    tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
proc readQueryEach(cluster: PgPoolCluster; sql: string;
                   params: seq[PgParam] = @[]; callback: RowCallback;
                   resultFormat: ResultFormat = rfAuto;
                   timeout: Duration = ZeroDuration): Future[int64] {.
    ...stackTrace: false, raises: [Exception, ValueError, CatchableError,
                                PgPoolError, PgQueryError, PgConnectionError,
                                SslError, KeyError, ProtocolError,
                                PgTimeoutError, AsyncTimeoutError],
    tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
proc readQueryExists(cluster: PgPoolCluster; sql: string;
                     params: seq[PgParam] = @[];
                     timeout: Duration = ZeroDuration): Future[bool] {.
    ...stackTrace: false, raises: [Exception, ValueError, CatchableError,
                                PgPoolError, PgQueryError, PgConnectionError,
                                SslError, KeyError, ProtocolError,
                                PgTimeoutError, AsyncTimeoutError],
    tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
proc readQueryOne(cluster: PgPoolCluster; sql: string;
                  params: seq[PgParam] = @[];
                  resultFormat: ResultFormat = rfAuto;
                  timeout: Duration = ZeroDuration): Future[Option[Row]] {.
    ...stackTrace: false, raises: [Exception, ValueError, CatchableError,
                                PgPoolError, PgQueryError, PgConnectionError,
                                SslError, KeyError, ProtocolError,
                                PgTimeoutError, AsyncTimeoutError],
    tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
proc readQueryValue(cluster: PgPoolCluster; sql: string;
                    params: seq[PgParam] = @[]; timeout: Duration = ZeroDuration): Future[
    string] {....stackTrace: false, raises: [Exception, ValueError, CatchableError,
    PgPoolError, PgQueryError, PgConnectionError, SslError, KeyError,
    ProtocolError, PgTimeoutError, AsyncTimeoutError, PgError, PgTypeError],
              tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
proc readQueryValue[T](cluster: PgPoolCluster; _: typedesc[T]; sql: string;
                       params: seq[PgParam] = @[];
                       timeout: Duration = ZeroDuration): Future[T] {.
    ...stackTrace: false.}
proc readQueryValueOpt(cluster: PgPoolCluster; sql: string;
                       params: seq[PgParam] = @[];
                       timeout: Duration = ZeroDuration): Future[Option[string]] {.
    ...stackTrace: false, raises: [Exception, ValueError, CatchableError,
                                PgPoolError, PgQueryError, PgConnectionError,
                                SslError, KeyError, ProtocolError,
                                PgTimeoutError, AsyncTimeoutError, PgTypeError],
    tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
proc readQueryValueOpt[T](cluster: PgPoolCluster; _: typedesc[T]; sql: string;
                          params: seq[PgParam] = @[];
                          timeout: Duration = ZeroDuration): Future[Option[T]] {.
    ...stackTrace: false.}
proc readQueryValueOrDefault(cluster: PgPoolCluster; sql: string;
                             params: seq[PgParam] = @[]; default: string = "";
                             timeout: Duration = ZeroDuration): Future[string] {.
    ...stackTrace: false, raises: [Exception, ValueError, CatchableError,
                                PgPoolError, PgQueryError, PgConnectionError,
                                SslError, KeyError, ProtocolError,
                                PgTimeoutError, AsyncTimeoutError, PgTypeError],
    tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
proc readQueryValueOrDefault[T](cluster: PgPoolCluster; _: typedesc[T];
                                sql: string; params: seq[PgParam] = @[];
                                default: T; timeout: Duration = ZeroDuration): Future[
    T] {....stackTrace: false.}
proc readSimpleQuery(cluster: PgPoolCluster; sql: string): Future[
    seq[QueryResult]] {....stackTrace: false, raises: [Exception, ValueError,
    CatchableError, PgPoolError, PgConnectionError, SslError, ProtocolError,
    PgQueryError, AsyncTimeoutError],
                        tags: [RootEffect, TimeEffect, WriteIOEffect],
                        forbids: [].}
proc replicaFallback(cluster: PgPoolCluster): ReplicaFallback {....raises: [],
    tags: [], forbids: [].}
The configured replica fallback behavior. 
proc replicaPool(cluster: PgPoolCluster): PgPool {....raises: [], tags: [],
    forbids: [].}
The replica (read-only) pool. 
proc writeExec(cluster: PgPoolCluster; sql: string; params: seq[PgParam] = @[];
               timeout: Duration = ZeroDuration): Future[CommandResult] {.
    ...stackTrace: false, raises: [Exception, ValueError, PgPoolError,
                                CatchableError, PgQueryError, PgConnectionError,
                                SslError, KeyError, ProtocolError,
                                PgTimeoutError, AsyncTimeoutError],
    tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
proc writeExecInTransaction(cluster: PgPoolCluster; sql: string;
                            params: seq[PgParam] = @[];
                            timeout: Duration = ZeroDuration): Future[
    CommandResult] {....stackTrace: false, raises: [Exception, ValueError,
    PgPoolError, CatchableError, PgTimeoutError, PgConnectionError, SslError,
    ProtocolError, PgQueryError, AsyncTimeoutError],
                     tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
proc writeNotify(cluster: PgPoolCluster; channel: string; payload: string = "";
                 timeout: Duration = ZeroDuration): Future[void] {.
    ...stackTrace: false, raises: [Exception, ValueError, PgPoolError,
                                CatchableError, PgQueryError, PgConnectionError,
                                SslError, KeyError, ProtocolError,
                                PgTimeoutError, AsyncTimeoutError],
    tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
proc writeQuery(cluster: PgPoolCluster; sql: string;
                params: seq[PgParam] = @[]; resultFormat: ResultFormat = rfAuto;
                timeout: Duration = ZeroDuration): Future[QueryResult] {.
    ...stackTrace: false, raises: [Exception, ValueError, PgPoolError,
                                CatchableError, PgQueryError, PgConnectionError,
                                SslError, KeyError, ProtocolError,
                                PgTimeoutError, AsyncTimeoutError],
    tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
proc writeQueryColumn(cluster: PgPoolCluster; sql: string;
                      params: seq[PgParam] = @[];
                      timeout: Duration = ZeroDuration): Future[seq[string]] {.
    ...stackTrace: false, raises: [Exception, ValueError, PgPoolError,
                                CatchableError, PgQueryError, PgConnectionError,
                                SslError, KeyError, ProtocolError,
                                PgTimeoutError, AsyncTimeoutError, PgTypeError],
    tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
proc writeQueryEach(cluster: PgPoolCluster; sql: string;
                    params: seq[PgParam] = @[]; callback: RowCallback;
                    resultFormat: ResultFormat = rfAuto;
                    timeout: Duration = ZeroDuration): Future[int64] {.
    ...stackTrace: false, raises: [Exception, ValueError, PgPoolError,
                                CatchableError, PgQueryError, PgConnectionError,
                                SslError, KeyError, ProtocolError,
                                PgTimeoutError, AsyncTimeoutError],
    tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
proc writeQueryExists(cluster: PgPoolCluster; sql: string;
                      params: seq[PgParam] = @[];
                      timeout: Duration = ZeroDuration): Future[bool] {.
    ...stackTrace: false, raises: [Exception, ValueError, PgPoolError,
                                CatchableError, PgQueryError, PgConnectionError,
                                SslError, KeyError, ProtocolError,
                                PgTimeoutError, AsyncTimeoutError],
    tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
proc writeQueryInTransaction(cluster: PgPoolCluster; sql: string;
                             params: seq[PgParam] = @[];
                             resultFormat: ResultFormat = rfAuto;
                             timeout: Duration = ZeroDuration): Future[
    QueryResult] {....stackTrace: false, raises: [Exception, ValueError,
    PgPoolError, CatchableError, PgTimeoutError, PgConnectionError, SslError,
    ProtocolError, PgQueryError, AsyncTimeoutError],
                   tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
proc writeQueryOne(cluster: PgPoolCluster; sql: string;
                   params: seq[PgParam] = @[];
                   resultFormat: ResultFormat = rfAuto;
                   timeout: Duration = ZeroDuration): Future[Option[Row]] {.
    ...stackTrace: false, raises: [Exception, ValueError, PgPoolError,
                                CatchableError, PgQueryError, PgConnectionError,
                                SslError, KeyError, ProtocolError,
                                PgTimeoutError, AsyncTimeoutError],
    tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
proc writeQueryValue(cluster: PgPoolCluster; sql: string;
                     params: seq[PgParam] = @[];
                     timeout: Duration = ZeroDuration): Future[string] {.
    ...stackTrace: false, raises: [Exception, ValueError, PgPoolError,
                                CatchableError, PgQueryError, PgConnectionError,
                                SslError, KeyError, ProtocolError,
                                PgTimeoutError, AsyncTimeoutError, PgError,
                                PgTypeError],
    tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
proc writeQueryValue[T](cluster: PgPoolCluster; _: typedesc[T]; sql: string;
                        params: seq[PgParam] = @[];
                        timeout: Duration = ZeroDuration): Future[T] {.
    ...stackTrace: false.}
proc writeQueryValueOpt(cluster: PgPoolCluster; sql: string;
                        params: seq[PgParam] = @[];
                        timeout: Duration = ZeroDuration): Future[Option[string]] {.
    ...stackTrace: false, raises: [Exception, ValueError, PgPoolError,
                                CatchableError, PgQueryError, PgConnectionError,
                                SslError, KeyError, ProtocolError,
                                PgTimeoutError, AsyncTimeoutError, PgTypeError],
    tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
proc writeQueryValueOpt[T](cluster: PgPoolCluster; _: typedesc[T]; sql: string;
                           params: seq[PgParam] = @[];
                           timeout: Duration = ZeroDuration): Future[Option[T]] {.
    ...stackTrace: false.}
proc writeQueryValueOrDefault(cluster: PgPoolCluster; sql: string;
                              params: seq[PgParam] = @[]; default: string = "";
                              timeout: Duration = ZeroDuration): Future[string] {.
    ...stackTrace: false, raises: [Exception, ValueError, PgPoolError,
                                CatchableError, PgQueryError, PgConnectionError,
                                SslError, KeyError, ProtocolError,
                                PgTimeoutError, AsyncTimeoutError, PgTypeError],
    tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
proc writeQueryValueOrDefault[T](cluster: PgPoolCluster; _: typedesc[T];
                                 sql: string; params: seq[PgParam] = @[];
                                 default: T; timeout: Duration = ZeroDuration): Future[
    T] {....stackTrace: false.}
proc writeSimpleExec(cluster: PgPoolCluster; sql: string;
                     timeout: Duration = ZeroDuration): Future[CommandResult] {.
    ...stackTrace: false, raises: [Exception, ValueError, PgPoolError,
                                CatchableError, PgQueryError, PgConnectionError,
                                SslError, ProtocolError, PgTimeoutError,
                                AsyncTimeoutError],
    tags: [RootEffect, TimeEffect, WriteIOEffect], forbids: [].}
proc writeSimpleQuery(cluster: PgPoolCluster; sql: string): Future[
    seq[QueryResult]] {....stackTrace: false, raises: [Exception, ValueError,
    PgPoolError, CatchableError, PgConnectionError, SslError, ProtocolError,
    PgQueryError, AsyncTimeoutError],
                        tags: [RootEffect, TimeEffect, WriteIOEffect],
                        forbids: [].}

Templates

template withPipeline(cluster: PgPoolCluster; pipeline, body: untyped)
Create a pipeline on the primary pool. 
template withReadConnection(cluster: PgPoolCluster; conn, body: untyped)
Acquire a read connection (from replica, with optional primary fallback), execute body, then release. 
template withWriteConnection(cluster: PgPoolCluster; conn, body: untyped)
Acquire a write connection from the primary pool, execute body, then release.

Made with Nim. Generated: 2026-04-03 06:45:30 UTC