async_postgres/pg_client/core

Search:
Group by:

Shared building blocks for pg_client submodules (transaction opts, inline params, recv loops).

Internal module: not part of the public API. Import the pg_client hub instead; what it re-exports is the supported surface (see tests/api_surface.golden).

Types

AccessMode = enum
  amDefault, amReadWrite, amReadOnly
PostgreSQL transaction access mode (read-write or read-only). 
DeferrableMode = enum
  dmDefault, dmDeferrable, dmNotDeferrable
PostgreSQL transaction deferrable mode (for serializable read-only transactions). 
IsolationLevel = enum
  ilDefault, ilReadCommitted, ilRepeatableRead, ilSerializable,
  ilReadUncommitted
PostgreSQL transaction isolation level. 
RetryOptions = object
  maxAttempts*: int = 3      ## Total attempts (``<=1`` = no retry).
  baseDelayMs*: int = 20     ## Initial backoff ms.
  maxDelayMs*: int = 1000    ## Max backoff ms.
  multiplier*: float = 2.0   ## Backoff multiplier.
  jitter*: bool = true       ## Full jitter via ``std/random``; call ``randomize()`` for cross-process de-correlation.
  retryableStates*: seq[string] = ["40001", "40P01"] ## SQLSTATEs that trigger retry.
Retry config for withTransactionRetry; unset fields keep the defaults below. 
TransactionOptions = object
  isolation*: IsolationLevel
  access*: AccessMode
  deferrable*: DeferrableMode
Options for BEGIN: isolation level, access mode, and deferrable mode.

Consts

copyBatchSize = 262144
256KB batch threshold for COPY IN buffering 
generatedPortalNameLen = 27
Same, for the portal name openCursorImpl generates. 
generatedStmtNameLen = 23
Longest name nextStmtName can produce. Size pre-flights charge it because the real name is only picked at send time. 
StmtCacheInvalidatingStates = ["26000", "0A000"]
SQLSTATEs that invalidate cached prepared statements (requires re-parse). 42P18 is absent on purpose: it is Parse-phase, so no cached statement can hit it.

Procs

proc backoffDelayMs(opts: RetryOptions; attempt: int): int {....raises: [],
    tags: [], forbids: [].}
Backoff ms for attempt (1-based). Exponential with jitter. 
proc buildBeginSql(opts: TransactionOptions): string {....raises: [], tags: [],
    forbids: [].}
Build a BEGIN SQL statement with the specified transaction options (isolation level, access mode, deferrable mode). 
func cacheHitColFmts(resultFormats: openArray[int16]; cachedColFmts: seq[int16];
                     numCols: int): seq[int16] {....raises: [], tags: [],
    forbids: [].}
Per-column formats for a cache hit. Prefers the formats this Bind requested: the same SQL may be re-issued with a different resultFormat, and the stale cached one would reinterpret the bytes. 
func deriveColFmts(resultFormats: openArray[int16]; numCols: int): seq[int16] {.
    ...raises: [], tags: [], forbids: [].}
Expand Bind result-format codes to per-column: one code broadcasts, an array applies positionally, columns past its end default to text (0). 
proc extractParams(params: openArray[PgParam]): tuple[oids: seq[int32],
    formats: seq[int16], values: seq[Option[seq[byte]]]] {....raises: [], tags: [],
    forbids: [].}
proc flattenInline(params: openArray[PgParamInline]; resultFormatsLen: int = 0): tuple[
    data: seq[byte], ranges: seq[tuple[off: int32, len: int32]],
    oids: seq[int32], formats: seq[int16]] {.
    ...raises: [PgTypeError, PgMessageTooLargeError, PgProtocolError], tags: [],
    forbids: [].}
proc invalidateIfOidMismatch(conn: PgConnection; sql: string;
                             cached: CachedStmt; currentOids: openArray[int32];
                             cacheHit: var bool) {....raises: [], tags: [],
    forbids: [].}
Evict cached statement if OIDs mismatch; sets cacheHit=false. cached may be nil iff cacheHit == false (only deref'd under it). 
proc invalidateIfOidMismatch(conn: PgConnection; sql: string;
                             cached: CachedStmt; params: openArray[PgParam];
                             cacheHit: var bool) {....raises: [], tags: [],
    forbids: [].}
PgParam overload (no seq[int32] alloc). Same nil precondition on cached. 
proc isRetryableTxError(e: ref CatchableError; states: openArray[string]): bool {.
    ...raises: [], tags: [], forbids: [].}
Whether e is a PgQueryError whose SQLSTATE is in states. Non-PgQueryError failures (connection drops, timeouts) are never retryable here: they leave the connection unusable for a fresh attempt. 
proc paramOidsMatch(cachedOids, currentOids: openArray[int32]): bool {.
    ...raises: [], tags: [], forbids: [].}
Whether cached param OIDs match current (0 = wildcard). 
proc paramOidsMatch(cachedOids: openArray[int32]; params: openArray[PgParam]): bool {.
    ...raises: [], tags: [], forbids: [].}
PgParam overload (avoids seq[int32] alloc). 
proc preflightResultFormatsLen(cached: CachedStmt; cacheHit: bool;
                               resultFormatsLen: int = 0): int {....raises: [],
    tags: [], forbids: [].}
Result-format count the send path will really emit: a cache hit replays cached.resultFormats when the caller passed none. Call after invalidateIfOidMismatch; cached may be nil iff not cacheHit. 
func toFormatCodes(rf: ResultFormat): seq[int16] {....raises: [], tags: [],
    forbids: [].}
Convert a high-level ResultFormat to wire-protocol format codes. 
proc validateEncodedParams(params: openArray[Option[seq[byte]]];
                           paramFormatsLen: int; resultFormatsLen: int = 0;
                           stmtNameLen: int = generatedStmtNameLen;
                           portalLen: int = 0) {.
    ...raises: [PgTypeError, PgMessageTooLargeError], tags: [], forbids: [].}
validateTypedParams for the already-encoded seq[Option[seq[byte]]] path. Runs before the send templates start filling sendBuf, so a call this rejects fails without having touched the connection. 
proc validateExtendedQuery(sql: string; nParams: int; nParamOids: int = nParams;
                           stmtNameLen: int = generatedStmtNameLen) {.
    ...raises: [PgTypeError, PgMessageTooLargeError], tags: [], forbids: [].}
Add-time validation shared by every extended-query entry point. nParamOids sizes the Parse, which the *Impl procs take separately from the values. Pass stmtNameLen 0 for an unnamed statement. 
proc validateParamCount(n: int; what: string) {....raises: [PgTypeError], tags: [],
    forbids: [].}
Reject a count the encoder would reject later: a send-phase failure would take the whole pipelined batch down. 
proc validateParseMsg(sql: string; nParams: int;
                      stmtNameLen = generatedStmtNameLen) {.
    ...raises: [PgTypeError, PgMessageTooLargeError], tags: [], forbids: [].}
Reject a Parse the encoder would reject later, so a pipelined op fails on its own instead of in buildSendPhase. Best-effort; patchMsgLen stays the authority on message size. 
proc validateRawBind(data: openArray[byte];
                     ranges: openArray[tuple[off: int32, len: int32]];
                     paramFormats: openArray[int16]; resultFormatsLen: int = 0;
                     stmtNameLen: int = generatedStmtNameLen) {.
    ...raises: [PgTypeError, PgMessageTooLargeError], tags: [], forbids: [].}
validateEncodedParams for the raw buffer/ranges path (addBindRaw). The *Impl procs are public, so data/ranges may never have gone through flattenInline. 
proc validateTypedParams(params: openArray[PgParam]; resultFormatsLen: int = 0;
                         stmtNameLen: int = generatedStmtNameLen) {.
    ...raises: [PgTypeError, PgMessageTooLargeError], tags: [], forbids: [].}
validateInlineParam's counterpart for the seq[PgParam] path: checks the Int16 count, the payload total and the Bind envelope at add time. resultFormatsLen 0 means unknown, making the check a lower bound.

Templates

template appendInlineParamUnchecked(data: var seq[byte]; ranges: var seq[
    tuple[off: int32, len: int32]]; oids: var seq[int32];
                                    formats: var seq[int16]; p: PgParamInline)
Encode one PgParamInline into SoA buffers. Raises PgTypeError. The caller must have run validateInlineParam on p: its bounds keep the overflow read in range. 
template execRecvLoop(conn: PgConnection; sql: string;
                      cacheHit, cacheMiss: bool; stmtName: string;
                      commandTag: var string)
Receive-loop counterpart of queryRecvLoop for the extended-query exec path: DataRows are dropped by the parser (bare pumpUntilReady uses skipDataRow = true); this loop only exposes the CommandComplete tag via the commandTag out-parameter. Shared by execImpl (both overloads), execInlineImpl, and execDirectRunImpl. 
template queryEachRecvLoop(conn: PgConnection; sql: string;
                           resultFormats: openArray[int16];
                           cacheHit, cacheMiss: bool; stmtName: string;
                           cachedFields: var seq[FieldDescription];
                           cachedColFmts: seq[int16]; cachedColOids: seq[int32];
                           callback: RowCallback; rowCount: var int64)
template queryRecvLoop(conn: PgConnection; sql: string;
                       resultFormats: openArray[int16];
                       cacheHit, cacheMiss: bool; stmtName: string;
                       cachedFields: var seq[FieldDescription];
                       cachedColFmts: seq[int16]; cachedColOids: seq[int32];
                       qr: var QueryResult)
template sendExtendedExec(conn: PgConnection; cached: CachedStmt;
                          cacheHit, cacheMiss: var bool; stmtName: var string;
                          parseStep, bindStep: untyped)
exec variant of sendExtendedQuery (no per-column format tracking). Precondition: cached may be nil iff cacheHit == false. 
template sendExtendedQuery(conn: PgConnection; resultFormats: seq[int16];
                           cached: CachedStmt; cacheHit, cacheMiss: var bool;
                           stmtName: var string;
                           cachedFields: var seq[FieldDescription];
                           cachedColFmts: var seq[int16];
                           cachedColOids: var seq[int32];
                           effectiveResultFormats: var seq[int16];
                           parseStep, bindStep: untyped)
Emit Parse/Bind/Describe/Execute/Sync sequence (cache hit/miss/disabled). Precondition: cached may be nil iff cacheHit == false; the cache-miss and cache-disabled branches never read it. 
template validateInlineParam(p: PgParamInline)
Reject a bad PgParamInline with PgTypeError, so a hand-built one stays catchable under PgError instead of a fatal RangeDefect.

Made with Nim. Generated: 2026-09-11 08:44:51 UTC