Use a connection provider rather than long-lived connections.

The previous implementation expected to work with a context object that
contained a field `conn` that represented a valid database connection.
However, this required the caller to manage the connection's lifecycle
(close, renew, etc.). Now we expect to receive a context object that
provides a `withConn` procedure or template that accepts a statement
block and provides a `conn` variable to that code block. For example:

    createRecord(db: DbContext): Record =
        # withConn must be defined for DbContext
        db.withConn:
          # conn must be injected into the statement block context
          conn.exec(sql("INSERT INTO..."))

In addition, this change provides a connection pooling mechanism
(`DbConnPool`) as a default implementation for this hypothetical
DbContext. There is also a new function `initPool` that will create an
DbConnPool instance.

Callers of this library can modify their DbContext objects to extend
DbConnPool or simply be a type alias of DbConnPool.

fiber_orm.nimble

@@ -1,6 +1,6 @@
 # Package
 
-version       = "0.1.0"
+version       = "1.0.0"
 author        = "Jonathan Bernard"
 description   = "Lightweight Postgres ORM for Nim."
 license       = "GPL-3.0"
@@ -10,4 +10,5 @@ srcDir        = "src"
 
 # Dependencies
 
-requires "nim >= 1.0.4"
+requires "nim >= 1.4.0"
+requires "https://git.jdb-software.com/jdb/nim-namespaced-logging.git"

src/fiber_orm.nim

@@ -1,11 +1,29 @@
-import db_postgres, macros, options, sequtils, strutils, uuids
+import std/db_postgres, std/macros, std/options, std/sequtils, std/strutils
+import namespaced_logging, uuids
 
-from unicode import capitalize
+from std/unicode import capitalize
 
+import ./fiber_orm/pool
 import ./fiber_orm/util
 
+export
+  pool,
+  util.columnNamesForModel,
+  util.dbFormat,
+  util.dbNameToIdent,
+  util.identNameToDb,
+  util.modelName,
+  util.rowToModel,
+  util.tableName
+
 type NotFoundError* = object of CatchableError
 
+var logNs {.threadvar.}: LoggingNamespace
+
+template log(): untyped =
+  if logNs.isNil: logNs = initLoggingNamespace(name = "fiber_orm", level = lvlNotice)
+  logNs
+
 proc newMutateClauses(): MutateClauses =
   return MutateClauses(
     columns: @[],
@@ -18,11 +36,14 @@ proc createRecord*[T](db: DbConn, rec: T): T =
 
   # Confusingly, getRow allows inserts and updates. We use it to get back the ID
   # we want from the row.
-  let newRow = db.getRow(sql(
+  let sqlStmt =
     "INSERT INTO " & tableName(rec) &
     " (" & mc.columns.join(",") & ") " &
     " VALUES (" & mc.placeholders.join(",") & ") " &
-    " RETURNING *"), mc.values)
+    " RETURNING *"
+
+  log().debug "createRecord: [" & sqlStmt & "]"
+  let newRow = db.getRow(sql(sqlStmt), mc.values)
 
   result = rowToModel(T, newRow)
 
@@ -30,53 +51,69 @@ proc updateRecord*[T](db: DbConn, rec: T): bool =
   var mc = newMutateClauses()
   populateMutateClauses(rec, false, mc)
 
-  let setClause = zip(mc.columns, mc.placeholders).mapIt(it.a & " = " & it.b).join(",")
-  let numRowsUpdated = db.execAffectedRows(sql(
+  let setClause = zip(mc.columns, mc.placeholders).mapIt(it[0] & " = " & it[1]).join(",")
+  let sqlStmt =
     "UPDATE " & tableName(rec) &
     " SET " & setClause &
-    " WHERE id = ? "), mc.values.concat(@[$rec.id]))
+    " WHERE id = ? "
+
+  log().debug "updateRecord: [" & sqlStmt & "] id: " & $rec.id
+  let numRowsUpdated = db.execAffectedRows(sql(sqlStmt), mc.values.concat(@[$rec.id]))
 
   return numRowsUpdated > 0;
 
 template deleteRecord*(db: DbConn, modelType: type, id: typed): untyped =
-  db.tryExec(sql("DELETE FROM " & tableName(modelType) & " WHERE id = ?"), $id)
+  let sqlStmt = "DELETE FROM " & tableName(modelType) & " WHERE id = ?"
+  log().debug "deleteRecord: [" & sqlStmt & "] id: " & $id
+  db.tryExec(sql(sqlStmt), $id)
 
 proc deleteRecord*[T](db: DbConn, rec: T): bool =
-  return db.tryExec(sql("DELETE FROM " & tableName(rec) & " WHERE id = ?"), $rec.id)
+  let sqlStmt = "DELETE FROM " & tableName(rec) & " WHERE id = ?"
+  log().debug "deleteRecord: [" & sqlStmt & "] id: " & $rec.id
+  return db.tryExec(sql(sqlStmt), $rec.id)
 
 template getRecord*(db: DbConn, modelType: type, id: typed): untyped =
-  let row = db.getRow(sql(
+  let sqlStmt =
     "SELECT " & columnNamesForModel(modelType).join(",") &
     " FROM " & tableName(modelType) &
-    " WHERE id = ?"), @[$id])
+    " WHERE id = ?"
 
-  if row.allIt(it.len == 0):
-    raise newException(NotFoundError, "no record for id " & $id)
+  log().debug "getRecord: [" & sqlStmt & "] id: " & $id
+  let row = db.getRow(sql(sqlStmt), @[$id])
+
+  if allIt(row, it.len == 0):
+    raise newException(NotFoundError, "no " & modelName(modelType) & " record for id " & $id)
 
   rowToModel(modelType, row)
 
 template findRecordsWhere*(db: DbConn, modelType: type, whereClause: string, values: varargs[string, dbFormat]): untyped =
-  db.getAllRows(sql(
+  let sqlStmt =
     "SELECT " & columnNamesForModel(modelType).join(",") &
     " FROM " & tableName(modelType) &
-    " WHERE " & whereClause), values)
-    .mapIt(rowToModel(modelType, it))
+    " WHERE " & whereClause
+
+  log().debug "findRecordsWhere: [" & sqlStmt & "] values: (" & values.join(", ") & ")"
+  db.getAllRows(sql(sqlStmt), values).mapIt(rowToModel(modelType, it))
 
 template getAllRecords*(db: DbConn, modelType: type): untyped =
-  db.getAllRows(sql(
+  let sqlStmt =
     "SELECT " & columnNamesForModel(modelType).join(",") &
-    " FROM " & tableName(modelType)))
-    .mapIt(rowToModel(modelType, it))
+    " FROM " & tableName(modelType)
+
+  log().debug "getAllRecords: [" & sqlStmt & "]"
+  db.getAllRows(sql(sqlStmt)).mapIt(rowToModel(modelType, it))
 
 template findRecordsBy*(db: DbConn, modelType: type, lookups: seq[tuple[field: string, value: string]]): untyped =
-  db.getAllRows(sql(
+  let sqlStmt =
     "SELECT " & columnNamesForModel(modelType).join(",") &
     " FROM " & tableName(modelType) &
-    " WHERE " & lookups.mapIt(it.field & " = ?").join(" AND ")),
-    lookups.mapIt(it.value))
-  .mapIt(rowToModel(modelType, it))
+    " WHERE " & lookups.mapIt(it.field & " = ?").join(" AND ")
+  let values = lookups.mapIt(it.value)
 
-macro generateProcsForModels*(modelTypes: openarray[type]): untyped =
+  log().debug "findRecordsBy: [" & sqlStmt & "] values (" & values.join(", ") & ")"
+  db.getAllRows(sql(sqlStmt), values).mapIt(rowToModel(modelType, it))
+
+macro generateProcsForModels*(dbType: type, modelTypes: openarray[type]): untyped =
   result = newStmtList()
 
   for t in modelTypes:
@@ -89,23 +126,36 @@ macro generateProcsForModels*(modelTypes: openarray[type]): untyped =
     let deleteName = ident("delete" & modelName)
     let idType = typeOfColumn(t, "id")
     result.add quote do:
-      proc `getName`*(db: PMApiDb, id: `idType`): `t` = getRecord(db.conn, `t`, id)
-      proc `getAllName`*(db: PMApiDb): seq[`t`] = getAllRecords(db.conn, `t`)
-      proc `findWhereName`*(db: PMApiDb, whereClause: string, values: varargs[string, dbFormat]): seq[`t`] =
-        return findRecordsWhere(db.conn, `t`, whereClause, values)
-      proc `createName`*(db: PMApiDb, rec: `t`): `t` = createRecord(db.conn, rec)
-      proc `updateName`*(db: PMApiDb, rec: `t`): bool = updateRecord(db.conn, rec)
-      proc `deleteName`*(db: PMApiDb, rec: `t`): bool = deleteRecord(db.conn, rec)
-      proc `deleteName`*(db: PMApiDb, id: `idType`): bool = deleteRecord(db.conn, `t`, id)
+      proc `getName`*(db: `dbType`, id: `idType`): `t` =
+        db.withConn: result = getRecord(conn, `t`, id)
 
-macro generateLookup*(modelType: type, fields: seq[string]): untyped =
+      proc `getAllName`*(db: `dbType`): seq[`t`] =
+        db.withConn: result = getAllRecords(conn, `t`)
+
+      proc `findWhereName`*(db: `dbType`, whereClause: string, values: varargs[string, dbFormat]): seq[`t`] =
+        db.withConn:
+          result = findRecordsWhere(conn, `t`, whereClause, values)
+
+      proc `createName`*(db: `dbType`, rec: `t`): `t` =
+        db.withConn: result = createRecord(conn, rec)
+
+      proc `updateName`*(db: `dbType`, rec: `t`): bool =
+        db.withConn: result = updateRecord(conn, rec)
+
+      proc `deleteName`*(db: `dbType`, rec: `t`): bool =
+        db.withConn: result = deleteRecord(conn, rec)
+
+      proc `deleteName`*(db: `dbType`, id: `idType`): bool =
+        db.withConn: result = deleteRecord(conn, `t`, id)
+
+macro generateLookup*(dbType: type, modelType: type, fields: seq[string]): untyped =
   let fieldNames = fields[1].mapIt($it)
-  let procName = ident("find" & $modelType.getType[1] & "sBy" & fieldNames.mapIt(it.capitalize).join("And"))
+  let procName = ident("find" & pluralize($modelType.getType[1]) & "By" & fieldNames.mapIt(it.capitalize).join("And"))
 
   # Create proc skeleton
   result = quote do:
-    proc `procName`*(db: PMApiDb): seq[`modelType`] =
-      return findRecordsBy(db.conn, `modelType`)
+    proc `procName`*(db: `dbType`): seq[`modelType`] =
+      db.withConn: result = findRecordsBy(conn, `modelType`)
 
   var callParams = quote do: @[]
 
@@ -115,12 +165,21 @@ macro generateLookup*(modelType: type, fields: seq[string]): untyped =
     paramTuple.add(newColonExpr(ident("field"), newLit(identNameToDb(n))))
     paramTuple.add(newColonExpr(ident("value"), ident(n)))
 
+    # Add the parameter to the outer call (the generated proc)
+    # result[3] is ProcDef -> [3]: FormalParams
     result[3].add(newIdentDefs(ident(n), ident("string")))
+
+    # Build up the AST for the inner procedure call
     callParams[1].add(paramTuple)
 
-  result[6][0][0].add(callParams)
+  # Add the call params to the inner procedure call
+  # result[6][0][1][0][1] is
+  #   ProcDef -> [6]: StmtList (body) -> [0]: Call ->
+  #     [1]: StmtList (withConn body) -> [0]: Asgn (result =) ->
+  #     [1]: Call (inner findRecords invocation)
+  result[6][0][1][0][1].add(callParams)
 
-macro generateProcsForFieldLookups*(modelsAndFields: openarray[tuple[t: type, fields: seq[string]]]): untyped =
+macro generateProcsForFieldLookups*(dbType: type, modelsAndFields: openarray[tuple[t: type, fields: seq[string]]]): untyped =
   result = newStmtList()
 
   for i in modelsAndFields:
@@ -131,20 +190,42 @@ macro generateProcsForFieldLookups*(modelsAndFields: openarray[tuple[t: type, fi
 
     # Create proc skeleton
     let procDefAST = quote do:
-      proc `procName`*(db: PMApiDb): seq[`modelType`] =
-        return findRecordsBy(db.conn, `modelType`)
+      proc `procName`*(db: `dbType`): seq[`modelType`] =
+        db.withConn: result = findRecordsBy(conn, `modelType`)
 
     var callParams = quote do: @[]
 
     # Add dynamic parameters for the proc definition and inner proc call
     for n in fieldNames:
       let paramTuple = newNimNode(nnkPar)
-      paramTuple.add(newColonExpr(ident("field"), newLit(n)))
+      paramTuple.add(newColonExpr(ident("field"), newLit(identNameToDb(n))))
       paramTuple.add(newColonExpr(ident("value"), ident(n)))
 
       procDefAST[3].add(newIdentDefs(ident(n), ident("string")))
       callParams[1].add(paramTuple)
 
-    procDefAST[6][0][0].add(callParams)
+    procDefAST[6][0][1][0][1].add(callParams)
 
     result.add procDefAST
+
+proc initPool*(
+    connect: proc(): DbConn,
+    poolSize = 10,
+    hardCap = false,
+    healthCheckQuery = "SELECT 'true' AS alive"): DbConnPool =
+
+  initDbConnPool(DbConnPoolConfig(
+    connect: connect,
+    poolSize: poolSize,
+    hardCap: hardCap,
+    healthCheckQuery: healthCheckQuery))
+
+template inTransaction*(db: DbConnPool, body: untyped) =
+  pool.withConn(db):
+    conn.exec(sql"BEGIN TRANSACTION")
+    try:
+      body
+      conn.exec(sql"COMMIT")
+    except:
+      conn.exec(sql"ROLLBACK")
+      raise getCurrentException()

src/fiber_orm/pool.nim

@@ -0,0 +1,97 @@
+import std/db_postgres, std/sequtils, std/strutils, std/sugar
+
+import namespaced_logging
+
+
+type
+  DbConnPoolConfig* = object
+    connect*: () -> DbConn
+    poolSize*: int
+    hardCap*: bool
+    healthCheckQuery*: string
+
+  PooledDbConn = ref object
+    conn: DbConn
+    id: int
+    free: bool
+
+  DbConnPool* = ref object
+    conns: seq[PooledDbConn]
+    cfg: DbConnPoolConfig
+    lastId: int
+
+var logNs {.threadvar.}: LoggingNamespace
+
+template log(): untyped =
+  if logNs.isNil: logNs = initLoggingNamespace(name = "fiber_orm/pool", level = lvlNotice)
+  logNs
+
+proc initDbConnPool*(cfg: DbConnPoolConfig): DbConnPool =
+  log().debug("Initializing new pool (size: " & $cfg.poolSize)
+  result = DbConnPool(
+    conns: @[],
+    cfg: cfg)
+
+proc newConn(pool: DbConnPool): PooledDbConn =
+  log().debug("Creating a new connection to add to the pool.")
+  pool.lastId += 1
+  let conn = pool.cfg.connect()
+  result = PooledDbConn(
+    conn: conn,
+    id: pool.lastId,
+    free: true)
+  pool.conns.add(result)
+
+proc maintain(pool: DbConnPool): void =
+  log().debug("Maintaining pool. $# connections." % [$pool.conns.len])
+  pool.conns.keepIf(proc (pc: PooledDbConn): bool =
+    if not pc.free: return true
+
+    try:
+      discard getRow(pc.conn, sql(pool.cfg.healthCheckQuery), [])
+      return true
+    except:
+      try: pc.conn.close()  # try to close the connection
+      except: discard ""
+      return false
+  )
+  log().debug(
+    "Pruned dead connections. $# connections remaining." %
+    [$pool.conns.len])
+
+  let freeConns = pool.conns.filterIt(it.free)
+  if pool.conns.len > pool.cfg.poolSize and freeConns.len > 0:
+    let numToCull = min(freeConns.len, pool.conns.len - pool.cfg.poolSize)
+    let toCull = freeConns[0..numToCull]
+    pool.conns.keepIf((pc) => toCull.allIt(it.id != pc.id))
+    for culled in toCull:
+      try: culled.conn.close()
+      except: discard ""
+    log().debug(
+      "Trimming pool size. Culled $# free connections. $# connections remaining." %
+      [$toCull.len, $pool.conns.len])
+
+proc take*(pool: DbConnPool): tuple[id: int, conn: DbConn] =
+  pool.maintain
+  let freeConns = pool.conns.filterIt(it.free)
+
+  log().debug(
+    "Providing a new connection ($# currently free)." % [$freeConns.len])
+
+  let reserved =
+    if freeConns.len > 0: freeConns[0]
+    else: pool.newConn()
+
+  reserved.free = false
+  log().debug("Reserve connection $#" % [$reserved.id])
+  return (id: reserved.id, conn: reserved.conn)
+
+proc release*(pool: DbConnPool, connId: int): void =
+  log().debug("Reclaiming released connaction $#" % [$connId])
+  let foundConn = pool.conns.filterIt(it.id == connId)
+  if foundConn.len > 0: foundConn[0].free = true
+
+template withConn*(pool: DbConnPool, stmt: untyped): untyped =
+  let (connId, conn {.inject.}) = take(pool)
+  stmt
+  release(pool, connId)

src/fiber_orm/util.nim

@@ -1,13 +1,7 @@
 import json, macros, options, sequtils, strutils, times, timeutils, unicode,
   uuids
 
-const UNDERSCORE_RUNE = "_".toRunes[0]
-const PG_TIMESTAMP_FORMATS = [
-  "yyyy-MM-dd HH:mm:sszz",
-  "yyyy-MM-dd HH:mm:ss'.'fzz",
-  "yyyy-MM-dd HH:mm:ss'.'ffzz",
-  "yyyy-MM-dd HH:mm:ss'.'fffzz"
-]
+import nre except toSeq
 
 type
   MutateClauses* = object
@@ -17,7 +11,7 @@ type
 
 # TODO: more complete implementation
 # see https://github.com/blakeembrey/pluralize
-proc pluralize(name: string): string =
+proc pluralize*(name: string): string =
   if name[^2..^1] == "ey": return name[0..^3] & "ies"
   if name[^1] == 'y': return name[0..^2] & "ies"
   return name & "s"
@@ -28,7 +22,9 @@ macro modelName*(model: object): string =
 macro modelName*(modelType: type): string =
   return newStrLitNode($modelType.getType[1])
 
+
 proc identNameToDb*(name: string): string =
+  const UNDERSCORE_RUNE = "_".toRunes[0]
   let nameInRunes = name.toRunes
   var prev: Rune
   var resultRunes = newSeq[Rune]()
@@ -68,10 +64,42 @@ type DbArrayParseState = enum
   expectStart, inQuote, inVal, expectEnd
 
 proc parsePGDatetime*(val: string): DateTime =
+
+  const PG_TIMESTAMP_FORMATS = [
+    "yyyy-MM-dd HH:mm:ss",
+    "yyyy-MM-dd'T'HH:mm:ss",
+    "yyyy-MM-dd HH:mm:sszz",
+    "yyyy-MM-dd'T'HH:mm:sszz",
+    "yyyy-MM-dd HH:mm:ss'.'fff",
+    "yyyy-MM-dd'T'HH:mm:ss'.'fff",
+    "yyyy-MM-dd HH:mm:ss'.'fffzz",
+    "yyyy-MM-dd'T'HH:mm:ss'.'fffzz",
+    "yyyy-MM-dd HH:mm:ss'.'fffzzz",
+    "yyyy-MM-dd'T'HH:mm:ss'.'fffzzz",
+  ]
+
+  var correctedVal = val;
+
+  # PostgreSQL will truncate any trailing 0's in the millisecond value leading
+  # to values like `2020-01-01 16:42.3+00`. This cannot currently be parsed by
+  # the standard times format as it expects exactly three digits for
+  # millisecond values. So we have to detect this and pad out the millisecond
+  # value to 3 digits.
+  let PG_PARTIAL_FORMAT_REGEX = re"(\d{4}-\d{2}-\d{2}( |'T')\d{2}:\d{2}:\d{2}\.)(\d{1,2})(\S+)?"
+  let match = val.match(PG_PARTIAL_FORMAT_REGEX)
+
+  if match.isSome:
+    let c = match.get.captures
+    if c.toSeq.len == 2: correctedVal = c[0] & alignLeft(c[2], 3, '0')
+    else: correctedVal = c[0] & alignLeft(c[2], 3, '0') & c[3]
+
   var errStr = ""
+
+  # Try to parse directly using known format strings.
   for df in PG_TIMESTAMP_FORMATS:
-    try: return val.parse(df)
-    except: errStr &= "\n" & getCurrentExceptionMsg()
+    try: return correctedVal.parse(df)
+    except: errStr &= "\n\t" & getCurrentExceptionMsg()
+
   raise newException(ValueError, "Cannot parse PG date. Tried:" & errStr)
 
 proc parseDbArray*(val: string): seq[string] =
@@ -144,7 +172,11 @@ proc createParseStmt*(t, value: NimNode): NimNode =
       result = quote do: parsePGDatetime(`value`)
 
     elif t.getTypeInst == Option.getType:
-      let innerType = t.getTypeImpl[2][0][0][1]
+      var innerType = t.getTypeImpl[2][0] # start at the first RecList
+      # If the value is a non-pointer type, there is another inner RecList
+      if innerType.kind == nnkRecList: innerType = innerType[0]
+      innerType = innerType[1] # now we can take the field type from the first symbol
+
       let parseStmt = createParseStmt(innerType, value)
       result = quote do:
         if `value`.len == 0:  none[`innerType`]()
@@ -173,9 +205,16 @@ proc createParseStmt*(t, value: NimNode): NimNode =
   elif t.typeKind == ntyInt:
     result = quote do: parseInt(`value`)
 
+  elif t.typeKind == ntyFloat:
+    result = quote do: parseFloat(`value`)
+
   elif t.typeKind == ntyBool:
     result = quote do: "true".startsWith(`value`.toLower)
 
+  elif t.typeKind == ntyEnum:
+    let innerType = t.getTypeInst
+    result = quote do: parseEnum[`innerType`](`value`)
+
   else:
     error "Unknown value type: " & $t.typeKind
 
@@ -245,7 +284,9 @@ proc typeOfColumn*(modelType: NimNode, colName: string): NimNode =
   raise newException(Exception,
     "model of type '" & $modelType & "' has no column named '" & colName & "'")
 
-proc isZero(val: int): bool = return val == 0
+proc isEmpty(val: int): bool = return val == 0
+proc isEmpty(val: UUID): bool = return val.isZero
+proc isEmpty(val: string): bool = return val.isEmptyOrWhitespace
 
 macro populateMutateClauses*(t: typed, newRecord: bool, mc: var MutateClauses): untyped =
 
@@ -257,14 +298,14 @@ macro populateMutateClauses*(t: typed, newRecord: bool, mc: var MutateClauses):
       # grab the field, it's string name, and it's type
       let fieldName = $fieldIdent
 
-      # we do not update the ID, but we do check: if we're creating a new
-      # record, we should not have an existing ID
+      # We only add clauses for the ID field if we're creating a new record and
+      # the caller provided a value..
       if fieldName == "id":
         result.add quote do:
-          if `newRecord` and not `t`.id.isZero:
-            raise newException(
-              AssertionError,
-              "Trying to create a new record, but the record already has an ID (" & $(`t`.id) & ").")
+          if `newRecord` and not `t`.id.isEmpty:
+            `mc`.columns.add(identNameToDb(`fieldName`))
+            `mc`.placeholders.add("?")
+            `mc`.values.add(dbFormat(`t`.`fieldIdent`))
 
       # if we're looking at an optional field, add logic to check for presence
       elif fieldType.kind == nnkBracketExpr and
@@ -273,7 +314,7 @@ macro populateMutateClauses*(t: typed, newRecord: bool, mc: var MutateClauses):
 
         result.add quote do:
           `mc`.columns.add(identNameToDb(`fieldName`))
-          if `t`.`fieldIdent`.isSome:
+          if isSome(`t`.`fieldIdent`):
             `mc`.placeholders.add("?")
             `mc`.values.add(dbFormat(`t`.`fieldIdent`.get))
           else: