scala DSL for dynamically building cached cassandra prepared statements in scala

PreparedStatementBuilder

package com.mgl.digital.dataservices.utils

import com.datastax.driver.core._



class PreparedStatementBuilder(session: Session) {
  private var statementCache: Map[String, PreparedStatement] = Map()

  def insertInto(table: String): Insert =
    Insert(table, Nil, this)

  def prepare(query: String): PreparedStatement = {
    statementCache.get(query) match {
      case Some(q) => q
      case None =>
        val pep = session.prepare(query)
        statementCache = statementCache ++ Map(query -> pep)
        pep
    }
  }

  def executeAsync(st: BoundStatement): ResultSetFuture = {
    session.executeAsync(st)
  }
}

case class NamedParameter[T](label: String, value: ParameterValue[T])
case class ParameterValue[T](value: T)

object NamedParameterImplicits {
  implicit def stringToParameterValue(str: String): ParameterValue[String] = ParameterValue(str)
  implicit def objectToParameterValue(ob: Object): ParameterValue[Object] = ParameterValue(ob)
}

case class Insert(table: String, values: Seq[(String, Any)], preparedStatementBuilder: PreparedStatementBuilder) {

  def value(label: String, value: Any): Insert = this.copy(table, values :+ label -> value)

  def toStatement: String = {
    s"insert into $table ("+ values.map(_._1).mkString(",") + ") VALUES ( "+values.map(_ => "?").mkString(",")+" );"
  }

  def toPreparedStatement: PreparedStatement = {
    preparedStatementBuilder.prepare(toStatement)
  }

  def toBoundStatement: BoundStatement = {
    toPreparedStatement.bind(values.map(_._2): _*)
  }

  def executeAsync() = {
    preparedStatementBuilder.executeAsync(toBoundStatement)
  }
}

object Usage {
  def session: Session = ???

  val builder = new PreparedStatementBuilder(session)

  val bound = builder
    .insertInto("transactions")
    .value("id", "213")
    .value("notes", "notes")
    .toBoundStatement

  builder.
    insertInto("transactions")
    .value("id", "213")
    .value("notes", "notes")
    .executeAsync()

  session.execute(bound)
}