Introduction Last updated: 2022-08-06

Kuzminki is query builder and access library for PostgreSQL written in Scala.

Kuzminki is written for those who like SQL. Queries are written with the same logic you write SQL statements. As a result the code is easy to read and memorise while the resulting SQL statement is predictable.

Kuzminki is available for ZIO, ZIO 2 and Akka.

On Github for ZIO 1 https://github.com/karimagnusson/kuzminki-zio

On Github for ZIO 2 https://github.com/karimagnusson/kuzminki-zio-2

On Github for Akka https://github.com/karimagnusson/kuzminki-akka

Installation

For ZIO 1
libraryDependencies += "io.github.karimagnusson" % "kuzminki-zio" % "0.9.4-RC1"
For ZIO 2
libraryDependencies += "io.github.karimagnusson" % "kuzminki-zio-2" % "0.9.4-RC1"
For Akka
libraryDependencies += "io.github.karimagnusson" % "kuzminki-akka" % "0.9.4-RC1"

Example

This example is for ZIO. For ZIO 2 and Akka refer to the respective github pages.

								import zio._
import zio.console._
import zio.blocking._
import kuzminki.api._

object ExampleApp extends zio.App {

  class Client extends Model("client") {
    val id = column[Int]("id")
    val username = column[String]("username")
    val age = column[Int]("age")
    def all = (id, username, age)
  }

  val client = Model.get[Client]

  val job = for {
    _ <- sql
      .insert(client)
      .cols2(t => (t.username, t.age))
      .run(("Joe", 35))
    
    _ <- sql
      .update(client)
      .set(_.age ==> 24)
      .where(_.id === 4)
      .run
    
    _ <- sql.delete(client).where(_.id === 7).run
    
    clients <- sql
      .select(client)
      .cols3(_.all)
      .where(_.age > 25)
      .limit(5)
      .run
    
    _ <- ZIO.foreach(clients) {
      case (id, username, age) =>
        putStrLn(s"$id $username $age")
    }
  } yield ()

  val dbConfig = DbConfig.forDb("company")
  val dbLayer = Kuzminki.layer(dbConfig)

  override def run(args: List[String]): ZIO[ZEnv, Nothing, ExitCode] = {
    job.provideCustomLayer(dbLayer).exitCode
  }
}
							

Results

                // query
val stm = sql.select(client).cols3(_.all).where(_.age > 25).limit(5)
stm.run: List[T]
stm.runHead: T
stm.runHeadOpt: Option[T]

// modify row
implicit val toMyType: SomeRow => MyType = row => //...
stm.runAs[MyType]: List[MyType]
stm.runHeadAs[MyType]: MyType
stm.runHeadOptAs[MyType]: Option[MyType]

// operation
val stm = sql.update(client).set(_.age ==> 24).where(_.id === 4)
stm.run: Unit
stm.runNum: Int
              

Connecting to the database

Config

For Akka replace forDb("{DB-NAME}") with withDispatcher("{NAME}") default = "kuzminki-dispatcher"

								val dbConfig = DbConfig
  .forDb("{DB-NAME}")
  .withPoolSize(10) // default = 10
  .withHost("{HOST}") // default = localhost
  .withPort("{PORT}") // default = 5432
  .withUser("{USER}}")
  .withPassword("{PASSWORD}")
  .withOptions(Map(...))
            	

Layer

Create a layer to make the driver instance accessable under Has[Kuzminki]

								object MyApp extends zio.App {
	
  val job = // ...

  val kuzminkiLayer = Kuzminki.layer(dbConfig)

  override def run(args: List[String]): ZIO[ZEnv, Nothing, ExitCode] = {
    job.provideCustomLayer(kuzminkiLayer).exitCode
  }
}
            	

Custom connection

								for { 
  db <- Kuzminki.create(dbConfig)
  users <- db.query {
    sql.select(user).cols1(_.username.all.orderBy(_.id.asc)
  }
} yield users
            	

Split connection

If you wish to have two connection pools, one for SELECT and another for INSERT, UPDATE, DELETE, you can use layerSplit. To create an instance rather than a Layer, use createSplit.

								// for .query .queryHead .queryHeadOpt
val getConfig = DbConfig.forDb("company").withPoolSize(5)

// for .exec .execNum
val setConfig = DbConfig.forDb("company").withPoolSize(5)

val dbLayer = Kuzminki.layerSplit(getConfig, setConfig)
            	

Model

Creating a model

Column types are listed under Data types

								import kuzminki.api._
import java.sql.Timestamp

class User extends Model("user_profile") {
  val id = column[Int]("id")
  val username = column[String]("username")
  val email = column[String]("email")
  val name = column[String]("name")
  val age = column[Int]("age")
  val gender = column[String]("gender")
  val country = column[String]("country")
  val city = column[String]("city")
  val discount = column[Int]("discount")
  val isActive = column[Boolean]("is_active")
  val created = column[Timestamp]("created")
}

Model.register[User]
            	

Custom methods

You can create custom methods to access columns that you regularly use.

								class Customer extends Model("customer") {
  val id = column[Int]("id")
  val userId = column[Int]("user_id")
  val spending = column[Int]("spending")
  def all = (id, userId, spending)
}
            	

Create a model instance

Model.register[User] creates an instance of the model for later use and makes sure there is only one instance of the model. Model.register[User] gets an existing instance of the model. If it does not exist, it is created.

								Model.register[User]
// ...
val user = Model.get[User]
            	

Array fields

Array fields can be created for all the available types. Array fields can be filtered using has hasNot overlap overlapNot and updated using append prepend remove.

                class Nums extends Model("demo") {
  val id = column[Int]("id")
  val numbers = column[Seq[Int]]("numbers")
}

val nums = Model.get[Nums]

for {
  id <- sql
    .insert(nums)
    .cols1(_.numbers)
    .returning1(_.id)
    .run(List(1, 2, 3))

  _ <- sql
    .update(nums)
    .set(_.numbers.append(4)) // append, prepend, remove
    .where(_.id === id)
    .run

  numbers <- sql
    .select(nums)
    .cols1(_.numbers)
    .where(_.numbers.has(2))  // has, hasNot, overlap, overlapNot
    .run
} yield numbers // List[Vector[Int]]
              

Select

Select query

You select the columns as tuple of model columns. The query will return tuple of the column types. In this case Seq[Tuple2[Int, String]]. If you need more than 22 columns you can use colsSeq or colsType. To order by ASC rather than DESC use orderBy(_.age.asc).

								import kuzminki.api._

val user = Model.get[User]

sql
  .select(user)
  .cols2(t => (
    t.id,
    t.username
  ))
  .where(t => Seq(
    t.gender === "f",
    t.age > 25
  ))
  .orderBy(_.age.desc)
  .limit(10)
  .run
            	
								SELECT
  "id",
  "username"
FROM "user_profile"
WHERE "gender" = 'f'
AND "age" > 25
ORDER BY "age" DESC
LIMIT 10
            	

Row as case class

              case class ProductItem(id: Int, name: String, price: Float)

class Product extends Model("product") {
  val id = column[Int]("id")
  val name = column[String]("name")
  val price = column[Float]("price")
  val item = read[ProductItem](id, name, price)
}

val product = Model.get[Product]

sql
  .select(product)
  .colsType(_.item)
  .where(_.price < 100.0)
  .run
  // returns List[ProductItem]
              

Row as Seq

              sql
  .select(user)
  .colsSeq(t => List(
    t.id,
    t.username,
    t.email
  ))
  .where(_.age < 25)
  .run
  // returns List[Vector[Any]]
              

Row as JSON

You can get the result as JSON, using your favorite JSON library. It comes in handy if you are, for example, writing a REST web service.

                // See PlayJsonLoader below
implicit val loadJson: Seq[Tuple2[String, Any]] => JsValue = data => PlayJsonLoader.load(data)

sql
  .select(user)
  .colsNamed(t => List(
    t.id,        // Column name is used as a key.
    t.username,  // If you want a different key:
    t.email      // ("user_id"  -> t.id)
  ))
  .where(_.age < 25)
  .runAs[JsValue]
  // returns List[JsValue]
              

Write something along these lines to use the JSON library of your choosing.

                import play.api.libs.json._

object PlayJsonLoader {

  val toJsValue: Any => JsValue = {
    case v: String      => JsString(v)
    case v: Boolean     => JsBoolean(v)
    case v: Short       => JsNumber(v)
    case v: Int         => JsNumber(v)
    case v: Long        => JsNumber(v)
    case v: Float       => JsNumber(v)
    case v: Double      => JsNumber(v)
    case v: BigDecimal  => JsNumber(v)
    case v: Time        => Json.toJson(v)
    case v: Date        => Json.toJson(v)
    case v: Timestamp   => Json.toJson(v)
    case v: Option[_]   => v.map(toJsValue).getOrElse(JsNull)
    case v: Seq[_]      => JsArray(v.map(toJsValue))
    case _              => throw new Exception("Cannot convert to JsValue")
  }

  def load(data: Seq[Tuple2[String, Any]]): JsValue = {
    JsObject(data.map(p => (p._1, toJsValue(p._2))))
  }
}
              

Where

Refer to Operators for a list of operators.

								.where(_.id > 100)

.where(t => Seq(
  t.gender === "f",
  t.age > 25
))
            	

AND / OR

								import kuzminki.fn._

.where(t => Seq(
  t.age > 25,
  Or(
    t.country === "RU",
    t.country === "FR"
  )
))
// WHERE "age" > 25 AND ("country" == 'RU' OR "country" == 'FR')

.where(t => Or(
  And(
    t.country === "RU",
    t.city === "Moscow"
  ),
  And(
    t.country === "FR",
    t.city === "Paris"
  )
))
// WHERE ("country" == 'RU' AND "city" == 'Moscow') OR ("country" == 'FR' AND "city" == 'Paris')
            	

Optional conditions

Optional conditions for example from http GET request.

								.whereOpt(_.id > Some(100))

.whereOpts(t => Seq(
  t.gender === None,
  t.age > Some(25)
))
// WHERE "age" > 25

.whereOpts(t => Seq(
  t.age > Some(25),
  Or.opts(
    t.country === Some("RU"),
    t.country === Some("FR")
  )
))
// WHERE "age" > 25 AND ("country" == 'RU' OR "country" == 'FR')
            	

Null values

If you have null values in a column you can use one of the folloing ways to turn the result into Option[T]

Convert the column in the model from T to Option[T]

								val city = column[String]("city").asOpt
							

Convert the column in the query from T to Option[T]

								cols1(_.city.asOpt)
							

Use Posgres's coalesce function to return a default value in case of null

								import kuzminki.fn._
cols1(t => Fn.coalesce(t.city, "No city"))
							

Nested query

								class Newsletter extends Model("newsletter") {
  val email = column[String]("email")
  val isSubscribed = column[Boolean]("is_subscribed")
}

val newsletter = Model.get[Newsletter]

sql
  .select(user)
  .cols1(_.username)
  .where(_.email.in(
    sql
      .select(newsletter)
      .cols1(_.email)
      .where(_.isSubscribed === true)
  ))
  .run
            	
								SELECT "username"
FROM "user_profile"
WHERE "email" = ANY(
  SELECT "email"
  FROM "newsletter"
  WHERE "is_subscribed" = true
)
            	

Cache

For increased performance you can cache your queries. The SQL string will be created only once and you get the same performance as you get with raw queries.

								val stm = sql
  .select(user)
  .cols1(_.username)
  .all
  .orderBy(_.age.asc)
  .cacheWhere2(t => (
    t.country.cacheEq,
    t.age.cacheGt
  ))

stm.run(("CN", 25))
            	
								SELECT "username"
FROM "user_profile"
WHERE "country" = 'CN'
AND "age" > 25
ORDER BY "age" ASC
            	

Cached with WHERE

You can use normal WHERE conditions with cached queries.

								val stm = sql
  .select(user)
  .cols1(_.username)
  .where(_.age > 25)
  .orderBy(_.age.asc)
  .cacheWhere1(_.country.cacheEq)

stm.run("CN")
            	
								SELECT "username"
FROM "user_profile"
WHERE "age" > 25
AND "country" = 'CN'
ORDER BY "age" ASC
            	

Join

Select join

To do Joins you just put two model instances as arguments and the models will be accessable under a and b

							sql
  .select(user, customer)
  .cols3(t => (
    t.a.id,
    t.a.username,
    t.b.spending
  ))
  .joinOn(_.id, _.userId)
  .where(t => Seq(
    t.a.age > 25,
    t.b.spending > 1000
  ))
  .orderBy(_.b.spending.desc)
  .limit(10)
  .run
  // returns List[Tuple3[Int, String, Int]]
            	
								SELECT
  "a"."id",
  "a"."username",
  "b"."spending"
FROM "user_profile" "a"
INNER JOIN "customer" "b"
ON "a"."id" = "b"."user_id"
WHERE "a"."age" > 25
AND "b"."spending" > 1000
ORDER BY "b"."spending" DESC
LIMIT 10
            	

Join result types

You can have a join query return a case class.

              case class UserSpending(userId: Int, username: String, spending: Int)

class UserCustomer extends JoinRead[User, Customer] {
  val userSpending = read[UserSpending](a.id, a.username, b.spending)
}

val userCustomer = new UserCustomer
implicit val userCustomerConv = userCustomer.convert

sql
  .select(userCustomer)
  .colsType(_.userSpending)
  .joinOn(_.id, _.userId)
  .where(_.a.age > 25)
  .orderBy(_.b.spending.desc)
  .limit(10)
  .run
  // returns List[UserCustomer]
              

You can also define results as tuples or a Vector

                class UserCustomer extends JoinRead[User, Customer] {
  val userSpending = read[UserSpending](a.id, a.username, b.spending)
  // colsType(_.userSpending)

  val userSpendingTuple = (a.id, a.username, b.spending)
  // cols3(_.userSpendingTuple)

  val userSpendingVector = Vector(a.id, a.username, b.spending)
  // colsVector(_.userSpendingVector)
}
              
                SELECT
  "a"."id",
  "a"."username",
  "b"."spending"
FROM "user_profile" "a"
INNER JOIN "customer" "b"
ON "a"."id" = "b"."user_id"
WHERE "a"."age" > 25
ORDER BY "b"."spending" DESC
LIMIT 10
              

Join types

The following joins are available. Refer to the section Null values to avoid problems that may come up with joins.

							.joinOn(_.id, _.userId) // INNER JOIN

.innerJoinOn(_.id, _.userId) // INNER JOIN

.leftJoinOn(_.id, _.userId) // LEFT JOIN

.leftOuterJoinOn(_.id, _.userId) // LEFT OUTER JOIN
 
.rightJoinOn(_.id, _.userId) // RIGHT JOIN

.rightOuterJoinOn(_.id, _.userId) // RIGHT OUTER JOIN

.fullOuterJoinOn(_.id, _.userId) // FULL OUTER JOIN

.crossJoin // CROSS JOIN
            	

Insert

Basic insert

							sql
  .insert(user)
  .cols2(t => (
    t.username,
    t.email
  ))
  .run(("bob", "bob@mail.com"))
            	
								INSERT INTO "user_profile" ("username", "email") VALUES ('bob', 'bob@mail.com')
            	

Insert many

              sql
  .insert(user)
  .cols2(t => (
    t.username,
    t.email
  ))
  .runSeq(Seq(
    ("bob", "bob@mail.com"),
    ("jane", "jane@mail.com"),
    ("jack", "jack@mail.com")
  ))
              
                INSERT INTO "user_profile" ("username", "email")
VALUES ('bob', 'bob@mail.com'),
       ('jane', 'jane@mail.com'),
       ('jack', 'jack@mail.com')
              

Insert case class

              case class AddProduct(name: String, price: Float)

class Product extends Model("product") {
  val id = column[Int]("id")
  val name = column[String]("name")
  val price = column[Float]("price")
  val add = write[AddProduct](name, price)
}

val product = Model.get[Product]

sql
  .insert(product)
  .colsType(_.add)
  .run(AddProduct("Banana", 12.5))
              
                INSERT INTO "product" ("name", "price") VALUES ('Banana', 12.5)
              

Cache insert queries

							val stm = sql
  .insert(user)
  .cols2(t => (
    t.username,
    t.email
  ))
  .cache

stm.run(("bob", "bob@mail.com"))
            	
								INSERT INTO "user_profile" ("username", "email") VALUES ('bob', 'bob@mail.com')
            	

Insert key/value

If you need to insert columns that exceed the limits of a tuple, larger than 22.

							sql
  .insert(user)
  .data(t => Seq(
    t.username ==> "bob",
    t.email ==> "bob@mail.com"
  ))
  .run
            	
								INSERT INTO "user_profile" ("username", "email") VALUES ('bob', 'bob@mail.com')
            	

Insert returning

							sql
  .insert(user)
  .cols2(t => (
    t.username,
    t.email
  ))
  .returning3(t => (
    t.id,
    t.username,
    t.email
  ))
  .runHead(("bob", "bob@mail.com"))
            	
								INSERT INTO "user_profile"
("username", "email")
VALUES ('bob', 'bob@mail.com')
RETURNING
  "id",
  "username",
  "email"
            	

Insert on conflict do nothing

You can take advantage of ON CONFLICT DO NOTHING to avoid errors on columns with UNIQUE constraint.

							sql
  .insert(user)
  .cols2(t => (
    t.username,
    t.email
  ))
  .onConflictDoNothing
  .run(("bob", "bob@mail.com"))
            	
								INSERT INTO "user_profile"
("username", "email")
VALUES ('bob', 'bob@mail.com')
ON CONFLICT DO NOTHING
            	

Upsert

The updated column has to be one of the columns you intend to insert.

							sql
  .insert(user)
  .cols2(t => (
    t.username,
    t.email
  ))
  .onConflictOnColumn(_.username)
  .doUpdate(_.email)
  .run(("bob", "bob@hotmail.com"))
            	
								INSERT INTO "user_profile"
("username", "email")
VALUES ('bob', 'bob@mail.com')
ON CONFLICT ("username")
DO UPDATE SET "email" = 'bob@mail.com'
            	

Insert where not exists

If you need to avoid duplication on a column that does not have a unique constraint you can use whereNotExists. Also, if you are makeing multible insert statements concurrently, from a stream for example, you will run into problems using onConflictDoNothing.

							sql
  .insert(user)
  .cols2(t => (
    t.username,
    t.email
  ))
  .whereNotExists(_.username)
  .run(("bob", "bob@mail.com"))
            	
								INSERT INTO "user_profile"
("username", "email")
SELECT 'bob', 'bob@mail.com'
WHERE NOT EXISTS (
  SELECT 1
  FROM "user_profile"
  WHERE "username" = 'bob'
)
            	

Insert from select

							sql
  .insert(newsletter)
  .cols1(_.email)
  .fromSelect(sql
    .select(user)
    .cols1(_.email)
    .where(_.isActive === true)
  )
  .run
            	
								INSERT INTO "newsletter" ("email")
SELECT "email"
FROM "user_profile"
WHERE "is_active" = true
            	

Update

Update statement

See Update operators

								sql
  .update(user)
  .set(_.country ==> "JP")
  .where(_.id === 103)
  .run
            	
								UPDATE "user" SET "country" = 'JP' WHERE id = 103
            	

Update returning

								sql
  .update(user)
  .set(t => Seq(
    t.country ==> "IS",
    t.city ==> "Reykjavik"
  ))
  .where(_.id === 31)
  .returning4(t => (
    t.id,
    t.email,
    t.country,
    t.city
  ))
  .runHeadOpt
            	
								UPDATE "user_profile"
SET country = 'IS',
       city = 'Reykjavik'
WHERE "id" = 31
RETURNING
  "id",
  "email",
  "country",
  "city"
            	

Cache update

See Update operators

								val stm = sql
  .update(user)
  .cacheSet2(t => (
    t.isActive.cacheAssign,
    t.discount.cacheIncrement
  ))
  .cacheWhere1(_.id.cacheEq)

stm.run((true, 10), 27)
            	
								UPDATE "user_profile"
SET "is_active" = true,
    "discount" = "discount" + 10
WHERE "id" = 27
            	

Delete

Delete statement

								sql
  .delete(user)
  .where(_.id === 103)
  .run
            	
								DELETE FROM "user_profile" WHERE "id" = 103
            	

Cache delete

								val stm = sql
  .delete(user)
  .where(_.id === 103)
  .cache

stm.run

// or

val stm = sql
  .delete(user)
  .cacheWhere1(_.id.cacheEq)

stm.run(103)
            	
								DELETE FROM "user_profile" WHERE "id" = 103
            	

Delete returning

								val stm = sql
  .delete(user)
  .where(_.id === 103)
  .returning1(_.email)
  .runHeadOpt
            	
								DELETE FROM "user_profile"
WHERE "id" = 103
RETURNING "email"
            	

Aggregation

Count

								sql
  .count(user)
  .where(_.country === "IT")
  .runHead
            	
								SELECT count(*) FROM "user_profile" WHERE "country" = 'IT'
            	

Avg Max Min

								import kuzminki.api._
import kuzminki.fn._

sql
  .select(user)
  .cols3(t => (
    Avg.int(t.age),
    Max.int(t.age),
    Min.int(t.age)
  ))
  .where(_.country === "US")
  .runHead
            	
								SELECT
  avg("age"),
  max("age"),
  min("age")
FROM "user_profile"
WHERE "country" = 'US'
            	

Raw SQL

Select

								def rawStm(country: String, minAge: Int) =
  rsql"""SELECT * FROM "user_profile" WHERE country = $country AND age > $minAge"""

val job = for {
  users <- db.query(rawStm("TR", 25))
} yield users
							

Operations

								val username = "bob"
val email = "bob@mail.com"

db.exec(rsql"""INSERT INTO "user_profile" ("username", "email") VALUES ($username, $email)""")
            	

Data types and operators

Data types

Postgres Scala
varchar / text String
bool Boolean
int2 Short
int4 Int
int8 Long
float4 Float
float8 Double
numeric BigDecimal
time java.sql.Time
date java.sql.Date
timestamp java.sql.Timestamp

Operators

Operator Alternative Cache Type
=== matches cacheEq Any
!== not cacheNot Any
> gt cacheGt Numbers and time
< lt cacheLt Numbers and time
>= gte cacheGte Numbers and time
< lt cacheLte Numbers and time
~ reMatch cacheReMatch String
~* reIMatch cacheReIMatch String
!~ reNotMatch cacheNotReMatch String
!~* reNotIMatch cacheNotIReMatch String
like cacheLike String
startsWith cacheStartsWith String
endsWith cacheEndsWith String
similarTo cacheSimilarTo String
isNull Any
isNotNull Any
in Any
notIn Any
has Array
hasNot Array
overlap Array
overlapNot Array

Update operators

Update operator Cache operator Type
==> cacheAssign Any
+= cacheIncrement Numbers
-= cacheDecrement Numbers
prepend Array
append Array
remove Array