Merge pull request #5 from armanbilge/pr/helpers

Add helpers on `Database` and `Statement`

core/js/src/main/scala/porcupine/dbplatform.scala

@@ -33,7 +33,7 @@ private abstract class DatabasePlatform:
         .make(F.delay(new sqlite3.Database(filename)))(db => F.delay(db.close()))
         .evalTap(db => F.delay(db.defaultSafeIntegers(true)))
         .map { db =>
-          new:
+          new AbstractDatabase[F]:
             def prepare[A, B](query: Query[A, B]): Resource[F, Statement[F, A, B]] =
               Resource
                 .eval(mutex.lock.surround(F.delay(db.prepare(query.sql))))
@@ -48,7 +48,7 @@ private abstract class DatabasePlatform:
                   }
 
                   if statement.reader then
-                    new:
+                    new AbstractStatement[F, A, B]:
                       def cursor(args: A): Resource[F, Cursor[F, B]] = mutex.lock *>
                         Resource
                           .eval {
@@ -87,10 +87,11 @@ private abstract class DatabasePlatform:
 
                           }
                   else
-                    args =>
-                      mutex.lock *> Resource.eval {
-                        F.delay(statement.run(bind(args)*)).as(_ => F.pure(Nil, false))
-                      }
+                    new AbstractStatement[F, A, B]:
+                      def cursor(args: A): Resource[F, Cursor[F, B]] =
+                        mutex.lock *> Resource.eval {
+                          F.delay(statement.run(bind(args)*)).as(_ => F.pure(Nil, false))
+                        }
 
                 }
 

core/jvm/src/main/scala/porcupine/dbplatform.scala

@@ -31,7 +31,7 @@ private abstract class DatabasePlatform:
       mutex <- Resource.eval(Mutex[F])
       connection <- Resource
         .fromAutoCloseable(F.blocking(DriverManager.getConnection("jdbc:sqlite:" + filename)))
-    yield new:
+    yield new AbstractDatabase[F]:
       def prepare[A, B](query: Query[A, B]): Resource[F, Statement[F, A, B]] =
         Resource
           .fromAutoCloseable {
@@ -40,7 +40,7 @@ private abstract class DatabasePlatform:
             }
           }
           .map { statement =>
-            new:
+            new AbstractStatement[F, A, B]:
               def cursor(args: A): Resource[F, Cursor[F, B]] = mutex.lock *>
                 Resource
                   .make {

core/native/src/main/scala/porcupine/dbplatform.scala

@@ -46,7 +46,7 @@ private abstract class DatabasePlatform:
           }
         }(db => F.blocking(guard(sqlite3_close(db))))
         .map { db =>
-          new:
+          new AbstractDatabase[F]:
             def prepare[A, B](query: Query[A, B]): Resource[F, Statement[F, A, B]] =
               Resource
                 .make {
@@ -61,76 +61,79 @@ private abstract class DatabasePlatform:
                 } { stmt =>
                   mutex.lock.surround(F.delay(guard(db)(sqlite3_finalize(stmt))))
                 }
-                .map { stmt => args =>
-                  mutex.lock *> Resource
-                    .make {
-                      F.delay {
-                        var i = 1
-                        query.encoder.encode(args).flatMap {
-                          case LiteValue.Null =>
-                            guard(db)(sqlite3_bind_null(stmt, i))
-                            i += 1
-                            Nil
-                          case LiteValue.Integer(j) =>
-                            guard(db)(sqlite3_bind_int64(stmt, i, j))
-                            i += 1
-                            Nil
-                          case LiteValue.Real(d) =>
-                            guard(db)(sqlite3_bind_double(stmt, i, d))
-                            i += 1
-                            Nil
-                          case LiteValue.Text(s) =>
-                            val b = s.getBytes
-                            guard(db)(
-                              sqlite3_bind_text(stmt, i, b.at(0), b.length, SQLITE_STATIC),
-                            )
-                            i += 1
-                            List(b)
-                          case LiteValue.Blob(b) =>
-                            val ba = b.toArray
-                            guard(db)(
-                              sqlite3_bind_blob64(stmt, i, ba.at(0), ba.length, SQLITE_STATIC),
-                            )
-                            i += 1
-                            List(ba)
+                .map { stmt =>
+                  new AbstractStatement[F, A, B]:
+                    def cursor(args: A): Resource[F, Cursor[F, B]] = mutex.lock *> Resource
+                      .make {
+                        F.delay {
+                          var i = 1
+                          query.encoder.encode(args).flatMap {
+                            case LiteValue.Null =>
+                              guard(db)(sqlite3_bind_null(stmt, i))
+                              i += 1
+                              Nil
+                            case LiteValue.Integer(j) =>
+                              guard(db)(sqlite3_bind_int64(stmt, i, j))
+                              i += 1
+                              Nil
+                            case LiteValue.Real(d) =>
+                              guard(db)(sqlite3_bind_double(stmt, i, d))
+                              i += 1
+                              Nil
+                            case LiteValue.Text(s) =>
+                              val b = s.getBytes
+                              guard(db)(
+                                sqlite3_bind_text(stmt, i, b.at(0), b.length, SQLITE_STATIC),
+                              )
+                              i += 1
+                              List(b)
+                            case LiteValue.Blob(b) =>
+                              val ba = b.toArray
+                              guard(db)(
+                                sqlite3_bind_blob64(stmt, i, ba.at(0), ba.length, SQLITE_STATIC),
+                              )
+                              i += 1
+                              List(ba)
+                          }
                         }
-                      }
-                    }(x => F.delay(x).void) // to keep in sight of gc
-                    .as { maxRows =>
-                      F.blocking {
-                        val rows = List.newBuilder[List[LiteValue]]
-                        var i = 0
-                        var continue = true
-                        while i < maxRows && continue do
-                          sqlite3_step(stmt) match
-                            case SQLITE_ROW =>
-                              rows += List.tabulate(sqlite3_column_count(stmt)) { j =>
-                                sqlite3_column_type(stmt, j) match
-                                  case SQLITE_NULL =>
-                                    LiteValue.Null
-                                  case SQLITE_INTEGER =>
-                                    LiteValue.Integer(sqlite3_column_int64(stmt, j))
-                                  case SQLITE_FLOAT =>
-                                    LiteValue.Real(sqlite3_column_double(stmt, j))
-                                  case SQLITE_TEXT =>
-                                    LiteValue.Text(fromCString(sqlite3_column_text(stmt, j)))
-                                  case SQLITE_BLOB =>
-                                    LiteValue.Blob(
-                                      ByteVector.fromPtr(
-                                        sqlite3_column_blob(stmt, j),
-                                        sqlite3_column_bytes(stmt, j),
-                                      ),
-                                    )
-                              }
-                            case SQLITE_DONE => continue = false
-                            case other => guard(db)(other)
+                      }(x => F.delay(x).void) // to keep in sight of gc
+                      .as { maxRows =>
+                        F.blocking {
+                          val rows = List.newBuilder[List[LiteValue]]
+                          var i = 0
+                          var continue = true
+                          while i < maxRows && continue do
+                            sqlite3_step(stmt) match
+                              case SQLITE_ROW =>
+                                rows += List.tabulate(sqlite3_column_count(stmt)) { j =>
+                                  sqlite3_column_type(stmt, j) match
+                                    case SQLITE_NULL =>
+                                      LiteValue.Null
+                                    case SQLITE_INTEGER =>
+                                      LiteValue.Integer(sqlite3_column_int64(stmt, j))
+                                    case SQLITE_FLOAT =>
+                                      LiteValue.Real(sqlite3_column_double(stmt, j))
+                                    case SQLITE_TEXT =>
+                                      LiteValue.Text(fromCString(sqlite3_column_text(stmt, j)))
+                                    case SQLITE_BLOB =>
+                                      LiteValue.Blob(
+                                        ByteVector.fromPtr(
+                                          sqlite3_column_blob(stmt, j),
+                                          sqlite3_column_bytes(stmt, j),
+                                        ),
+                                      )
+                                }
+                              case SQLITE_DONE => continue = false
+                              case other => guard(db)(other)
 
-                          i += 1
-                        (rows.result(), continue)
-                      }.flatMap { (rows, continue) =>
-                        rows.traverse(query.decoder.decode.runA(_).liftTo).tupleRight(continue)
+                            i += 1
+                          (rows.result(), continue)
+                        }.flatMap { (rows, continue) =>
+                          rows
+                            .traverse(query.decoder.decode.runA(_).liftTo)
+                            .tupleRight(continue)
+                        }
                       }
-                    }
                 }
         }
     }

core/shared/src/main/scala/porcupine/db.scala

@@ -16,15 +16,122 @@
 
 package porcupine
 
+import cats.effect.kernel.MonadCancelThrow
 import cats.effect.kernel.Resource
+import cats.syntax.all.*
+import fs2.Chunk
+import fs2.Pipe
+import fs2.Stream
+
+import java.util.NoSuchElementException
+import scala.annotation.targetName
 
 abstract class Database[F[_]] private[porcupine]:
   def prepare[A, B](query: Query[A, B]): Resource[F, Statement[F, A, B]]
 
+  final def cursor[A, B](query: Query[A, B], args: A): Resource[F, Cursor[F, B]] =
+    prepare(query).flatMap(_.cursor(args))
+
+  def execute[A, B](query: Query[A, B], args: A): F[List[B]]
+
+  final def execute[A](query: Query[Unit, A]): F[List[A]] = execute(query, ())
+
+  @targetName("executeVoid")
+  def execute[A](query: Query[A, Unit], args: A): F[Unit]
+
+  @targetName("executeVoid")
+  final def execute(query: Query[Unit, Unit]): F[Unit] = execute(query, ())
+
+  def option[A, B](query: Query[A, B], args: A): F[Option[B]]
+
+  final def option[A](query: Query[Unit, A]): F[Option[A]] = option(query, ())
+
+  def unique[A, B](query: Query[A, B], args: A): F[B]
+
+  final def unique[A](query: Query[Unit, A]): F[A] = unique(query, ())
+
+  def stream[A, B](query: Query[A, B], args: A, chunkSize: Int): Stream[F, B]
+
+  def pipe[A, B](query: Query[A, B], chunkSize: Int): Pipe[F, A, B]
+
+  final def pipe[A](query: Query[A, Unit], args: A): Pipe[F, A, Nothing] =
+    in => pipe(query, 1)(in).drain
+
+private abstract class AbstractDatabase[F[_]](using F: MonadCancelThrow[F]) extends Database[F]:
+
+  final def execute[A, B](query: Query[A, B], args: A) = prepare(query).use(_.execute(args))
+
+  @targetName("executeVoid")
+  final def execute[A](query: Query[A, Unit], args: A) = prepare(query).use(_.execute(args))
+
+  final def option[A, B](query: Query[A, B], args: A) = prepare(query).use(_.option(args))
+
+  final def unique[A, B](query: Query[A, B], args: A) = prepare(query).use(_.unique(args))
+
+  final def stream[A, B](query: Query[A, B], args: A, chunkSize: Int) =
+    Stream.resource(prepare(query)).flatMap(_.stream(args, chunkSize))
+
+  final def pipe[A, B](query: Query[A, B], chunkSize: Int) =
+    in => Stream.resource(prepare(query)).flatMap(_.pipe(chunkSize)(in))
+
 object Database extends DatabasePlatform
 
 abstract class Statement[F[_], A, B] private[porcupine]:
   def cursor(args: A): Resource[F, Cursor[F, B]]
 
+  def execute(args: A): F[List[B]]
+
+  final def execute(using ev: Unit <:< A): F[List[B]] = execute(())
+
+  def execute(using Unit <:< B)(args: A): F[Unit]
+
+  final def execute(using Unit <:< A, Unit <:< B): F[Unit] = execute(())
+
+  def option(args: A): F[Option[B]]
+
+  final def option(using Unit <:< A): F[Option[B]] = option(())
+
+  def unique(args: A): F[B]
+
+  final def unique(using Unit <:< A): F[B] = unique(())
+
+  def stream(args: A, chunkSize: Int): Stream[F, B]
+
+  final def pipe(chunkSize: Int): Pipe[F, A, B] =
+    _.flatMap(stream(_, chunkSize))
+
+  final def pipe(using Unit <:< B): Pipe[F, A, Nothing] =
+    in => pipe(1)(in).drain
+
+private abstract class AbstractStatement[F[_], A, B](using F: MonadCancelThrow[F])
+    extends Statement[F, A, B]:
+
+  final def execute(args: A) = cursor(args).use(_.fetch(Int.MaxValue).map(_._1))
+
+  final def execute(using Unit <:< B)(args: A) = cursor(args).use(_.fetch(1).void)
+
+  final def option(args: A) = cursor(args).use(_.fetch(1).flatMap {
+    case (Nil, false) => F.pure(None)
+    case (head :: Nil, false) => F.pure(Some(head))
+    case _ => F.raiseError(new RuntimeException("More than 1 row"))
+  })
+
+  final def unique(args: A) = cursor(args).use(_.fetch(1).flatMap {
+    case (Nil, false) => F.raiseError(new NoSuchElementException)
+    case (head :: Nil, false) => F.pure(head)
+    case _ => F.raiseError(new RuntimeException("More than 1 row"))
+  })
+
+  final def stream(args: A, chunkSize: Int) =
+    Stream.resource(cursor(args)).flatMap { cursor =>
+      Stream
+        .unfoldLoopEval(()) { _ =>
+          cursor.fetch(chunkSize).map { (chunk, more) =>
+            (Chunk.seq(chunk), Option.when(more)(()))
+          }
+        }
+        .unchunks
+    }
+
 abstract class Cursor[F[_], A] private[porcupine]:
   def fetch(maxRows: Int): F[(List[A], Boolean)]

core/shared/src/test/scala/porcupine/PorcupineTest.scala

@@ -26,23 +26,16 @@ import Codec.*
 object PorcupineTest extends IOApp.Simple:
 
   def run = Database.open[IO](":memory:").use { db =>
-    db.prepare(sql"create table porcupine (n, i, r, t, b);".command).use {
-      _.cursor(()).use(_.fetch(1).void)
-    } *>
-      db.prepare(
+    db.execute(sql"create table porcupine (n, i, r, t, b);".command) *>
+      db.execute(
         sql"insert into porcupine values(${`null`}, $integer, $real, $text, $blob);".command,
-      ).use(
-        _.cursor((None, 42, 3.14, "quill-pig", ByteVector(0, 1, 2, 3))).use(_.fetch(1).void),
+        (None, 42L, 3.14, "quill-pig", ByteVector(0, 1, 2, 3)),
       ) *>
-      db.prepare(
+      db.execute(
         sql"select b, t, r, i, n from porcupine;"
           .query(blob *: text *: real *: integer *: `null` *: nil),
-      ).use {
-        _.cursor(()).use {
-          _.fetch(100).flatMap {
-            case (List((ByteVector(0, 1, 2, 3), "quill-pig", 3.14, 42, None)), false) => IO.unit
-            case other => IO.raiseError(new AssertionError(other))
-          }
-        }
+      ).flatMap {
+        case List((ByteVector(0, 1, 2, 3), "quill-pig", 3.14, 42, None)) => IO.unit
+        case other => IO.raiseError(new AssertionError(other))
       }
   }