introduce methods close and query in the base API and refactor accordingly

datastore/datastore.nim

@@ -13,33 +13,37 @@ push: {.upraises: [].}
 type
   Datastore* = ref object of RootObj
 
+  QueryIterator* = iterator (
+    datastore: Datastore,
+    query: Query): Future[QueryResponse] {.closure, gcsafe.}
+
+method close*(self: Datastore): Future[void] {.base, locks: "unknown".} =
+  raiseAssert("Not implemented!")
+
 method contains*(
   self: Datastore,
-  key: Key): Future[?!bool] {.async, base, locks: "unknown".} =
+  key: Key): Future[?!bool] {.base, locks: "unknown".} =
 
   raiseAssert("Not implemented!")
 
 method delete*(
   self: Datastore,
-  key: Key): Future[?!void] {.async, base, locks: "unknown".} =
+  key: Key): Future[?!void] {.base, locks: "unknown".} =
 
   raiseAssert("Not implemented!")
 
 method get*(
   self: Datastore,
-  key: Key): Future[?!(?seq[byte])] {.async, base, locks: "unknown".} =
+  key: Key): Future[?!(?seq[byte])] {.base, locks: "unknown".} =
 
   raiseAssert("Not implemented!")
 
 method put*(
   self: Datastore,
   key: Key,
-  data: seq[byte]): Future[?!void] {.async, base, locks: "unknown".} =
+  data: seq[byte]): Future[?!void] {.base, locks: "unknown".} =
 
   raiseAssert("Not implemented!")
 
-iterator query*(
-  self: Datastore,
-  query: Query): Future[QueryResponse] =
-
+method query*(self: Datastore): QueryIterator {.base, locks: "unknown".} =
   raiseAssert("Not implemented!")

datastore/filesystem_datastore.nim

@@ -59,6 +59,9 @@ proc path*(
 
   self.root / joinPath(segments) & objExt
 
+method close*(self: FileSystemDatastore) {.async, locks: "unknown".} =
+  discard
+
 method contains*(
   self: FileSystemDatastore,
   key: Key): Future[?!bool] {.async, locks: "unknown".} =
@@ -154,9 +157,3 @@ method put*(
 
   except OSError as e:
     return failure e
-
-# method query*(
-#   self: FileSystemDatastore,
-#   query: ...): Future[?!(?...)] {.async, locks: "unknown".} =
-#
-#   return success ....some

datastore/null_datastore.nim

@@ -15,6 +15,9 @@ type
 proc new*(T: type NullDatastore): T =
   T()
 
+method close*(self: NullDatastore) {.async, locks: "unknown".} =
+  discard
+
 method contains*(
   self: NullDatastore,
   key: Key): Future[?!bool] {.async, locks: "unknown".} =
@@ -40,8 +43,11 @@ method put*(
 
   return success()
 
-iterator query*(
-  self: NullDatastore,
-  query: Query): Future[QueryResponse] =
+iterator queryImpl(
+  datastore: Datastore,
+  query: Query): Future[QueryResponse] {.closure.} =
 
   discard
+
+method query*(self: NullDatastore): QueryIterator {.locks: "unknown".} =
+  queryImpl

datastore/sqlite_datastore.nim

@@ -284,7 +284,10 @@ proc dbPath*(self: SQLiteDatastore): string =
 proc env*(self: SQLiteDatastore): SQLite =
   self.env
 
-proc close*(self: SQLiteDatastore) =
+proc timestamp*(t = epochTime()): int64 =
+  (t * 1_000_000).int64
+
+method close*(self: SQLiteDatastore) {.async, locks: "unknown".} =
   self.containsStmt.dispose
   self.getStmt.dispose
 
@@ -295,9 +298,6 @@ proc close*(self: SQLiteDatastore) =
   self.env.dispose
   self[] = SQLiteDatastore()[]
 
-proc timestamp*(t = epochTime()): int64 =
-  (t * 1_000_000).int64
-
 method contains*(
   self: SQLiteDatastore,
   key: Key): Future[?!bool] {.async, locks: "unknown".} =
@@ -367,13 +367,14 @@ method put*(
 
   return await self.put(key, data, timestamp())
 
-iterator query*(
-  self: SQLiteDatastore,
-  query: Query): Future[QueryResponse] =
+iterator queryImpl(
+  datastore: Datastore,
+  query: Query): Future[QueryResponse] {.closure.} =
 
   let
+    datastore = SQLiteDatastore(datastore)
     queryStmt = QueryStmt.prepare(
-      self.env, queryStmtStr).expect("should not fail")
+      datastore.env, queryStmtStr).expect("should not fail")
 
     s = RawStmtPtr(queryStmt)
 
@@ -428,3 +429,6 @@ iterator query*(
       break
     else:
       raise (ref Defect)(msg: $sqlite3_errstr(v))
+
+method query*(self: SQLiteDatastore): QueryIterator {.locks: "unknown".} =
+  queryImpl

datastore/tiered_datastore.nim

@@ -28,6 +28,10 @@ proc new*(
 proc stores*(self: TieredDatastore): seq[Datastore] =
   self.stores
 
+method close*(self: TieredDatastore) {.async, locks: "unknown".} =
+  for store in self.stores:
+    await store.close
+
 method contains*(
   self: TieredDatastore,
   key: Key): Future[?!bool] {.async, locks: "unknown".} =
@@ -49,7 +53,10 @@ method delete*(
     pending = await allFinished(self.stores.mapIt(it.delete(key)))
 
   for fut in pending:
-    if fut.read().isErr: return fut.read()
+    let
+      delRes = await fut
+
+    if delRes.isErr: return delRes
 
   return success()
 
@@ -90,12 +97,26 @@ method put*(
     pending = await allFinished(self.stores.mapIt(it.put(key, data)))
 
   for fut in pending:
-    if fut.read().isErr: return fut.read()
+    let
+      putRes = await fut
+
+    if putRes.isErr: return putRes
 
   return success()
 
-# method query*(
-#   self: TieredDatastore,
-#   query: ...): Future[?!(?...)] {.async, locks: "unknown".} =
-#
-#   return success ....some
+iterator queryImpl(
+  datastore: Datastore,
+  query: Query): Future[QueryResponse] {.closure.} =
+
+  let
+    datastore = TieredDatastore(datastore)
+    # https://github.com/datastore/datastore/blob/7ccf0cd4748001d3dbf5e6dda369b0f63e0269d3/datastore/core/basic.py#L1027-L1035
+    bottom = datastore.stores[^1]
+
+  try:
+    let q = bottom.query(); for kv in q(bottom, query): yield kv
+  except Exception as e:
+    raise (ref Defect)(msg: e.msg)
+
+method query*(self: TieredDatastore): QueryIterator {.locks: "unknown".} =
+  queryImpl

tests/datastore/test_datastore.nim

@@ -25,5 +25,4 @@ suite "Datastore (base)":
     expect Defect: discard ds.get(key)
 
   asyncTest "query":
-    expect Defect:
-      for n in ds.query(Query.init(key)): discard
+    expect Defect: discard ds.query

tests/datastore/test_filesystem_datastore.nim

@@ -190,7 +190,3 @@ suite "FileSystemDatastore":
     getOpt = getRes.get
 
     check: getOpt.isNone
-
-  # asyncTest "query":
-  #   check:
-  #     true

tests/datastore/test_null_datastore.nim

@@ -35,7 +35,7 @@ suite "NullDatastore":
     var
       x = true
 
-    for n in ds.query(Query.init(key)):
+    let q = ds.query; for n in q(ds, Query.init(key)):
       # `iterator query` for NullDatastore never yields so the following lines
       # are not run (else the test would hang)
       x = false