ZIO with http4s and doobie

If you’re using Scala and you’d like to implement reactive applications using Functional programming approach, there are many open source FP Scala libraries that enable you to write purely functional code.

But.. you might ask why would you use Functional programming Libraries?

Usually when you call a bench of statements, they will be executed right away which makes it harder to test and you couldn’t control any interaction that is executing and sometimes you need to use external libraries to test them and sometimes it wouldn’t be possible to test your program!. Every interaction with the outside of the context of your program or your function called “side effect”,

Functional programming approach enables you to control the effects!

How?

Functional programming is a declarative paradigm in which programs describe what you want to achieve and separates between the Data and its functionality. You can pass the Data through different functions in order to change its behaviors.

And functional programming deals with pure functions which are:

• Easy to reason about.
• Testable because for given inputs you get an output and if you pass the same inputs you will get the same result.
• Free of side effects which means the ouput has to be computed from the inputs without interacting with the outside World!

Functional programmers deal with effectful programs by describing them in immutable data types, Any interaction (with an API/Database/Console…) is called effectful program will be described in immutable values that you can compose, test and pass them through different functions and you can compute effectful programs from another and return them which is impossible in imperative paradigm, you cannot do that with statements.

In Scala standard library there is no data type that we can use to make functional effects and describe the different interactions with the outside world, this is why there are functional programing Scala libraries that provide this data type the most popular are: cats-effect, Monix and ZIO

In this article, we’re going to use ZIO

ZIO

ZIO is a zero dependency Scala library that helps us to describe effectful programs, and build asynchronous, concurrent applications and provides resource safety, ZIO enables us to write purely functional code.

First step: add zio to your dependencies

"dev.zio" %% "zio" %  ZIOVersion

→ This example uses the version: 1.0.0-RC18-2

Second step: Overview of the functional effect build on ZIO

The functional effect data type build on zio is called ZIO :

• ZIO[R, E, A]: description of an effectful program that has dependencies of type R which enables us to use the functionalities of R then at the end we can provide different implementations for Production and for Test, and this functional effect might fail with an error of type E or produce a value of type A.

There are different type aliases in zio that you can use if you would like to have less type parameters:

• UIO[A]: a description of an effectful program that computes a value of type A and never fail.
• Task[A]: a description of an effectful program that might fail with a Throwable or produce a value of type A.
• IO[E, A]: a description of an effectful program that might fail with any value of type E or produce a value of type A.
• RIO[R, A]: the same as ZIO but this program might fail with an error of type Throwable and produces a value of type A.

After describing execute!

After building your effectful programs which are only a description, you would need to run them at the end of the day in order to interact with the real world,

Using the runtime system built on ZIO: DefaultRuntime you can call unsafeRun

In this blog, you will learn how to use ZIO to interact with an HTTP Api and a Database using http4s and Doobie.

Motivation:

We’re going to use:

Http4s

http4s is Type safe, functional, streaming HTTP for Scala. its servers and clients share an immutable model of requests and responses. Http4s deals with I/O using cats effect.

Add to your dependencies:

"org.http4s" %% "http4s-blaze-server" % "0.21.1",
"org.http4s" %% "http4s-circe" % "0.21.1",
"org.http4s" %% "http4s-dsl" % "0.21.1"

Doobie

doobie is a pure functional JDBC layer for Scala and Cats. It provides a functional way to construct programs that use JDBC.

Add to your dependencies:

"org.tpolecat" %% "doobie-core" % "0.8.8",
"org.tpolecat" %% "doobie-h2" % "0.8.8"

How could you use ZIO with Http4s and Doobie?

ZIO has a separate module called: zio-interop-cats
which contains instances for the Cats Effect library, and allow you to use ZIO with any libraries that rely on Cats Effect like in our case Http4s and Doobie.

You could add this module to your dependencies:

"dev.zio" %% "zio-interop-cats" % "2.0.0.0-RC12"

Let’s implement a simple application that enable us to create/read/delete a user in Database via HTTP calls:

Application Dependencies:

Using ZIO environment we can define the dependencies that are used in our Application and we can have different implementations for these dependencies.

1. Configuration:

The configuration is a dependency which is required in our application, it has:

• The API endpoint and port
• Database configuration

case class Config(api: ApiConfig, dbConfig: DbConfig)
case class ApiConfig(endpoint: String, port: Int)
case class DbConfig(url: String, user: String, password: String)

You can add application.conf that contains the api and db configuration, as a simple example we’re going to use h2 in-memory db

api {
  endpoint = "127.0.0.1"
  port = 8083
}

db-config {
  url = "jdbc:h2:~/test;DB_CLOSE_DELAY=-1"
  user = ""
  password = ""
}

1. Define Configuration module: the abstraction of the configuration API

object Configuration {
  trait Service {
    val load: Task[Config]
  }
}

load returns a functional effect that doesn’t require anything, might fail with a Throwable and return Config in case of success.

2. Representation of the configuration dependency:

package configuration {
 type Configuration = zio.Has[Configuration.Service]
 ...
}

Has enables us to define the implementation of the service on ZLayer this will make it easier to combine the dependencies of our App.

To specify an effect that requires Configuration we can use this type alias.

example:

val database: ZIO[Configuration, Throwable, DB] = ???

which is the same as:

val database: ZIO[Has[Configuration.Service], Throwable, DB] = ???

3. Access the API :a helper that accesses the functions of the dependency:

package configuration {
 val load:ZIO[Configuration, Throwable, AppConfig] = 
  ZIO.accessM(_.get.load)
...
}

- Has#get retrieves the service and accesses its methods.

Note: we defined the helper in the package to have a scope for the dependency functionalities, and this makes it easier to use them in the App.

But you don’t have to define them, if you would like to do that with different way you can access different dependencies in your App directly using:

ZIO.accessM[Dependency1](_.functionality1)

4. Implementation for Configuration using ZLayer

In production we can use pureconfig to load the configuration file that

contains the configuration of the API and the Database.

"com.github.pureconfig" %% "pureconfig" % "0.11.0"

The implementation of load using pureConfig would be:

trait Live extends Configuration.Service {
  val load: Task[Config] = 
    Task.effect(loadConfigOrThrow[Config])
}

ZLayer wraps the service Live ‘implementation:

object Configuration {
  ...
 val live: ZLayer[Any, Nothing, Configuration] = 
    ZLayer.succeed(new Live {} )
}

ZLayer[A, E, B] is a layer of the application that requires service(s) A (in our case no requirements is needed), might fail with an error of type E, and produces service(s) B .

B represents the current service (Configuration.Service)

In our case, Configuration doesn’t have any dependencies and provides an implementation for the Configuration.Service.

How to use layer?

Later when we want to use a specific implementation (for Test or Production) we can provide the layer.
example:
zio.provideLayer(Configuration.live)

→ Checkout the code with this approach here. In this section I wanted to go with a simple approach, there is a tip section bellow explains a better way to implement Configuration dependency.

Now let’s move to the next dependency with the same way.

2. Database:

We can make a module that describes the operations that persist the user data in our Database, but what if we would like to use these operations for other types for example Account, Profile etc?

Let’s the service Api generic!

object Persistence {
  trait Service[A] {
    def get(id: Int): Task[A]
    def create(user: User): Task[A]
    def delete(id: Int): Task[Boolean]
  }
}

Now we can implement the methods in Persistence.Service for the Production module using Doobie, but every query needs to be executed in a transactor that will wrap the connection pool provided by our database that we want to choose (in our example it would be h2) specifying a target effect type that will take the effectful computation, in our case we will use zio.Task then when the transaction will be performed by calling transact we wanna get a zio.Task but .. transact requires implicit ev: Bracket[M, Throwable] you can solve that by importing: zio.interop.catz.taskConcurrentInstances

import doobie.{ Query0, Transactor, Update0 }
import zio._
import doobie.implicits._
import zio.interop.catz._
final class UserPersistenceService(tnx: Transactor[Task]) extends Persistence.Service[User] {
  import UserPersistenceService._

  def get(id: Int): Task[User] =
    SQL
      .get(id)
      .option
      .transact(tnx)
      .foldM(
        err => Task.fail(err),
        maybeUser => Task.require(UserNotFound(id))(Task.succeed(maybeUser))
      )

  def create(user: User): Task[User] =
    SQL
      .create(user)
      .run
      .transact(tnx)
      .foldM(err => Task.fail(err), _ => Task.succeed(user))

  def delete(id: Int): Task[Boolean] =
    SQL
      .delete(id)
      .run
      .transact(tnx)
      .fold(_ => false, _ => true)
}
object UserPersistenceService {

  object SQL {

    def get(id: Int): Query0[User] =
      sql"""SELECT * FROM USERS WHERE ID = $id """.query[User]

    def create(user: User): Update0 =
      sql"""INSERT INTO USERS (id, name) VALUES (${user.id}, ${user.name})""".update

    def delete(id: Int): Update0 =
      sql"""DELETE FROM USERS WHERE id = $id""".update
  }
...
}

How could we make a transaction?

We need to reserve the transaction and use it in the application and when the application is closed, the resource should be cleaned up.

In ZIO there is a data type called zio.Managed that describes a managed resource, the resource will be acquired before it is used and automatically released after using it.

In our example, the creation of a new H2Transactor returns cats.effect.Resource .

In zio-interop we can turn Resource into Managed by calling toManagedZIO as following:

import doobie.h2.H2Transactor
import scala.concurrent.ExecutionContext
import zio.Task
import zio.interop.catz._
object UserPersistenceService {
def mkTransactor(
  conf: DbConfig,
  connectEC: ExecutionContext,
  transactEC: ExecutionContext
 ): Managed[Throwable, H2Transactor[Task]] = {
  H2Transactor
    .newH2Transactor[Task](conf.url,
      conf.user,
      conf.password,
      connectEC,
      Blocker.liftExecutionContext(transactEC)
    )
    .toManagedZIO
    .map(new UserPersistenceService(_))
}

Cool! Now how could we make the layer from this Managed resources?

We can use ZLayer.fromManaged ! and we want to include in the acquired action the implementation of the UserPersistenceService

we can use map in managed :

Managed(res).map(new UserPersistenceService(_))

Awesome! now the mkTransactor returns: Managed[Throwable, UserPersistenceService]

so we can create our ZLayer, but wait ! we need a DB configuration which means that layer has a dependency of type Configuration, and 2 execution contexts! connectEc could be the same execution context that our App uses, transactEC needs to execute its operation in a blocking execution context, we can use the execution context of zio.blocking !

object UserPersistenceService {
  val live: 
     ZLayer[Configuration, Throwable, Has[UserPersistenceService]] =
  ZLayer.fromManaged (
    (for {
      config <- configuration.loadConfig.toManaged_
      connectEC <- ZIO.descriptor.map(_.executor.asEC).toManaged_          
      blockingEC <- blocking { ZIO.descriptor.map(_.executor.asEC) 
                     }.toManaged_
      managed <- mkTransactor(config.dbConfig, connectEC,     blockingEC)
    } yield managed).provideSomeLayer[Configuration](Blocking.live)
  )
}

Awesome! 🤩

You can add a type alias for Has[UserPersistenceService] if you want to :-)

type UserPersistence = Has[UserPersistenceService]

We used above Blocking.live so the layer wouldn’t require a Blocking dependency because we have provided it, but you can make it as a requirement:

val live: ZLayer[Configuration with Blocking, Throwable, UserPersistence] =
  ZLayer.fromManaged (
    for {
      config <- configuration.loadConfig.toManaged_
      connectEC <- ZIO.descriptor.map(_.executor.asEC).toManaged_    
      blockingEC <- blocking { ZIO.descriptor.map(_.executor.asEC)   
                     }.toManaged_
      managed <- mkTransactor(config.dbConfig, connectEC, blockingEC)
    } yield managed
  )

For the test we can use zio.Ref to persist the data:

case class Test(users: Ref[Vector[User]]) extends Service {
    def get(id: Int): Task[User] =
      users.get.flatMap(users => Task.require(UserNotFound(id))(
        Task.succeed(users.find(_.id == id))))
    def create(user: User): Task[User] =
      users.update(_ :+ user).map(_ => user)
    def delete(id: Int): Task[Boolean]    =
      users.modify(users => true -> users.filterNot(_.id == id))  
}

then we can build a ZLayer from function that initializes the Ref as following:

def test(users: Ref[Vector[User]]): Layer[Nothing, UserPersistence] =
  ZLayer.fromEffect(Ref.make(Vector.empty[User]).map(Test(_)))

Note: You can make the Transactor[Task] as a dependency then the UserPersistence layer will require it, then you can make different transactor’s implementations for test and production.

3. Http Api

Using Http4s, let’s implement our HttpRoutes with the following HTTP calls:

• GET → get the user for a given id
• POST → create a new user
• DELETE → delete a user with a given id

HttpRoutes returns the result in our effectful program which uses Persistence service for the Database interaction and it might use more other services.

Thanks to the helper functions, we can use: persistence.getUser(???) , persistence.createUser(???) and persistence.deleteUser(???)

Let’s describe the routes :

import io.circe.{Decoder, Encoder}
import org.http4s.{EntityDecoder, EntityEncoder, HttpRoutes}
import org.http4s.dsl.Http4sDsl
import zio._
import org.http4s.circe._
import zio.interop.catz._
import io.circe.generic.auto._
type UserTask[A] = RIO[R, A]
def routes: HttpRoutes[UserTask] =
  HttpRoutes.of[UserTask] {
    case GET -> Root / IntVar(id) => 
      getUser(id).foldM(_ => NotFound(), Ok(_))
    case request @ POST -> Root =>
      request.decode[User] { user =>
        Created(createUser(user))
      }
    case DELETE -> Root / IntVar(id) =>
      (get(id) *> deleteUser(id)).foldM(_ => NotFound(), Ok(_))
  }

But.. in order to return the response that contains User wrapped in the UserTask we have to define an implicit: EntityEncoder[UserTask, A] .

And in order to be able to decode the body in POST Request, we need to define an implicit: EntityDecoder[UserTask, A]

We can use http4s-circe for that:

implicit def circeJsonDecoder[A](implicit decoder: Decoder[A]): 
  EntityDecoder[UserTask, A] = jsonOf[UserTask, A]
implicit def circeJsonEncoder[A](implicit decoder: Encoder[A]): 
  EntityEncoder[UserTask, A] = jsonEncoderOf[UserTask, A]

Don’t forget to import zio.interop.catz.taskConcurrentInstances ! Because Http4s deals with I/O using cats effect and ZIO can interoperate with cats effects and make it zio.Task.

Cool!

Now in the Main we can interact with the real world providing the layers that we have implemented for production and also our application is easy to test using Test modules so we can check our zio.Ref instead of interacting with the Database.

4. Interaction with the real World!

Create your Main App using the zio interpreter:

• extend zio.App
• implement run which returns a functional effect that:

→ Requires ZEnv (ZEnv gather all the environments build on zio to enable the user to use them for free)

→ Never fails: so we should handle the errors

→ Produces Int which is the exit code of the program.

object Main extends zio.App {
 def run(args: List[String]): ZIO[ZEnv, Nothing, Int] = ???
}

In order to open the Api connection we will use BlazeServerBuilder to serve the requests that requires cats.effect.Timer and cats.effect.ConcurrentEffect in order to convert them respectively to zio.Clock and zio.Task we can import: zio.interop.catz._ and we can define our types:

type AppEnvironment = Configuration with Clock with UserPersistence

And we need to load the configuration:

val program = for {
  config <- configuration.load
  
  httpApp = Router[AppTask](
  "/users" -> Api(s"${conf.api.endpoint}/users").route
).orNotFound

server <- ZIO.runtime[AppEnvironment].flatMap { implicit rts =>
  db.createTable *>
    BlazeServerBuilder[AppTask] //requires a Clock environment
      .bindHttp(conf.api.port, "0.0.0.0")
      .withHttpApp(CORS(httpApp))
      .serve
      .compile[AppTask, AppTask, ExitCode]
      .drain
} yield server

And finally we need to provide the AppEnvironment dependency using the production implementation.

• The persistence layer requires Blocking with Configuration

val userPersistence = (Configuration.live ++ Blocking.live) >>> UserPersistence.live

• ++ combines two layers
• >>> feeds the persistence layer with the layers in the left hand side

Awesome !

Our program needs also a Configuration and Clock and UserPersistence

Because Clock is built in zio and it is a part of ZEnv we can leave it without providing it, zio will do that internally.

So we can use provideSomeLayer[ZEnv] in which you have to specify the dependencies minus ZEnv, in our case Configuration and UserPersistence

val io = program.provideSomeLayer[ZEnv](Configuration.live ++ userPersistence)

And then you can recover from error and return the exit code to make your zio program run!

io.fold(_ => 1, _ => 0)

Cool! We did it!

Before we finish, I would like to give you a tip that you can use in your dependencies

Tip:

As you saw Configuration dependency is needed to create the server using the ApiConfig and to use the data base using the DbConfig, and both of these programs call configuration.load then each of them take the specific configuration (either config.api or config.dbConfig) we can make it more specific and say that we have a dependency that has the information of ApiConfig and DbConfig :

type Configuration = Has[ApiConfig] with Has[DbConfig]

The Production layer and test layer will be made from an effect that loads the configuration using fromEffectMany where the R of ZIO would be Has[ApiConfig] with Has[DbConfig] :

val live: Layer[Throwable, Configuration] = ZLayer.fromEffectMany(
    Task
      .effect(loadConfigOrThrow[Config])
      .map(config => Has(config.api) ++ Has(config.dbConfig)))
}

If the layer doesn’t have a dependency you can use the type aliasLayer like in this case.

In order to access them you can do this:

val apiConfig: ZIO[Has[ApiConfig], Nothing, ApiConfig] = 
  ZIO.access(_.get)
val dbConfig: ZIO[Has[DbConfig], Nothing, DbConfig] =     
  ZIO.access(_.get)

Now let’s see the usage of Configuration dependency in our App and replace it with this new approach:

• UserPersistence requires the dbConfig

val live: ZLayer[Has[DbConfig] with Blocking, Throwable, UserPersistence] =
  ZLayer.fromManaged (
    for {
      config <- configuration.dbConfig.orDie.toManaged_        
      connectEC <- ZIO.descriptor.map(_.executor.asEC).toManaged_      
      blockingEC <- blocking.blocking {    
         ZIO.descriptor.map(_.executor.asEC) }.toManaged_
      managed <- mkTransactor(config, connectEC, blockingEC)
    } yield managed
  )

• persistence layer requires Blocking and DBConfig: We can use the combinators of ZLayer exactly like we did before

val userPersistence = (Configuration.live ++ Blocking.live) >>> UserPersistenceService.live

• httpApp : uses the apiConfig

val program = for {
  api <- configuration.apiConfig
  httpApp = Router[AppTask](
  "/users" -> Api(s"${api.endpoint}/users").route).orNotFound
...
} ...

Then at the end we can provide the AppEnvironment:

program.provideSomeLayer[ZEnv](Configuration.live ++ userPersistence)

If you are interested about the code check out this repository.

As we have seen, each step was a description of what we want to achieve, this code is purely functional because:

• We created effects, composed them together and passed them through different functions.
• The types tell a lot about what we’re doing.
• We were able to control the effects and test them.

I hope you learned more about the power of ZIO and how it can interoperate with other libraries and especially how to use the environment feature using ZLayer.

Credit

zio-todo-backend was a good guidance to get started with Http4s and doobie using ZIO. Thanks Maxim for the nice example!

Thanks to John De Goes and to all zio contributors for making zio awesome! ❤️

Zio 👋!