Skip to main content

How to Use the sql Module from a ZIO Application

In this guide, we connect the ZIO Blocks sql module to a ZIO application. By the end, you will have a bookstore BookRepo service that ZIO wires with ZLayer, a sale that writes to the database and calls a payment service inside one transaction, error handling with retries and timeouts, and a ZIO Test spec that gives every test its own database.

We start from a database provided as a layer, then add one capability at a time:

  • A TransactorZIO provided as a ZLayer, for SQLite and for a pooled PostgreSQL database
  • A ZIO service that runs its queries on ZIO's blocking thread pool
  • A transaction whose body is a ZIO effect, and the rollback when that effect fails
  • Typed errors, retries and timeouts with ZIO's own operators
  • A ZIO Test layer that gives each test a fresh database
  • Reading a large result set without a ZStream

This guide builds on Build Your First Database Repository on SQLite. We use Table, Repo, DbCon and sql"..." the way that tutorial introduces them, and assume you know ZIO, ZLayer and ZIO.serviceWithZIO.

The Problem​

The sql module is plain blocking Scala: JdbcTransactor#connect and JdbcTransactor#transact hold a thread until the database answers. A ZIO application runs its fibers on a small pool of threads, one per CPU core, and expects blocking work to move to a separate blocking pool. A JDBC call made on the main pool stalls every fiber scheduled behind it.

A first attempt usually wraps the transactor by hand:

import zio.*
import zio.blocks.schema.Schema
import zio.blocks.sql.*

case class Book(id: Long, title: String, year: Int, stock: Int)
object Book { given Schema[Book] = Schema.derived }

val books      = Repo.derived[Book, Long]
val transactor = JdbcTransactor.fromUrl("jdbc:sqlite:bookstore.db", SqlDialect.SQLite)

// Runs JDBC on ZIO's main thread pool
val allBooks: Task[List[Book]] =
  ZIO.attempt(transactor.connect(books.all))

// Correct pool, but every call site has to remember it
val count: Task[Long] =
  ZIO.attemptBlocking(transactor.connect(books.count))

// No way to call another ZIO effect between two statements of one transaction
def sell(id: Long): Task[Int] =
  ZIO.attemptBlocking(transactor.transact(sql"UPDATE book SET stock = stock - 1 WHERE id = $id".update))

Every query needs the right wrapper, and one forgotten ZIO.attemptBlocking slows down the whole application without failing a test. The transaction body is synchronous, so a sale that must also charge a customer through a ZIO service can't do both in one transaction.

In this guide, we solve this with the sql-zio module. Its TransactorZIO returns ZIO effects that already run on the blocking pool, comes as a ZLayer, and has TransactorZIO#transactZIO for transactions whose body is a ZIO effect.

Prerequisites​

Add sql-zio and the SQLite JDBC driver to your build.sbt. sql-zio brings in the sql module and ZIO 2.1.27:

// Scala 3.3 to 3.8
libraryDependencies += "dev.zio" %% "zio-blocks-sql-zio" % "0.0.56"

// Scala 3.9 and newer, which also needs -experimental
libraryDependencies += "dev.zio" %% "zio-blocks-next-sql-zio" % "0.0.56"
scalacOptions += "-experimental"

// The SQLite JDBC driver, for every Scala version
libraryDependencies += "org.xerial" % "sqlite-jdbc" % "3.53.4.0"

sql and sql-zio are published for Scala 3 only and run on the JVM. Run your application on JDK 21 or newer.

We import ZIO, the ZIO integration and the sql module:

import zio.*
import zio.blocks.schema.Schema
import zio.blocks.sql.zio.*
import zio.blocks.sql.*
Put import zio.blocks.sql.* last

Once sql-zio is on the classpath, import zio.blocks.sql.* brings the subpackage zio.blocks.sql.zio into scope under the name zio. After it, zio no longer means the root zio package, so every later import that starts with zio, such as import zio.blocks.sql.zio.* or import zio.blocks.schema.Schema, fails with value blocks is not a member of zio.blocks.sql.zio, and so does a fully qualified name such as zio.Duration. This also breaks existing files that compiled before you added sql-zio. Put import zio.blocks.sql.* after every other zio import, as above, or write _root_.zio where you need the root package after it.

This page runs its effects inside mdoc, so we use a small helper that runs an effect and returns its result. In an application, ZIOAppDefault#run does this for you:

def runDemo[A](effect: ZIO[Any, Any, A]): A =
  Unsafe.unsafe { implicit unsafe =>
    Runtime.default.unsafe.run(effect).getOrThrowFiberFailure()
  }

The Bookstore Model​

We keep a bookstore's stock in a book table. Each Book has an ID, a title, the year it was published and the copies on the shelf. The stock numbers are made up:

case class Book(id: Long, title: String, year: Int, stock: Int)

object Book {
  given Schema[Book] = Schema.derived
}

val books: Repo[Book, Long] = Repo.derived[Book, Long]

val shelf = List(
  Book(1L, "Zionomicon", 2024, 5),
  Book(2L, "Programming in Scala", 2008, 3),
  Book(3L, "Functional Programming in Scala", 2014, 4)
)

def createBookTable()(using DbCon): Unit =
  sql"""CREATE TABLE book (
    id INTEGER PRIMARY KEY,
    title TEXT NOT NULL,
    year INTEGER NOT NULL,
    stock INTEGER NOT NULL
  )""".update

We write the CREATE TABLE statement by hand instead of using Table#createTable, because the derived statement has no PRIMARY KEY, as How to Test Database Code Against SQLite explains. createBookTable takes a (using DbCon), so it runs inside any TransactorZIO#connect or TransactorZIO#transact block.

Provide the Database as a ZLayer​

ZIO applications get their dependencies from layers, so the database is a layer too. TransactorZIO.layer takes a JDBC URL and a dialect and returns a ZLayer[Any, Nothing, TransactorZIO]. We point it at a temporary SQLite file:

val dbFile = java.nio.file.Files.createTempFile("bookstore", ".db")
dbFile.toFile.deleteOnExit()

val transactorLayer: ZLayer[Any, Nothing, TransactorZIO] =
  TransactorZIO.layer(s"jdbc:sqlite:$dbFile", SqlDialect.SQLite)

The layer can't fail because it opens nothing: TransactorZIO opens a new JDBC connection for each TransactorZIO#connect or TransactorZIO#transact call and closes it afterwards. A wrong URL shows up at the first query, not at startup.

warning

Don't use jdbc:sqlite::memory: here. Each new connection to :memory: gets its own empty database, so the table we create in one call is gone in the next. Use a file, as above.

We create the table and put the books on the shelf. TransactorZIO#connect runs its synchronous body on ZIO's blocking thread pool and returns a Task:

runDemo {
  ZIO.serviceWithZIO[TransactorZIO] { transactor =>
    transactor.connect {
      createBookTable()
      shelf.foreach(books.insert)
      books.count
    }
  }.provide(transactorLayer)
}
// res2: Long = 3L

In production, a PostgreSQL database is usually reached through a connection pool. A pool is a javax.sql.DataSource that ZIO opens at startup and closes at shutdown, which is what ZLayer.scoped with ZIO.acquireRelease expresses. TransactorZIO.fromDataSource then builds the transactor on the pool. The pool's settings are the ones from How to Connect to PostgreSQL with a Connection Pool:

import com.zaxxer.hikari.{HikariConfig, HikariDataSource}
import javax.sql.DataSource

def poolConfig(): HikariConfig = {
  val config = new HikariConfig()
  config.setJdbcUrl(sys.env.getOrElse("BOOKSTORE_DB_URL", "jdbc:postgresql://localhost:5432/bookstore"))
  config.setUsername(sys.env.getOrElse("BOOKSTORE_DB_USER", "bookstore"))
  config.setPassword(sys.env("BOOKSTORE_DB_PASSWORD"))
  config.setMaximumPoolSize(10)
  config
}

val poolLayer: ZLayer[Any, Throwable, DataSource] =
  ZLayer.scoped {
    ZIO.acquireRelease(ZIO.attemptBlocking(new HikariDataSource(poolConfig())))(pool =>
      ZIO.attemptBlocking(pool.close()).orDie
    )
  }

val postgresTransactorLayer: ZLayer[DataSource, Nothing, TransactorZIO] =
  ZLayer.fromFunction((pool: DataSource) => TransactorZIO.fromDataSource(pool, SqlDialect.PostgreSQL))

val databaseLayer: ZLayer[Any, Throwable, TransactorZIO] =
  poolLayer >>> postgresTransactorLayer

HikariDataSource connects in its constructor, so databaseLayer fails at startup when the database is unreachable, and the pool closes when the application stops. The rest of this guide only depends on TransactorZIO, so swapping transactorLayer for databaseLayer is the only change it needs to run on PostgreSQL.

sql-zio also adds JdbcTransactor.postgresLayer and JdbcTransactor.sqliteLayer, which turn a DataSource layer into a layer of the synchronous Transactor. They are extension methods defined in zio.blocks.sql.zio, so they need import zio.blocks.sql.zio.*. A Transactor runs its body on the calling thread, so you would have to wrap each call in ZIO.attemptBlocking yourself; prefer TransactorZIO in ZIO code.

Write a ZIO Service on the Transactor​

The rest of the application shouldn't see SQL. We hide it behind a BookRepo service whose methods return ZIO effects:

trait BookRepo {
  def all: Task[List[Book]]
  def find(id: Long): Task[Option[Book]]
  def add(book: Book): Task[Unit]
}

final case class BookRepoLive(transactor: TransactorZIO) extends BookRepo {
  def all: Task[List[Book]] =
    transactor.connect(books.all)

  def find(id: Long): Task[Option[Book]] =
    transactor.connect(books.find(id).toOption)

  def add(book: Book): Task[Unit] =
    transactor.transact(books.insert(book)).unit
}

object BookRepoLive {
  val layer: ZLayer[TransactorZIO, Nothing, BookRepo] =
    ZLayer.fromFunction(BookRepoLive(_))
}

BookRepoLive takes the transactor as a constructor parameter, and ZLayer.fromFunction turns that constructor into a layer that needs a TransactorZIO. Each method is one TransactorZIO#connect or TransactorZIO#transact call, so each one runs on the blocking pool and fails with the JDBC exception as a Task error. TransactorZIO#transact turns off auto-commit, commits when its body returns and rolls back when it throws.

Callers ask ZIO for the service and provide both layers:

runDemo {
  ZIO.serviceWithZIO[BookRepo] { repo =>
    repo.add(Book(4L, "Scala with Cats", 2020, 2)) *> repo.all
  }.provide(BookRepoLive.layer, transactorLayer)
}.foreach(println)
// Book(1,Zionomicon,2024,5)
// Book(2,Programming in Scala,2008,3)
// Book(3,Functional Programming in Scala,2014,4)
// Book(4,Scala with Cats,2020,2)

Mix Database Writes with Other Effects in One Transaction​

TransactorZIO#transact is enough while the whole transaction is SQL. A sale is not: it lowers the stock and charges the customer, and the charge is a ZIO service. If the charge fails, the stock change must be undone. That calls for TransactorZIO#transactZIO, whose body is a ZIO effect: it opens a connection, turns off auto-commit, runs the effect, then commits if the effect succeeds and rolls back if it fails.

We model the payment provider as a ZIO service that declines one customer:

final case class PaymentDeclined(customer: String) extends Exception(s"Payment declined for $customer")
final case class OutOfStock(bookId: Long)          extends Exception(s"Book $bookId is out of stock")

trait Payments {
  def charge(customer: String, copies: Int): IO[PaymentDeclined, Unit]
}

val paymentsLayer: ZLayer[Any, Nothing, Payments] =
  ZLayer.succeed(new Payments {
    def charge(customer: String, copies: Int): IO[PaymentDeclined, Unit] =
      if (customer == "mallory") ZIO.fail(PaymentDeclined(customer)) else ZIO.unit
  })

The sale lowers the stock, fails with OutOfStock if there weren't enough copies, then charges the customer, all inside one TransactorZIO#transactZIO:

def sell(bookId: Long, copies: Int, customer: String): ZIO[TransactorZIO & Payments, Throwable, Unit] =
  ZIO.serviceWithZIO[TransactorZIO] { transactor =>
    transactor.transactZIO {
      for {
        updated <- ZIO.attemptBlocking(
                     sql"UPDATE book SET stock = stock - $copies WHERE id = $bookId AND stock >= $copies".update
                   )
        _       <- ZIO.fail(OutOfStock(bookId)).when(updated == 0)
        _       <- ZIO.serviceWithZIO[Payments](_.charge(customer, copies))
      } yield ()
    }
  }

def stockOf(bookId: Long): ZIO[TransactorZIO, Throwable, Int] =
  ZIO.serviceWithZIO[TransactorZIO](
    _.connect(sql"SELECT stock FROM book WHERE id = $bookId".queryOne[Int].getOrElse(0))
  )

Alice buys two copies of Zionomicon and Mallory tries to buy one:

runDemo {
  (for {
    before  <- stockOf(1L)
    alice   <- sell(1L, 2, "alice").either
    mallory <- sell(1L, 1, "mallory").either
    after   <- stockOf(1L)
  } yield {
    println(s"Stock before: $before")
    println(s"Alice: $alice")
    println(s"Mallory: ${mallory.left.map(_.getMessage)}")
    println(s"Stock after: $after")
  }).provide(transactorLayer, paymentsLayer)
}
// Stock before: 5
// Alice: Right(())
// Mallory: Left(Payment declined for mallory)
// Stock after: 3

Alice's sale committed and took two copies. Mallory's UPDATE ran, but the declined charge failed the body, so TransactorZIO#transactZIO rolled the UPDATE back and the stock only dropped by Alice's two copies. The error type is Throwable here because both of our errors extend Exception; in general TransactorZIO#transactZIO returns ZIO[R, E | Throwable, A], adding Throwable for the JDBC errors to whatever error type the body has.

Inside TransactorZIO#transactZIO, wrap every statement in ZIO.attemptBlocking. TransactorZIO#transactZIO runs only the connection setup, the commit and the rollback on the blocking pool, and the body's effect runs on whatever thread the fiber is on. ZIO.attemptBlocking moves the statement to the blocking pool and turns its exception into a typed failure that triggers the rollback.

Two ways TransactorZIO#transactZIO 0.0.56 commits instead of rolling back

A statement written directly in the body, such as transactZIO { sql"...".update; ZIO.unit }, runs while the body's effect is being built, before TransactorZIO#transactZIO watches it for failures. If such a statement throws, TransactorZIO#transactZIO does not roll back, and the statements before it in the same body are committed. Wrapping each statement in ZIO.attemptBlocking, as above, avoids this.

Interrupting a running TransactorZIO#transactZIO from outside, for example with ZIO#timeout, ZIO#race or by interrupting its fiber, also commits the statements that already ran. Don't put ZIO#timeout around a TransactorZIO#transactZIO; put a time limit on the slow effect inside the body with ZIO#timeoutFail, which fails the body and so rolls the transaction back, as shown in Limit How Long an Effect Waits.

TransactorZIO#transactZIO keeps the driver's default isolation level. When a transaction needs a specific one, pass it with the two-argument overload, transactor.transactZIO(TransactionIsolation.Serializable, readOnly = false) { ... }. SQL Transactions: Isolation, Read-Only and Savepoints explains the levels.

Operations that belong to one transaction must also share its connection: every TransactorZIO#connect, TransactorZIO#transact, TransactorZIO#connectZIO and TransactorZIO#transactZIO call takes its own connection. Calling one from inside another, for example a service method that calls BookRepo#add while it is inside its own TransactorZIO#transactZIO, opens a second connection while the first one still holds its locks. On SQLite, the inner write waits for the outer transaction and fails with SQLITE_BUSY after the busy timeout; on PostgreSQL, enough concurrent callers doing this use up the pool and wait for each other.

Instead, write the operations that should share a transaction as functions with a (using DbCon) parameter, like createBookTable above, and call them inside one TransactorZIO#transactZIO:

def restock(bookId: Long, copies: Int)(using DbCon): Int =
  sql"UPDATE book SET stock = stock + $copies WHERE id = $bookId".update

def restockAll(copies: Int): ZIO[TransactorZIO, Throwable, Int] =
  ZIO.serviceWithZIO[TransactorZIO](_.transactZIO {
    ZIO.foreach(List(1L, 2L, 3L))(bookId => ZIO.attemptBlocking(restock(bookId, copies))).map(_.sum)
  })

Both restock calls use the DbTx that TransactorZIO#transactZIO passes to its body, so they run on one connection and commit together.

Handle Errors, Retries and Timeouts the ZIO Way​

Every TransactorZIO method returns an ordinary ZIO effect, so ZIO's error operators work on database calls like on any other effect. This section shows the three you need most.

Turn Database Errors into Domain Errors​

A JDBC failure arrives as a java.sql.SQLException in the error channel. Callers usually care about a few specific cases, such as a duplicate key, and should see them as their own error type. With SQLite, SQLiteException#getResultCode tells the cases apart; on PostgreSQL, check SQLException#getSQLState, which is "23505" for a unique violation:

import org.sqlite.{SQLiteErrorCode, SQLiteException}

final case class DuplicateBook(id: Long) extends Exception(s"Book $id already exists")

def addNew(book: Book): ZIO[BookRepo, Throwable, Unit] =
  ZIO.serviceWithZIO[BookRepo](_.add(book)).catchSome {
    case e: SQLiteException if e.getResultCode == SQLiteErrorCode.SQLITE_CONSTRAINT_PRIMARYKEY =>
      ZIO.fail(DuplicateBook(book.id))
  }

Adding Zionomicon a second time fails with the domain error, not the driver's exception:

runDemo {
  addNew(Book(1L, "Zionomicon", 2024, 1)).either.provide(BookRepoLive.layer, transactorLayer)
}
// res6: Either[Throwable, Unit] = Left(DuplicateBook(1L))

Retry Transient Failures​

Some failures go away on their own: the database is restarting, the network dropped a connection, or another writer held a lock. ZIO#retry reruns the effect with a Schedule. Retrying a whole TransactorZIO#connect or TransactorZIO#transact is safe, because each attempt takes a new connection and a failed TransactorZIO#transact has already rolled back.

To see a retry without a real outage, we build a TransactorZIO whose connection factory fails twice before it connects, the way a database that is still starting up would:

val attempts = new java.util.concurrent.atomic.AtomicInteger(0)

val startingUp = new TransactorZIO(
  () =>
    if (attempts.incrementAndGet() <= 2) throw new java.sql.SQLException("Connection refused")
    else java.sql.DriverManager.getConnection(s"jdbc:sqlite:$dbFile"),
  SqlDialect.SQLite
)

We retry up to five times, waiting 10 milliseconds before the first retry and twice as long before each next one:

runDemo {
  startingUp
    .connect(books.count)
    .retry(Schedule.exponential(10.millis) && Schedule.recurs(5))
    .map(count => println(s"Counted $count books after ${attempts.get} attempts"))
}
// Counted 4 books after 3 attempts

Retry only what is safe to repeat. A TransactorZIO#transactZIO body that calls a payment service would charge the customer again on every retry; retry the database part inside the body instead, or make the charge idempotent.

Limit How Long an Effect Waits​

ZIO#timeout interrupts an effect that takes too long, but interruption can't stop a JDBC call that is already running. TransactorZIO runs statements with ZIO.attemptBlocking, which doesn't interrupt its thread, so the fiber waits until the statement finishes and only then reports the timeout. We time a query that counts to three million inside SQLite, with a timeout of 10 milliseconds:

val slowCount =
  sql"WITH RECURSIVE n(x) AS (SELECT 1 UNION ALL SELECT x + 1 FROM n WHERE x < 3000000) SELECT count(*) FROM n"

def timed[A](effect: Task[A]): Task[(A, Duration)] =
  effect.timed.map(_.swap)

The timeout reports None, but only after the whole query ran:

runDemo {
  ZIO.serviceWithZIO[TransactorZIO] { transactor =>
    timed(transactor.connect(slowCount.queryOne[Long]).timeout(10.millis)).map { case (result, elapsed) =>
      println(s"Result: $result")
      println(s"Waited longer than the timeout: ${elapsed.toMillis > 100}")
    }
  }.provide(transactorLayer)
}
// Result: None
// Waited longer than the timeout: true

When the caller must get its answer on time, add ZIO#disconnect before the timeout. The fiber then stops waiting at the deadline, while the statement finishes in the background and still holds its connection until it does:

runDemo {
  ZIO.serviceWithZIO[TransactorZIO] { transactor =>
    timed(transactor.connect(slowCount.queryOne[Long]).disconnect.timeout(10.millis)).map { case (result, elapsed) =>
      println(s"Result: $result")
      println(s"Returned before the query finished: ${elapsed.toMillis < 100}")
    }
  }.provide(transactorLayer)
}
// Result: None
// Returned before the query finished: true

To stop the statement itself, set a limit in the database, such as PostgreSQL's statement_timeout on the pool's connections.

Inside a TransactorZIO#transactZIO, put the time limit on the non-database effect with ZIO#timeoutFail. The timeout becomes a typed failure of the body, and the transaction rolls back. Here the payment provider hangs, so the sale fails after 50 milliseconds and the stock stays as it was:

val hangingPayments: ZLayer[Any, Nothing, Payments] =
  ZLayer.succeed(new Payments {
    def charge(customer: String, copies: Int): IO[PaymentDeclined, Unit] = ZIO.never
  })

def sellWithin(bookId: Long, copies: Int, customer: String): ZIO[TransactorZIO & Payments, Throwable, Unit] =
  ZIO.serviceWithZIO[TransactorZIO] { transactor =>
    transactor.transactZIO {
      for {
        _ <- ZIO.attemptBlocking(sql"UPDATE book SET stock = stock - $copies WHERE id = $bookId".update)
        _ <- ZIO
               .serviceWithZIO[Payments](_.charge(customer, copies))
               .timeoutFail(new java.util.concurrent.TimeoutException("Payment took too long"))(50.millis)
      } yield ()
    }
  }

Bob's payment never answers:

runDemo {
  (for {
    before <- stockOf(2L)
    result <- sellWithin(2L, 1, "bob").either
    after  <- stockOf(2L)
  } yield {
    println(s"Sale: $result")
    println(s"Stock before: $before, after: $after")
  }).provide(transactorLayer, hangingPayments)
}
// Sale: Left(java.util.concurrent.TimeoutException: Payment took too long)
// Stock before: 3, after: 3

Test Services Against a Fresh Database​

A ZIO Test spec gets its services from layers, so a test database is a layer that creates a temporary SQLite file, builds the schema and seeds the books, and deletes the file when the test ends. ZLayer.scoped with ZIO.acquireRelease ties the file's lifetime to the layer:

import java.nio.file.{Files, Path}

val testTransactorLayer: ZLayer[Any, Throwable, TransactorZIO] =
  ZLayer.scoped {
    for {
      file      <- ZIO.acquireRelease(ZIO.attemptBlocking(Files.createTempFile("bookstore-test", ".db")))(file =>
                     ZIO.attemptBlocking(Files.deleteIfExists(file)).orDie
                   )
      transactor = TransactorZIO.fromUrl(s"jdbc:sqlite:$file", SqlDialect.SQLite)
      _         <- transactor.connect {
                     createBookTable()
                     shelf.foreach(books.insert)
                   }
    } yield transactor
  }

Each time ZIO builds the layer, it gets a new file. Two "tests" that each add a book both start from the same three books:

List("first test", "second test").foreach { name =>
  val size = runDemo {
    ZIO.serviceWithZIO[BookRepo] { repo =>
      repo.add(Book(10L, "Scala with Cats", 2020, 2)) *> repo.all.map(_.size)
    }.provide(BookRepoLive.layer, testTransactorLayer)
  }
  println(s"$name sees $size books")
}
// first test sees 4 books
// second test sees 4 books

In a spec, Spec#provide builds the layer once for each test, so every test gets its own database, and Spec#provideShared builds it once for the whole suite. The companion spec below uses Spec#provide, so its tests can run in any order and in parallel:

BookRepoSpec.scala
sql-examples/src/test/scala/sqlfromzio/BookRepoSpec.scala
package sqlfromzio

import zio.*
import zio.test.*
import zio.blocks.schema.Schema
import zio.blocks.sql.zio.*
import zio.blocks.sql.*
import java.nio.file.Files

object BookRepoSpec extends ZIOSpecDefault {

  val testTransactorLayer: ZLayer[Any, Throwable, TransactorZIO] =
    ZLayer.scoped {
      for {
        file      <- ZIO.acquireRelease(ZIO.attemptBlocking(Files.createTempFile("bookstore-test", ".db")))(file =>
                       ZIO.attemptBlocking(Files.deleteIfExists(file)).orDie
                     )
        transactor = TransactorZIO.fromUrl(s"jdbc:sqlite:$file", SqlDialect.SQLite)
        _         <- transactor.connect {
                       createBookTable()
                       shelf.foreach(books.insert)
                     }
      } yield transactor
    }

  def spec: Spec[Any, Throwable] =
    suite("BookRepoSpec")(
      test("starts with the three seeded books") {
        for {
          repo <- ZIO.service[BookRepo]
          all  <- repo.all
        } yield assertTrue(all == shelf)
      },
      test("first test adds a book and sees four books") {
        for {
          repo <- ZIO.service[BookRepo]
          _    <- repo.add(Book(10L, "Scala with Cats", 2020, 2))
          all  <- repo.all
        } yield assertTrue(all.size == 4)
      },
      test("second test adds the same ID and still sees four books") {
        for {
          repo <- ZIO.service[BookRepo]
          _    <- repo.add(Book(10L, "Scala with Cats", 2020, 2))
          all  <- repo.all
        } yield assertTrue(all.size == 4)
      },
      test("find returns a seeded book") {
        for {
          repo  <- ZIO.service[BookRepo]
          found <- repo.find(1L)
        } yield assertTrue(found.map(_.title) == Some("Zionomicon"))
      }
    ).provide(BookRepoLive.layer, testTransactorLayer)
}

How to Test Database Code Against SQLite covers the same ideas for the synchronous API, including fixtures and rollback-only tests.

Read a Large Result Set​

Frag#queryStream and Frag#queryChunked read a result set in chunks as a ZIO Blocks Stream, without loading every row into memory. sql-zio has no bridge to ZStream, and the stream must be consumed while its connection is open. So we run it inside TransactorZIO#connect and reduce it there. A terminal operation such as Stream#runFold returns an Either, and ZIO#absolve turns its Left into a failure:

runDemo {
  ZIO.serviceWithZIO[TransactorZIO] { transactor =>
    transactor.connect {
      sql"SELECT id, title, year, stock FROM book"
        .queryChunked[Book](2)
        .runFold(0)((total, chunk) => total + chunk.foldLeft(0)(_ + _.stock))
    }.absolve
  }.provide(transactorLayer)
}
// res12: Int = 12

The fold reads two rows at a time and adds up the stock across the shelf. The whole fold runs inside TransactorZIO#connect, so it runs on the blocking pool. When each row needs a ZIO effect, such as publishing it to a queue, collect the rows you need inside TransactorZIO#connect and run the effects after it returns, or read the table in pages with Repo#pageAfter.

Putting It Together​

The complete example below builds the bookstore application with ZIOAppDefault. It provides the transactor, the BookRepo service and the payment service as layers, sells books in TransactorZIO#transactZIO transactions, and prints the stock after a declined payment:

CompleteExample.scala
sql-examples/src/main/scala/sqlfromzio/CompleteExample.scala
package sqlfromzio

import zio.*
import zio.blocks.schema.Schema
import zio.blocks.sql.zio.*
import zio.blocks.sql.*

/**
 * How to Use the sql Module from a ZIO Application — Complete Example
 *
 * A bookstore application: layers for the transactor, the `BookRepo` service
 * and the payment service, sales in `transactZIO` transactions, and the stock
 * after a declined payment.
 *
 * Run with: cd sql-examples && sbt "runMain sqlfromzio.CompleteExample"
 */
object CompleteExample extends ZIOAppDefault {

  val dbFile = java.nio.file.Files.createTempFile("bookstore", ".db")
  dbFile.toFile.deleteOnExit()

  val transactorLayer: ZLayer[Any, Nothing, TransactorZIO] =
    TransactorZIO.layer(s"jdbc:sqlite:$dbFile", SqlDialect.SQLite)

  def sell(bookId: Long, copies: Int, customer: String): ZIO[TransactorZIO & Payments, Throwable, Unit] =
    ZIO.serviceWithZIO[TransactorZIO] { transactor =>
      transactor.transactZIO {
        for {
          updated <- ZIO.attemptBlocking(
                       sql"UPDATE book SET stock = stock - $copies WHERE id = $bookId AND stock >= $copies".update
                     )
          _ <- ZIO.fail(OutOfStock(bookId)).when(updated == 0)
          _ <- ZIO.serviceWithZIO[Payments](_.charge(customer, copies))
        } yield ()
      }
    }

  val program: ZIO[TransactorZIO & BookRepo & Payments, Throwable, Unit] =
    for {
      _ <- ZIO.serviceWithZIO[TransactorZIO](_.connect {
             createBookTable()
             shelf.foreach(books.insert)
           })
      _       <- ZIO.serviceWithZIO[BookRepo](_.add(Book(4L, "Scala with Cats", 2020, 2)))
      _       <- sell(1L, 2, "alice")
      mallory <- sell(1L, 1, "mallory").either
      _       <- Console.printLine(s"Mallory: $mallory")
      all     <- ZIO.serviceWithZIO[BookRepo](_.all)
      _       <- ZIO.foreachDiscard(all)(book => Console.printLine(s"${book.title}: ${book.stock} in stock"))
    } yield ()

  val run =
    program.provide(transactorLayer, BookRepoLive.layer, paymentsLayer)
}

Running the Examples​

The companion examples live in the sql-examples build, which uses the published ZIO Blocks artifacts. Clone the repository and change into it:

git clone https://github.com/zio/zio-blocks.git
cd zio-blocks/sql-examples

Every example uses a temporary SQLite file and deletes it when it exits, so you can run them in any order.

The examples share the bookstore model from The Bookstore Model, defined once in Bookstore.scala:

Bookstore.scala: the shared Book model, books, shelf, createBookTable and stockOf
sql-examples/src/main/scala/sqlfromzio/Bookstore.scala
package sqlfromzio

import zio.*
import zio.blocks.schema.Schema
import zio.blocks.sql.zio.*
import zio.blocks.sql.*

/**
 * How to Use the sql Module from a ZIO Application — the shared bookstore model
 *
 * The `Book` model, its `Repo`, the books on the shelf, the hand-written
 * `CREATE TABLE` statement and a stock lookup. The examples in this package
 * import them.
 */
case class Book(id: Long, title: String, year: Int, stock: Int)

object Book {
  given Schema[Book] = Schema.derived
}

val books: Repo[Book, Long] = Repo.derived[Book, Long]

val shelf = List(
  Book(1L, "Zionomicon", 2024, 5),
  Book(2L, "Programming in Scala", 2008, 3),
  Book(3L, "Functional Programming in Scala", 2014, 4)
)

def createBookTable()(using DbCon): Unit =
  sql"""CREATE TABLE book (
    id INTEGER PRIMARY KEY,
    title TEXT NOT NULL,
    year INTEGER NOT NULL,
    stock INTEGER NOT NULL
  )""".update

def stockOf(bookId: Long): ZIO[TransactorZIO, Throwable, Int] =
  ZIO.serviceWithZIO[TransactorZIO](
    _.connect(sql"SELECT stock FROM book WHERE id = $bookId".queryOne[Int].getOrElse(0))
  )

To check that every example compiles without running any of them:

sbt compile
BookRepoLayerExample.scala: provide the transactor and the BookRepo service as layers
sql-examples/src/main/scala/sqlfromzio/BookRepoLayerExample.scala
package sqlfromzio

import zio.*
import zio.blocks.schema.Schema
import zio.blocks.sql.zio.*
import zio.blocks.sql.*

trait BookRepo {
  def all: Task[List[Book]]
  def find(id: Long): Task[Option[Book]]
  def add(book: Book): Task[Unit]
}

final case class BookRepoLive(transactor: TransactorZIO) extends BookRepo {
  def all: Task[List[Book]] =
    transactor.connect(books.all)

  def find(id: Long): Task[Option[Book]] =
    transactor.connect(books.find(id).toOption)

  def add(book: Book): Task[Unit] =
    transactor.transact(books.insert(book)).unit
}

object BookRepoLive {
  val layer: ZLayer[TransactorZIO, Nothing, BookRepo] =
    ZLayer.fromFunction(BookRepoLive(_))
}

/**
 * How to Use the sql Module from a ZIO Application — Provide the Database as a
 * ZLayer
 *
 * Provides a `TransactorZIO` over a temporary SQLite file as a layer, creates
 * the table, and reads the books through a `BookRepo` service.
 *
 * Run with: cd sql-examples && sbt "runMain sqlfromzio.BookRepoLayerExample"
 */
object BookRepoLayerExample extends ZIOAppDefault {

  val dbFile = java.nio.file.Files.createTempFile("bookstore", ".db")
  dbFile.toFile.deleteOnExit()

  val transactorLayer: ZLayer[Any, Nothing, TransactorZIO] =
    TransactorZIO.layer(s"jdbc:sqlite:$dbFile", SqlDialect.SQLite)

  val run =
    for {
      count <- ZIO
                 .serviceWithZIO[TransactorZIO] { transactor =>
                   transactor.connect {
                     createBookTable()
                     shelf.foreach(books.insert)
                     books.count
                   }
                 }
                 .provide(transactorLayer)
      _   <- ZIO.succeed(println(s"Seeded $count books"))
      all <- ZIO
               .serviceWithZIO[BookRepo] { repo =>
                 repo.add(Book(4L, "Scala with Cats", 2020, 2)) *> repo.all
               }
               .provide(BookRepoLive.layer, transactorLayer)
      _ <- ZIO.succeed(all.foreach(println))
    } yield ()
}

Run it with:

sbt "runMain sqlfromzio.BookRepoLayerExample"
TransactZIOExample.scala: a sale that charges the customer inside the transaction, and the rollback on a declined payment
sql-examples/src/main/scala/sqlfromzio/TransactZIOExample.scala
package sqlfromzio

import zio.*
import zio.blocks.schema.Schema
import zio.blocks.sql.zio.*
import zio.blocks.sql.*

final case class PaymentDeclined(customer: String) extends Exception(s"Payment declined for $customer")
final case class OutOfStock(bookId: Long)          extends Exception(s"Book $bookId is out of stock")

trait Payments {
  def charge(customer: String, copies: Int): IO[PaymentDeclined, Unit]
}

val paymentsLayer: ZLayer[Any, Nothing, Payments] =
  ZLayer.succeed(new Payments {
    def charge(customer: String, copies: Int): IO[PaymentDeclined, Unit] =
      if (customer == "mallory") ZIO.fail(PaymentDeclined(customer)) else ZIO.unit
  })

/**
 * How to Use the sql Module from a ZIO Application — Mix Database Writes with
 * Other Effects in One Transaction
 *
 * A sale lowers the stock and charges the customer inside one `transactZIO`. A
 * declined payment rolls the stock change back.
 *
 * Run with: cd sql-examples && sbt "runMain sqlfromzio.TransactZIOExample"
 */
object TransactZIOExample extends ZIOAppDefault {

  val dbFile = java.nio.file.Files.createTempFile("bookstore", ".db")
  dbFile.toFile.deleteOnExit()

  val transactorLayer: ZLayer[Any, Nothing, TransactorZIO] =
    TransactorZIO.layer(s"jdbc:sqlite:$dbFile", SqlDialect.SQLite)

  def sell(bookId: Long, copies: Int, customer: String): ZIO[TransactorZIO & Payments, Throwable, Unit] =
    ZIO.serviceWithZIO[TransactorZIO] { transactor =>
      transactor.transactZIO {
        for {
          updated <- ZIO.attemptBlocking(
                       sql"UPDATE book SET stock = stock - $copies WHERE id = $bookId AND stock >= $copies".update
                     )
          _ <- ZIO.fail(OutOfStock(bookId)).when(updated == 0)
          _ <- ZIO.serviceWithZIO[Payments](_.charge(customer, copies))
        } yield ()
      }
    }

  val run =
    (for {
      _ <- ZIO.serviceWithZIO[TransactorZIO](_.connect {
             createBookTable()
             shelf.foreach(books.insert)
           })
      before  <- stockOf(1L)
      alice   <- sell(1L, 2, "alice").either
      mallory <- sell(1L, 1, "mallory").either
      after   <- stockOf(1L)
    } yield {
      println(s"Stock before: $before")
      println(s"Alice: $alice")
      println(s"Mallory: $mallory")
      println(s"Stock after: $after")
    }).provide(transactorLayer, paymentsLayer)
}

Run it with:

sbt "runMain sqlfromzio.TransactZIOExample"
ErrorsRetriesTimeoutsExample.scala: domain errors, retries and timeouts
sql-examples/src/main/scala/sqlfromzio/ErrorsRetriesTimeoutsExample.scala
package sqlfromzio

import zio.*
import zio.blocks.schema.Schema
import zio.blocks.sql.zio.*
import zio.blocks.sql.*
import org.sqlite.{SQLiteErrorCode, SQLiteException}

final case class DuplicateBook(id: Long) extends Exception(s"Book $id already exists")

/**
 * How to Use the sql Module from a ZIO Application — Handle Errors, Retries and
 * Timeouts the ZIO Way
 *
 * Maps a primary key violation to a domain error, retries a connection that
 * fails twice, compares `timeout` with `disconnect.timeout` on a slow query,
 * and uses `timeoutFail` inside `transactZIO` so a hanging payment rolls back.
 *
 * Run with: cd sql-examples && sbt "runMain
 * sqlfromzio.ErrorsRetriesTimeoutsExample"
 */
object ErrorsRetriesTimeoutsExample extends ZIOAppDefault {

  val dbFile = java.nio.file.Files.createTempFile("bookstore", ".db")
  dbFile.toFile.deleteOnExit()

  val transactorLayer: ZLayer[Any, Nothing, TransactorZIO] =
    TransactorZIO.layer(s"jdbc:sqlite:$dbFile", SqlDialect.SQLite)

  def addNew(book: Book): ZIO[BookRepo, Throwable, Unit] =
    ZIO.serviceWithZIO[BookRepo](_.add(book)).catchSome {
      case e: SQLiteException if e.getResultCode == SQLiteErrorCode.SQLITE_CONSTRAINT_PRIMARYKEY =>
        ZIO.fail(DuplicateBook(book.id))
    }

  val attempts = new java.util.concurrent.atomic.AtomicInteger(0)

  val startingUp = new TransactorZIO(
    () =>
      if (attempts.incrementAndGet() <= 2) throw new java.sql.SQLException("Connection refused")
      else java.sql.DriverManager.getConnection(s"jdbc:sqlite:$dbFile"),
    SqlDialect.SQLite
  )

  val slowCount =
    sql"WITH RECURSIVE n(x) AS (SELECT 1 UNION ALL SELECT x + 1 FROM n WHERE x < 3000000) SELECT count(*) FROM n"

  def timed[A](effect: Task[A]): Task[(A, Duration)] =
    effect.timed.map(_.swap)

  val hangingPayments: ZLayer[Any, Nothing, Payments] =
    ZLayer.succeed(new Payments {
      def charge(customer: String, copies: Int): IO[PaymentDeclined, Unit] = ZIO.never
    })

  def sellWithin(bookId: Long, copies: Int, customer: String): ZIO[TransactorZIO & Payments, Throwable, Unit] =
    ZIO.serviceWithZIO[TransactorZIO] { transactor =>
      transactor.transactZIO {
        for {
          _ <- ZIO.attemptBlocking(sql"UPDATE book SET stock = stock - $copies WHERE id = $bookId".update)
          _ <- ZIO
                 .serviceWithZIO[Payments](_.charge(customer, copies))
                 .timeoutFail(new java.util.concurrent.TimeoutException("Payment took too long"))(50.millis)
        } yield ()
      }
    }

  val run =
    for {
      _ <- ZIO
             .serviceWithZIO[TransactorZIO](_.connect {
               createBookTable()
               shelf.foreach(books.insert)
             })
             .provide(transactorLayer)
      duplicate <- addNew(Book(1L, "Zionomicon", 2024, 1)).either.provide(BookRepoLive.layer, transactorLayer)
      _         <- ZIO.succeed(println(s"Adding Zionomicon again: $duplicate"))
      _         <- startingUp
             .connect(books.count)
             .retry(Schedule.exponential(10.millis) && Schedule.recurs(5))
             .map(count => println(s"Counted $count books after ${attempts.get} attempts"))
      _ <- ZIO
             .serviceWithZIO[TransactorZIO] { transactor =>
               timed(transactor.connect(slowCount.queryOne[Long]).timeout(10.millis)).map { case (result, elapsed) =>
                 println(s"timeout: result $result")
                 println(s"Waited longer than the timeout: ${elapsed.toMillis > 100}")
               }
             }
             .provide(transactorLayer)
      _ <- ZIO
             .serviceWithZIO[TransactorZIO] { transactor =>
               timed(transactor.connect(slowCount.queryOne[Long]).disconnect.timeout(10.millis)).map {
                 case (result, elapsed) =>
                   println(s"disconnect.timeout: result $result")
                   println(s"Returned before the query finished: ${elapsed.toMillis < 100}")
               }
             }
             .provide(transactorLayer)
      before <- stockOf(2L).provide(transactorLayer)
      result <- sellWithin(2L, 1, "bob").either.provide(transactorLayer, hangingPayments)
      after  <- stockOf(2L).provide(transactorLayer)
      _      <- ZIO.succeed {
             println(s"Sale: $result")
             println(s"Stock before: $before, after: $after")
           }
    } yield ()
}

Run it with:

sbt "runMain sqlfromzio.ErrorsRetriesTimeoutsExample"
LargeResultExample.scala: fold a chunked query inside connect
sql-examples/src/main/scala/sqlfromzio/LargeResultExample.scala
package sqlfromzio

import zio.*
import zio.blocks.schema.Schema
import zio.blocks.sql.zio.*
import zio.blocks.sql.*

/**
 * How to Use the sql Module from a ZIO Application — Read a Large Result Set
 *
 * Folds a chunked query inside `TransactorZIO#connect` and turns the `Either`
 * of the fold into a ZIO failure with `absolve`.
 *
 * Run with: cd sql-examples && sbt "runMain sqlfromzio.LargeResultExample"
 */
object LargeResultExample extends ZIOAppDefault {

  val dbFile = java.nio.file.Files.createTempFile("bookstore", ".db")
  dbFile.toFile.deleteOnExit()

  val transactorLayer: ZLayer[Any, Nothing, TransactorZIO] =
    TransactorZIO.layer(s"jdbc:sqlite:$dbFile", SqlDialect.SQLite)

  val run =
    (for {
      _ <- ZIO.serviceWithZIO[TransactorZIO](_.connect {
             createBookTable()
             shelf.foreach(books.insert)
           })
      total <- ZIO.serviceWithZIO[TransactorZIO] { transactor =>
                 transactor.connect {
                   sql"SELECT id, title, year, stock FROM book"
                     .queryChunked[Book](2)
                     .runFold(0)((total, chunk) => total + chunk.foldLeft(0)(_ + _.stock))
                 }.absolve
               }
      _ <- ZIO.succeed(println(s"Total stock on the shelf: $total"))
    } yield ()).provide(transactorLayer)
}

Run it with:

sbt "runMain sqlfromzio.LargeResultExample"

The complete example runs with:

sbt "runMain sqlfromzio.CompleteExample"

The ZIO Test spec runs with:

sbt "testOnly sqlfromzio.BookRepoSpec"

Going Further​