在上一篇博文里我们把JDBC-Engine的读取操作部分分离出来进行了讨论，在这篇准备把更新Update部分功能介绍一下。当然，JDBC-Engine的功能是基于ScalikeJDBC的，所有的操作和属性都包嵌在SQL这个类型中：

  1. /**
  2.  * SQL abstraction.
  3.  *
  4.  * @param statement SQL template
  5.  * @param rawParameters parameters
  6.  * @param f  extractor function
  7.  * @tparam A return type
  8.  */
  9. abstract class SQL[A, E <: WithExtractor](
  10.   val statement: String,
  11.   private[scalikejdbc] val rawParameters: Seq[Any]
  12. )(f: WrappedResultSet => A)
  13. {...}

Update功能置于下面这几个子类中：

  1. /**
  2.  * SQL which execute java.sql.Statement#executeUpdate().
  3.  *
  4.  * @param statement SQL template
  5.  * @param parameters parameters
  6.  * @param before before filter
  7.  * @param after after filter
  8.  */
  9. class SQLUpdate(val statement: String, val parameters: Seq[Any], val tags: Seq[String] = Nil)(
  10.     val before: (PreparedStatement) => Unit
  11. )(
  12.     val after: (PreparedStatement) => Unit
  13. ) {
  15.   def apply()(implicit session: DBSession): Int = {
  16.     val attributesSwitcher = new DBSessionAttributesSwitcher(SQL("").tags(tags: _*))
  17.     session match {
  18.       case AutoSession =>
  19.         DB.autoCommit(DBSessionWrapper(_, attributesSwitcher).updateWithFilters(before, after, statement, parameters: _*))
  20.       case NamedAutoSession(name, _) =>
  21.         NamedDB(name, session.settings).autoCommit(DBSessionWrapper(_, attributesSwitcher).updateWithFilters(before, after, statement, parameters: _*))
  22.       case ReadOnlyAutoSession =>
  23.         DB.readOnly(DBSessionWrapper(_, attributesSwitcher).updateWithFilters(before, after, statement, parameters: _*))
  24.       case ReadOnlyNamedAutoSession(name, _) =>
  25.         NamedDB(name, session.settings).readOnly(DBSessionWrapper(_, attributesSwitcher).updateWithFilters(before, after, statement, parameters: _*))
  26.       case _ =>
  27.         DBSessionWrapper(session, attributesSwitcher).updateWithFilters(before, after, statement, parameters: _*)
  28.     }
  29.   }
  31. }
  33. /**
  34.  * SQL which execute java.sql.Statement#execute().
  35.  *
  36.  * @param statement SQL template
  37.  * @param parameters parameters
  38.  * @param before before filter
  39.  * @param after after filter
  40.  */
  41. class SQLExecution(val statement: String, val parameters: Seq[Any], val tags: Seq[String] = Nil)(
  42.     val before: (PreparedStatement) => Unit
  43. )(
  44.     val after: (PreparedStatement) => Unit
  45. ) {
  47.   def apply()(implicit session: DBSession): Boolean = {
  48.     val attributesSwitcher = new DBSessionAttributesSwitcher(SQL("").tags(tags: _*))
  49.     val f: DBSession => Boolean = DBSessionWrapper(_, attributesSwitcher).executeWithFilters(before, after, statement, parameters: _*)
  50.     // format: OFF
  51.     session match {
  52.       case AutoSession                       => DB.autoCommit(f)
  53.       case NamedAutoSession(name, _)         => NamedDB(name, session.settings).autoCommit(f)
  54.       case ReadOnlyAutoSession               => DB.readOnly(f)
  55.       case ReadOnlyNamedAutoSession(name, _) => NamedDB(name, session.settings).readOnly(f)
  56.       case _                                 => f(session)
  57.     }
  58.     // format: ON
  59.   }
  61. }
  62. /**
  63.  * SQL which execute java.sql.Statement#executeBatch().
  64.  *
  65.  * @param statement SQL template
  66.  * @param parameters parameters
  67.  */
  68. class SQLBatch(val statement: String, val parameters: Seq[Seq[Any]], val tags: Seq[String] = Nil) {
  70.   def apply[C[_]]()(implicit session: DBSession, cbf: CanBuildFrom[Nothing, Int, C[Int]]): C[Int] = {
  71.     val attributesSwitcher = new DBSessionAttributesSwitcher(SQL("").tags(tags: _*))
  72.     val f: DBSession => C[Int] = DBSessionWrapper(_, attributesSwitcher).batch(statement, parameters: _*)
  73.     // format: OFF
  74.     session match {
  75.       case AutoSession                       => DB.autoCommit(f)
  76.       case NamedAutoSession(name, _)         => NamedDB(name, session.settings).autoCommit(f)
  77.       case ReadOnlyAutoSession               => DB.readOnly(f)
  78.       case ReadOnlyNamedAutoSession(name, _) => NamedDB(name, session.settings).readOnly(f)
  79.       case _                                 => f(session)
  80.     }
  81.     // format: ON
  82.   }
  84. }

按照JDBC-Engine的功能设计要求，我们大约把Update功能分成数据表构建操作DDL、批次运算Batch、和普通Update几种类型。我们是通过JDBCContext来定义具体的Update功能类型：

  1. object JDBCContext {
  2.     type SQLTYPE = Int
  3.     val SQL_SELECT: Int = 0
  4.     val SQL_EXEDDL= 1
  5.     val SQL_UPDATE = 2
  6.     val RETURN_GENERATED_KEYVALUE = true
  7.     val RETURN_UPDATED_COUNT = false
  9.   }
  11.   case class JDBCContext(
  12.                           dbName: Symbol,
  13.                           statements: Seq[String] = Nil,
  14.                           parameters: Seq[Seq[Any]] = Nil,
  15.                           fetchSize: Int = 100,
  16.                           queryTimeout: Option[Int] = None,
  17.                           queryTags: Seq[String] = Nil,
  18.                           sqlType: JDBCContext.SQLTYPE = JDBCContext.SQL_SELECT,
  19.                           batch: Boolean = false,
  20.                           returnGeneratedKey: Seq[Option[Any]] = Nil,
  21.                           // no return: None, return by index: Some(1), by name: Some("id")
  22.                           preAction: Option[PreparedStatement => Unit] = None,
  23.                           postAction: Option[PreparedStatement => Unit] = None) {
  25.     ctx =>
  27.     //helper functions
  29.     def appendTag(tag: String): JDBCContext = ctx.copy(queryTags = ctx.queryTags :+ tag)
  31.     def appendTags(tags: Seq[String]): JDBCContext = ctx.copy(queryTags = ctx.queryTags ++ tags)
  33.     def setFetchSize(size: Int): JDBCContext = ctx.copy(fetchSize = size)
  35.     def setQueryTimeout(time: Option[Int]): JDBCContext = ctx.copy(queryTimeout = time)
  37.     def setPreAction(action: Option[PreparedStatement => Unit]): JDBCContext = {
  38.       if (ctx.sqlType == JDBCContext.SQL_UPDATE &&
  39.         !ctx.batch && ctx.statements.size == 1)
  40.         ctx.copy(preAction = action)
  41.       else
  42.         throw new IllegalStateException("JDBCContex setting error: preAction not supported!")
  43.     }
  45.     def setPostAction(action: Option[PreparedStatement => Unit]): JDBCContext = {
  46.       if (ctx.sqlType == JDBCContext.SQL_UPDATE &&
  47.         !ctx.batch && ctx.statements.size == 1)
  48.         ctx.copy(postAction = action)
  49.       else
  50.         throw new IllegalStateException("JDBCContex setting error: preAction not supported!")
  51.     }
  53.     def appendDDLCommand(_statement: String, _parameters: Any*): JDBCContext = {
  54.       if (ctx.sqlType == JDBCContext.SQL_EXEDDL) {
  55.         ctx.copy(
  56.           statements = ctx.statements ++ Seq(_statement),
  57.           parameters = ctx.parameters ++ Seq(Seq(_parameters))
  58.         )
  59.       } else
  60.         throw new IllegalStateException("JDBCContex setting error: option not supported!")
  61.     }
  63.     def appendUpdateCommand(_returnGeneratedKey: Boolean, _statement: String, _parameters: Any*): JDBCContext = {
  64.       if (ctx.sqlType == JDBCContext.SQL_UPDATE && !ctx.batch) {
  65.         ctx.copy(
  66.           statements = ctx.statements ++ Seq(_statement),
  67.           parameters = ctx.parameters ++ Seq(_parameters),
  68.           returnGeneratedKey = ctx.returnGeneratedKey ++ (if (_returnGeneratedKey) Seq(Some(1)) else Seq(None))
  69.         )
  70.       } else
  71.         throw new IllegalStateException("JDBCContex setting error: option not supported!")
  72.     }
  74.     def appendBatchParameters(_parameters: Any*): JDBCContext = {
  75.       if (ctx.sqlType != JDBCContext.SQL_UPDATE || !ctx.batch)
  76.         throw new IllegalStateException("JDBCContex setting error: batch parameters only supported for SQL_UPDATE and batch = true!")
  78.       var matchParams = true
  79.       if (ctx.parameters != Nil)
  80.         if (ctx.parameters.head.size != _parameters.size)
  81.           matchParams = false
  82.       if (matchParams) {
  83.         ctx.copy(
  84.           parameters = ctx.parameters ++ Seq(_parameters)
  85.         )
  86.       } else
  87.         throw new IllegalStateException("JDBCContex setting error: batch command parameters not match!")
  88.     }
  90.     def setBatchReturnGeneratedKeyOption(returnKey: Boolean): JDBCContext = {
  91.       if (ctx.sqlType != JDBCContext.SQL_UPDATE || !ctx.batch)
  92.          throw new IllegalStateException("JDBCContex setting error: only supported in batch update commands!")
  93.       ctx.copy(
  94.         returnGeneratedKey = if (returnKey) Seq(Some(1)) else Nil
  95.       )
  96.     }
  98.      def setQueryCommand(_statement: String, _parameters: Any*): JDBCContext = {
  99.         ctx.copy(
  100.           statements = Seq(_statement),
  101.           parameters = Seq(_parameters),
  102.           sqlType = JDBCContext.SQL_SELECT,
  103.           batch = false
  104.         )
  105.       }
  107.       def setDDLCommand(_statement: String, _parameters: Any*): JDBCContext = {
  108.         ctx.copy(
  109.           statements = Seq(_statement),
  110.           parameters = Seq(_parameters),
  111.           sqlType = JDBCContext.SQL_EXEDDL,
  112.           batch = false
  113.         )
  114.       }
  116.       def setUpdateCommand(_returnGeneratedKey: Boolean, _statement: String, _parameters: Any*): JDBCContext = {
  117.         ctx.copy(
  118.           statements = Seq(_statement),
  119.           parameters = Seq(_parameters),
  120.           returnGeneratedKey = if (_returnGeneratedKey) Seq(Some(1)) else Seq(None),
  121.           sqlType = JDBCContext.SQL_UPDATE,
  122.           batch = false
  123.         )
  124.       }
  125.       def setBatchCommand(_statement: String): JDBCContext = {
  126.         ctx.copy (
  127.           statements = Seq(_statement),
  128.           sqlType = JDBCContext.SQL_UPDATE,
  129.           batch = true
  130.         )
  131.       }
  132.   }

JDBCContext还提供了不少的Helper函数来协助构建特别功能的JDBCContext对象，如：setQueryCommand, setDDLCommand, setUpdateCommand, setBatchCommand。这些Helper函数提供Update功能定义的几个主要元素包括：SQL语句主体包括参数占位的statement、输入参数parameter、是否需要返回系统自动产生的主键returnGeneratedKey。在ScalikeJDBC中所有类型的Update功能可以用下面几类内部函数实现，包括：

  1.   private[this] def batchInternal[C[_], A](
  2.     template: String,
  3.     paramsList: Seq[Seq[Any]],
  4.     execute: StatementExecutor => scala.Array[A]
  5.   )(implicit cbf: CanBuildFrom[Nothing, A, C[A]]): C[A] = {
  6.     ensureNotReadOnlySession(template)
  7.     paramsList match {
  8.       case Nil => Seq.empty[A].to[C]
  9.       case _ =>
  10.         using(createBatchStatementExecutor(
  11.           conn = conn,
  12.           template = template,
  13.           returnGeneratedKeys = false,
  14.           generatedKeyName = None
  15.         )) { executor =>
  16.           paramsList.foreach {
  17.             params =>
  18.               executor.bindParams(params)
  19.               executor.addBatch()
  20.           }
  21.           execute(executor).to[C]
  22.         }
  23.     }
  24.   }
  25.   private[this] def updateWithFiltersInternal[A](
  26.     returnGeneratedKeys: Boolean,
  27.     before: (PreparedStatement) => Unit,
  28.     after: (PreparedStatement) => Unit,
  29.     template: String,
  30.     execute: StatementExecutor => A,
  31.     params: Seq[Any]
  32.   ): A = {
  33.     ensureNotReadOnlySession(template)
  34.     using(createStatementExecutor(
  35.       conn = conn,
  36.       template = template,
  37.       params = params,
  38.       returnGeneratedKeys = returnGeneratedKeys
  39.     )) {
  40.       executor =>
  41.         before(executor.underlying)
  42.         val count = execute(executor)
  43.         after(executor.underlying)
  44.         count
  45.     }
  46.   }
  47.   private[this] def updateWithAutoGeneratedKeyNameAndFiltersInternal[A](
  48.     returnGeneratedKeys: Boolean,
  49.     generatedKeyName: String,
  50.     before: (PreparedStatement) => Unit,
  51.     after: (PreparedStatement) => Unit,
  52.     template: String,
  53.     execute: StatementExecutor => A,
  54.     params: Seq[Any]
  55.   ): A = {
  56.     ensureNotReadOnlySession(template)
  57.     using(createStatementExecutor(
  58.       conn = conn,
  59.       template = template,
  60.       params = params,
  61.       returnGeneratedKeys = returnGeneratedKeys,
  62.       generatedKeyName = Option(generatedKeyName)
  63.     )) {
  64.       executor =>
  65.         before(executor.underlying)
  66.         val count = execute(executor)
  67.         after(executor.underlying)
  68.         count
  69.     }
  70.   }

我们可以看到所有类型的Update都是通过构建StatementExecutor并按其属性进行运算来实现的：

  1. /**
  2.  * java.sql.Statement Executor.
  3.  *
  4.  * @param underlying preparedStatement
  5.  * @param template SQL template
  6.  * @param singleParams parameters for single execution (= not batch execution)
  7.  * @param isBatch is batch flag
  8.  */
  9. case class StatementExecutor(
  10.     underlying: PreparedStatement,
  11.     template: String,
  12.     connectionAttributes: DBConnectionAttributes,
  13.     singleParams: Seq[Any] = Nil,
  14.     tags: Seq[String] = Nil,
  15.     isBatch: Boolean = false,
  16.     settingsProvider: SettingsProvider = SettingsProvider.default
  17. ) extends LogSupport with UnixTimeInMillisConverterImplicits with AutoCloseable {...}

这个StatementExcutor类的属性和我们的JDBCContext属性很接近。好了，回到JDBC-Engine Update功能定义。首先是DDL功能：

  1.  def jdbcExcuteDDL(ctx: JDBCContext): Try[String] = {
  2.        if (ctx.sqlType != SQL_EXEDDL) {
  3.         Failure(new IllegalStateException("JDBCContex setting error: sqlType must be 'SQL_EXEDDL'!"))
  4.       }
  5.       else {
  6.         NamedDB(ctx.dbName) localTx { implicit session =>
  7.           Try {
  8.                 ctx.statements.foreach { stm =>
  9.                   val ddl = new SQLExecution(statement = stm, parameters = Nil)(
  10.                     before = WrappedResultSet => {})(
  11.                     after = WrappedResultSet => {})
  13.                   ddl.apply()
  14.               }
  15.             "SQL_EXEDDL executed succesfully."
  16.           }
  17.         }
  18.       }
  19.     }

 所有JDBC-Engine的Update功能都是一个事务处理Transaction中的多条更新语句。DDL语句不需要参数所以只需要提供statement就足够了。下面是这个函数的使用示范：

  1.  ConfigDBsWithEnv("dev").setup('h2)
  2.   ConfigDBsWithEnv("dev").loadGlobalSettings()
  4.   val dropSQL: String ="""
  5.       drop table members
  6.     """
  8.   val createSQL: String ="""
  9.     create table members (
  10.       id serial not null primary key,
  11.       name varchar(30) not null,
  12.       description varchar(1000),
  13.       birthday date,
  14.       created_at timestamp not null,
  15.       picture blob
  16.     )"""
  18.   var ctx = JDBCContext('h2)
  19.     try {
  20.       ctx = ctx.setDDLCommand(dropSQL)
  21.         .appendDDLCommand(createSQL)
  22.     }
  23.     catch {
  24.        case e: Exception => println(e.getMessage)
  25.     }
  27.   val resultCreateTable = jdbcExcuteDDL(ctx)
  29.   resultCreateTable match {
  30.     case Success(msg) => println(msg)
  31.     case Failure(err) => println(s"${err.getMessage}")
  32.   }

在这里我们修改了上次使用的members表，增加了一个blob类的picture列。这个示范在一个完整的Transaction里包括了两条DDL语句。

批次更新batch-update是指多条输入参数在一条统一的statement上施用：

  1.   def jdbcBatchUpdate[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(
  2.       implicit cbf: CanBuildFrom[Nothing, Long, C[Long]]): Try[C[Long]] = {
  3.       if (ctx.statements == Nil)
  4.         throw new IllegalStateException("JDBCContex setting error: statements empty!")
  5.       if (ctx.sqlType != SQL_UPDATE) {
  6.         Failure(new IllegalStateException("JDBCContex setting error: sqlType must be 'SQL_UPDATE'!"))
  7.       }
  8.       else {
  9.         if (ctx.batch) {
  10.           if (noReturnKey(ctx)) {
  11.             val usql = SQL(ctx.statements.head)
  12.               .tags(ctx.queryTags: _*)
  13.               .batch(ctx.parameters: _*)
  14.             Try {
  15.               NamedDB(ctx.dbName) localTx { implicit session =>
  16.                 ctx.queryTimeout.foreach(session.queryTimeout(_))
  17.                 usql.apply[Seq]()
  18.                 Seq.empty[Long].to[C]
  19.               }
  20.             }
  21.           } else {
  22.             val usql = new SQLBatchWithGeneratedKey(ctx.statements.head, ctx.parameters, ctx.queryTags)(None)
  23.             Try {
  24.               NamedDB(ctx.dbName) localTx { implicit session =>
  25.                 ctx.queryTimeout.foreach(session.queryTimeout(_))
  26.                 usql.apply[C]()
  27.               }
  28.             }
  29.           }
  31.         } else {
  32.           Failure(new IllegalStateException("JDBCContex setting error: must set batch = true !"))
  33.         }
  34.       }
  35.     }

如果batch-update是某种Insert操作的话我们可以通过cox.batch注明返回由JDBC系统自动产生的唯一键。这些主键一般在构建表时注明，包括：serial, auto_increment等。如果不返回主键则返回update语句的更新状态如更新数据条数等。在上面这个函数里SQLBatchWithGeneratedKey.apply()返回insert数据主键，所以statement必须是INSERT语句。SQLBatch.apply()则用来运算update语句并返回更新数据的条数。下面是jdbcBatchUpdate函数的使用示范：

  1.  val insertSQL = "insert into members(name,birthday,description,created_at,picture) values (?, ?, ?, ?, ?)"
  2.   val dateCreated = DateTime.now
  4.   import java.io.FileInputStream
  6.   val picfile = new File("/users/tiger/Nobody.png")
  7.   val fis = new FileInputStream(picfile)
  9.   ctx = JDBCContext('h2)
  10.   try {
  11.     ctx = ctx.setBatchCommand(insertSQL).appendBatchParameters(
  12.       "John",new LocalDate("2008-03-01"),"youngest user",dateCreated,None).appendBatchParameters(
  13.       "peter", None, "no birth date", dateCreated, fis)
  14.       .appendBatchParameters(
  15.         "susan", None, "no birth date", dateCreated, None)
  16.       .setBatchReturnGeneratedKeyOption(JDBCContext.RETURN_GENERATED_KEYVALUE)
  17.   }
  18.   catch {
  19.     case e: Exception => println(e.getMessage)
  20.   }
  22.   var resultInserts = jdbcBatchUpdate(ctx)
  24.   resultInserts match {
  25.     case Success(msg) => println(msg)
  26.     case Failure(err) => println(s"${err.getMessage}")
  27.   }

上面这个例子里一个transaction批次包含了三条Insert语句，其中一条涉及存入picture字段：我们只需要把图像文件InputStream作为普通参数传人即可。我们也可以把任何类型的非batch-update语句捆绑在统一的transaction里运算，而且可以指定每条update返回类型：自动产生的主键或者更新数据条数：

  1. def jdbcTxUpdates[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(
  2.       implicit cbf: CanBuildFrom[Nothing, Long, C[Long]]): Try[C[Long]] = {
  3.       if (ctx.statements == Nil)
  4.         throw new IllegalStateException("JDBCContex setting error: statements empty!")
  5.       if (ctx.sqlType != SQL_UPDATE) {
  6.         Failure(new IllegalStateException("JDBCContex setting error: sqlType must be 'SQL_UPDATE'!"))
  7.       }
  8.       else {
  9.         if (!ctx.batch) {
  10.           if (ctx.statements.size == 1)
  11.             singleTxUpdate(ctx)
  12.           else
  13.             multiTxUpdates(ctx)
  14.         } else
  15.           Failure(new IllegalStateException("JDBCContex setting error: must set batch = false !"))
  17.       }
  18.     }

这个update函数又被细分为单条语句singleTxUpdate和多条语句multiTxUpdates。无论单条或多条update函数又被分为返回主键或更新状态类型的函数：

 

  1.  private def singleTxUpdateWithReturnKey[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(
  2.        implicit cbf: CanBuildFrom[Nothing, Long, C[Long]]): Try[C[Long]] = {
  3.        val Some(key) :: xs = ctx.returnGeneratedKey
  4.        val params: Seq[Any] = ctx.parameters match {
  5.          case Nil => Nil
  6.          case p@_ => p.head
  7.        }
  8.        val usql = new SQLUpdateWithGeneratedKey(ctx.statements.head, params, ctx.queryTags)(key)
  9.        Try {
  10.          NamedDB(ctx.dbName) localTx { implicit session =>
  11.            session.fetchSize(ctx.fetchSize)
  12.            ctx.queryTimeout.foreach(session.queryTimeout(_))
  13.            val result = usql.apply()
  14.            Seq(result).to[C]
  15.          }
  16.        }
  17.      }
  19.       private def singleTxUpdateNoReturnKey[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(
  20.         implicit cbf: CanBuildFrom[Nothing, Long, C[Long]]): Try[C[Long]] = {
  21.       val params: Seq[Any] = ctx.parameters match {
  22.         case Nil => Nil
  23.         case p@_ => p.head
  24.       }
  25.       val before = ctx.preAction match {
  26.         case None => pstm: PreparedStatement => {}
  27.         case Some(f) => f
  28.       }
  29.       val after = ctx.postAction match {
  30.         case None => pstm: PreparedStatement => {}
  31.         case Some(f) => f
  32.       }
  33.       val usql = new SQLUpdate(ctx.statements.head,params,ctx.queryTags)(before)(after)
  34.       Try {
  35.         NamedDB(ctx.dbName) localTx {implicit session =>
  36.           session.fetchSize(ctx.fetchSize)
  37.           ctx.queryTimeout.foreach(session.queryTimeout(_))
  38.           val result = usql.apply()
  39.           Seq(result.toLong).to[C]
  40.         }
  41.       }
  43.     }
  45.     private def singleTxUpdate[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(
  46.       implicit cbf: CanBuildFrom[Nothing, Long, C[Long]]): Try[C[Long]] = {
  47.       if (noReturnKey(ctx))
  48.         singleTxUpdateNoReturnKey(ctx)
  49.       else
  50.         singleTxUpdateWithReturnKey(ctx)
  51.     }
  53.     private def noReturnKey(ctx: JDBCContext): Boolean = {
  54.       if (ctx.returnGeneratedKey != Nil) {
  55.         val k :: xs = ctx.returnGeneratedKey
  56.          k match {
  57.           case None => true
  58.           case Some(k) => false
  59.         }
  60.       } else true
  61.     }
  63.     def noActon: PreparedStatement=>Unit = pstm => {}
  65.     def multiTxUpdates[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(
  66.       implicit cbf: CanBuildFrom[Nothing, Long, C[Long]]): Try[C[Long]] = {
  67.         Try {
  68.           NamedDB(ctx.dbName) localTx { implicit session =>
  69.             session.fetchSize(ctx.fetchSize)
  70.             ctx.queryTimeout.foreach(session.queryTimeout(_))
  71.             val keys: Seq[Option[Any]] = ctx.returnGeneratedKey match {
  72.               case Nil => Seq.fill(ctx.statements.size)(None)
  73.               case k@_ => k
  74.             }
  75.             val sqlcmd = ctx.statements zip ctx.parameters zip keys
  76.             val results = sqlcmd.map { case ((stm, param), key) =>
  77.               key match {
  78.                 case None =>
  79.                   new SQLUpdate(stm, param, Nil)(noActon)(noActon).apply().toLong
  80.                 case Some(k) =>
  81.                   new SQLUpdateWithGeneratedKey(stm, param, Nil)(k).apply().toLong
  82.               }
  83.             }
  84.             results.to[C]
  85.           }
  86.         }
  87.      }

下面是这个函数的使用示范： 

  1.  val updateSQL = "update members set description = ? where id < ?"
  2.   ctx = JDBCContext('h2)
  3.   try {
  4.      ctx = ctx.setUpdateCommand(JDBCContext.RETURN_GENERATED_KEYVALUE,insertSQL,
  5.        "max", None, "no birth date", dateCreated, None)
  6.        .appendUpdateCommand(JDBCContext.RETURN_UPDATED_COUNT, updateSQL, "id++", 10)
  7.       .appendUpdateCommand(JDBCContext.RETURN_UPDATED_COUNT,"delete members where id = 1")
  8.   }
  9.   catch {
  10.     case e: Exception => println(e.getMessage)
  11.   }
  12.   var resultUpdates = jdbcTxUpdates[Vector](ctx)
  14.   resultUpdates match {
  15.     case Success(msg) => println(msg)
  16.     case Failure(err) => println(s"${err.getMessage}")
  17.   }

在这个例子里我们把insert，update和delete混在了一个transaction里。最后，我们再把试验数据，包括blob字段读出来：

  1.   //data model
  2.   case class Member(
  3.                      id: Long,
  4.                      name: String,
  5.                      description: Option[String] = None,
  6.                      birthday: Option[LocalDate] = None,
  7.                      createdAt: DateTime,
  8.                      picture: InputStream)
  10.   //data row converter
  11.   val toMember = (rs: WrappedResultSet) => Member(
  12.     id = rs.long("id"),
  13.     name = rs.string("name"),
  14.     description = rs.stringOpt("description"),
  15.     birthday = rs.jodaLocalDateOpt("birthday"),
  16.     createdAt = rs.jodaDateTime("created_at"),
  17.     picture = rs.binaryStream("picture")
  18.   )
  20.   ctx = JDBCContext('h2)
  21.   ctx = ctx.setQueryCommand("select * from members").setQueryTimeout(Some(1000))
  23.   val vecMember: Vector[Member] = jdbcQueryResult[Vector,Member](ctx,toMember)
  25.   val buffer = new Array[Byte](1024)
  27.   vecMember.foreach {row =>
  28.     println(s"id: ${row.id} name: ${row.name}")
  29.     println(s"name: ${row.name}")
  30.     if (row.picture == null)
  31.       println("picture empty")
  32.     else {
  33.       val fname = s"/users/tiger/pic${row.id}.png"
  34.       val file = new File(fname)
  35.       val output = new FileOutputStream(file)
  37.       println(s"saving picture to $fname")
  39.       row.picture.available()
  40.       while (row.picture.read(buffer) > 0) {
  41.         output.write(buffer)
  42.       }
  44.       output.close()
  46.     }
  47.   }

下面是本次讨论的示范源代码：

build.sbt

  1. name := "learn-scalikeJDBC"
  3. version := "0.1"
  5. scalaVersion := "2.12.4"
  7. // Scala 2.10, 2.11, 2.12
  8. libraryDependencies ++= Seq(
  9.   "org.scalikejdbc" %% "scalikejdbc"       % "3.1.0",
  10.   "org.scalikejdbc" %% "scalikejdbc-test"   % "3.1.0"   % "test",
  11.   "org.scalikejdbc" %% "scalikejdbc-config"  % "3.1.0",
  12.   "com.h2database"  %  "h2"                % "1.4.196",
  13.   "mysql" % "mysql-connector-java" % "6.0.6",
  14.   "org.postgresql" % "postgresql" % "42.2.0",
  15.   "commons-dbcp" % "commons-dbcp" % "1.4",
  16.   "org.apache.tomcat" % "tomcat-jdbc" % "9.0.2",
  17.   "com.zaxxer" % "HikariCP" % "2.7.4",
  18.   "com.jolbox" % "bonecp" % "0.8.0.RELEASE",
  19.   "com.typesafe.slick" %% "slick" % "3.2.1",
  20.   "ch.qos.logback"  %  "logback-classic"   % "1.2.3"
  21. )

resources/application.conf

 

  1. # JDBC settings
  2. test {
  3.   db {
  4.     h2 {
  5.       driver = "org.h2.Driver"
  6.       url = "jdbc:h2:tcp://localhost/~/slickdemo"
  7.       user = ""
  8.       password = ""
  9.       poolInitialSize = 5
  10.       poolMaxSize = 7
  11.       poolConnectionTimeoutMillis = 1000
  12.       poolValidationQuery = "select 1 as one"
  13.       poolFactoryName = "commons-dbcp2"
  14.     }
  15.   }
  17.   db.mysql.driver = "com.mysql.cj.jdbc.Driver"
  18.   db.mysql.url = "jdbc:mysql://localhost:3306/testdb"
  19.   db.mysql.user = "root"
  20.   db.mysql.password = "123"
  21.   db.mysql.poolInitialSize = 5
  22.   db.mysql.poolMaxSize = 7
  23.   db.mysql.poolConnectionTimeoutMillis = 1000
  24.   db.mysql.poolValidationQuery = "select 1 as one"
  25.   db.mysql.poolFactoryName = "bonecp"
  27.   # scallikejdbc Global settings
  28.   scalikejdbc.global.loggingSQLAndTime.enabled = true
  29.   scalikejdbc.global.loggingSQLAndTime.logLevel = info
  30.   scalikejdbc.global.loggingSQLAndTime.warningEnabled = true
  31.   scalikejdbc.global.loggingSQLAndTime.warningThresholdMillis = 1000
  32.   scalikejdbc.global.loggingSQLAndTime.warningLogLevel = warn
  33.   scalikejdbc.global.loggingSQLAndTime.singleLineMode = false
  34.   scalikejdbc.global.loggingSQLAndTime.printUnprocessedStackTrace = false
  35.   scalikejdbc.global.loggingSQLAndTime.stackTraceDepth = 10
  36. }
  37. dev {
  38.   db {
  39.     h2 {
  40.       driver = "org.h2.Driver"
  41.       url = "jdbc:h2:tcp://localhost/~/slickdemo"
  42.       user = ""
  43.       password = ""
  44.       poolFactoryName = "hikaricp"
  45.       numThreads = 10
  46.       maxConnections = 12
  47.       minConnections = 4
  48.       keepAliveConnection = true
  49.     }
  50.     mysql {
  51.       driver = "com.mysql.cj.jdbc.Driver"
  52.       url = "jdbc:mysql://localhost:3306/testdb"
  53.       user = "root"
  54.       password = "123"
  55.       poolInitialSize = 5
  56.       poolMaxSize = 7
  57.       poolConnectionTimeoutMillis = 1000
  58.       poolValidationQuery = "select 1 as one"
  59.       poolFactoryName = "bonecp"
  61.     }
  62.     postgres {
  63.       driver = "org.postgresql.Driver"
  64.       url = "jdbc:postgresql://localhost:5432/testdb"
  65.       user = "root"
  66.       password = "123"
  67.       poolFactoryName = "hikaricp"
  68.       numThreads = 10
  69.       maxConnections = 12
  70.       minConnections = 4
  71.       keepAliveConnection = true
  72.     }
  73.   }
  74.   # scallikejdbc Global settings
  75.   scalikejdbc.global.loggingSQLAndTime.enabled = true
  76.   scalikejdbc.global.loggingSQLAndTime.logLevel = info
  77.   scalikejdbc.global.loggingSQLAndTime.warningEnabled = true
  78.   scalikejdbc.global.loggingSQLAndTime.warningThresholdMillis = 1000
  79.   scalikejdbc.global.loggingSQLAndTime.warningLogLevel = warn
  80.   scalikejdbc.global.loggingSQLAndTime.singleLineMode = false
  81.   scalikejdbc.global.loggingSQLAndTime.printUnprocessedStackTrace = false
  82.   scalikejdbc.global.loggingSQLAndTime.stackTraceDepth = 10
  83. }

JDBCEngine.scala

  1. package jdbccontext
  2. import java.sql.PreparedStatement
  4. import scala.collection.generic.CanBuildFrom
  5. import scalikejdbc._
  7. import scala.util._
  8. import scalikejdbc.TxBoundary.Try._
  10.   object JDBCContext {
  11.     type SQLTYPE = Int
  12.     val SQL_SELECT: Int = 0
  13.     val SQL_EXEDDL= 1
  14.     val SQL_UPDATE = 2
  15.     val RETURN_GENERATED_KEYVALUE = true
  16.     val RETURN_UPDATED_COUNT = false
  18.   }
  20.   case class JDBCContext(
  21.                           dbName: Symbol,
  22.                           statements: Seq[String] = Nil,
  23.                           parameters: Seq[Seq[Any]] = Nil,
  24.                           fetchSize: Int = 100,
  25.                           queryTimeout: Option[Int] = None,
  26.                           queryTags: Seq[String] = Nil,
  27.                           sqlType: JDBCContext.SQLTYPE = JDBCContext.SQL_SELECT,
  28.                           batch: Boolean = false,
  29.                           returnGeneratedKey: Seq[Option[Any]] = Nil,
  30.                           // no return: None, return by index: Some(1), by name: Some("id")
  31.                           preAction: Option[PreparedStatement => Unit] = None,
  32.                           postAction: Option[PreparedStatement => Unit] = None) {
  34.     ctx =>
  36.     //helper functions
  38.     def appendTag(tag: String): JDBCContext = ctx.copy(queryTags = ctx.queryTags :+ tag)
  40.     def appendTags(tags: Seq[String]): JDBCContext = ctx.copy(queryTags = ctx.queryTags ++ tags)
  42.     def setFetchSize(size: Int): JDBCContext = ctx.copy(fetchSize = size)
  44.     def setQueryTimeout(time: Option[Int]): JDBCContext = ctx.copy(queryTimeout = time)
  46.     def setPreAction(action: Option[PreparedStatement => Unit]): JDBCContext = {
  47.       if (ctx.sqlType == JDBCContext.SQL_UPDATE &&
  48.         !ctx.batch && ctx.statements.size == 1)
  49.         ctx.copy(preAction = action)
  50.       else
  51.         throw new IllegalStateException("JDBCContex setting error: preAction not supported!")
  52.     }
  54.     def setPostAction(action: Option[PreparedStatement => Unit]): JDBCContext = {
  55.       if (ctx.sqlType == JDBCContext.SQL_UPDATE &&
  56.         !ctx.batch && ctx.statements.size == 1)
  57.         ctx.copy(postAction = action)
  58.       else
  59.         throw new IllegalStateException("JDBCContex setting error: preAction not supported!")
  60.     }
  62.     def appendDDLCommand(_statement: String, _parameters: Any*): JDBCContext = {
  63.       if (ctx.sqlType == JDBCContext.SQL_EXEDDL) {
  64.         ctx.copy(
  65.           statements = ctx.statements ++ Seq(_statement),
  66.           parameters = ctx.parameters ++ Seq(Seq(_parameters))
  67.         )
  68.       } else
  69.         throw new IllegalStateException("JDBCContex setting error: option not supported!")
  70.     }
  72.     def appendUpdateCommand(_returnGeneratedKey: Boolean, _statement: String, _parameters: Any*): JDBCContext = {
  73.       if (ctx.sqlType == JDBCContext.SQL_UPDATE && !ctx.batch) {
  74.         ctx.copy(
  75.           statements = ctx.statements ++ Seq(_statement),
  76.           parameters = ctx.parameters ++ Seq(_parameters),
  77.           returnGeneratedKey = ctx.returnGeneratedKey ++ (if (_returnGeneratedKey) Seq(Some(1)) else Seq(None))
  78.         )
  79.       } else
  80.         throw new IllegalStateException("JDBCContex setting error: option not supported!")
  81.     }
  83.     def appendBatchParameters(_parameters: Any*): JDBCContext = {
  84.       if (ctx.sqlType != JDBCContext.SQL_UPDATE || !ctx.batch)
  85.         throw new IllegalStateException("JDBCContex setting error: batch parameters only supported for SQL_UPDATE and batch = true!")
  87.       var matchParams = true
  88.       if (ctx.parameters != Nil)
  89.         if (ctx.parameters.head.size != _parameters.size)
  90.           matchParams = false
  91.       if (matchParams) {
  92.         ctx.copy(
  93.           parameters = ctx.parameters ++ Seq(_parameters)
  94.         )
  95.       } else
  96.         throw new IllegalStateException("JDBCContex setting error: batch command parameters not match!")
  97.     }
  99.     def setBatchReturnGeneratedKeyOption(returnKey: Boolean): JDBCContext = {
  100.       if (ctx.sqlType != JDBCContext.SQL_UPDATE || !ctx.batch)
  101.          throw new IllegalStateException("JDBCContex setting error: only supported in batch update commands!")
  102.       ctx.copy(
  103.         returnGeneratedKey = if (returnKey) Seq(Some(1)) else Nil
  104.       )
  105.     }
  107.      def setQueryCommand(_statement: String, _parameters: Any*): JDBCContext = {
  108.         ctx.copy(
  109.           statements = Seq(_statement),
  110.           parameters = Seq(_parameters),
  111.           sqlType = JDBCContext.SQL_SELECT,
  112.           batch = false
  113.         )
  114.       }
  116.       def setDDLCommand(_statement: String, _parameters: Any*): JDBCContext = {
  117.         ctx.copy(
  118.           statements = Seq(_statement),
  119.           parameters = Seq(_parameters),
  120.           sqlType = JDBCContext.SQL_EXEDDL,
  121.           batch = false
  122.         )
  123.       }
  125.       def setUpdateCommand(_returnGeneratedKey: Boolean, _statement: String, _parameters: Any*): JDBCContext = {
  126.         ctx.copy(
  127.           statements = Seq(_statement),
  128.           parameters = Seq(_parameters),
  129.           returnGeneratedKey = if (_returnGeneratedKey) Seq(Some(1)) else Seq(None),
  130.           sqlType = JDBCContext.SQL_UPDATE,
  131.           batch = false
  132.         )
  133.       }
  134.       def setBatchCommand(_statement: String): JDBCContext = {
  135.         ctx.copy (
  136.           statements = Seq(_statement),
  137.           sqlType = JDBCContext.SQL_UPDATE,
  138.           batch = true
  139.         )
  140.       }
  141.   }
  143.   object JDBCEngine {
  145.     import JDBCContext._
  147.     private def noExtractor(message: String): WrappedResultSet => Nothing = { (rs: WrappedResultSet) =>
  148.       throw new IllegalStateException(message)
  149.     }
  151.     def jdbcQueryResult[C[_] <: TraversableOnce[_], A](
  152.          ctx: JDBCContext, rowConverter: WrappedResultSet => A)(
  153.           implicit cbf: CanBuildFrom[Nothing, A, C[A]]): C[A] = {
  155.       ctx.sqlType match {
  156.         case SQL_SELECT => {
  157.           val params: Seq[Any] = ctx.parameters match {
  158.             case Nil => Nil
  159.             case p@_ => p.head
  160.           }
  161.           val rawSql = new SQLToCollectionImpl[A, NoExtractor](ctx.statements.head, params)(noExtractor(""))
  162.           ctx.queryTimeout.foreach(rawSql.queryTimeout(_))
  163.           ctx.queryTags.foreach(rawSql.tags(_))
  164.           rawSql.fetchSize(ctx.fetchSize)
  165.           implicit val session = NamedAutoSession(ctx.dbName)
  166.           val sql: SQL[A, HasExtractor] = rawSql.map(rowConverter)
  167.           sql.collection.apply[C]()
  168.         }
  169.         case _ => throw new IllegalStateException("JDBCContex setting error: sqlType must be 'SQL_SELECT'!")
  170.       }
  171.     }
  173.     def jdbcExcuteDDL(ctx: JDBCContext): Try[String] = {
  174.        if (ctx.sqlType != SQL_EXEDDL) {
  175.         Failure(new IllegalStateException("JDBCContex setting error: sqlType must be 'SQL_EXEDDL'!"))
  176.       }
  177.       else {
  178.         NamedDB(ctx.dbName) localTx { implicit session =>
  179.           Try {
  180.                 ctx.statements.foreach { stm =>
  181.                   val ddl = new SQLExecution(statement = stm, parameters = Nil)(
  182.                     before = WrappedResultSet => {})(
  183.                     after = WrappedResultSet => {})
  185.                   ddl.apply()
  186.               }
  187.             "SQL_EXEDDL executed succesfully."
  188.           }
  189.         }
  190.       }
  191.     }
  193.     def jdbcBatchUpdate[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(
  194.       implicit cbf: CanBuildFrom[Nothing, Long, C[Long]]): Try[C[Long]] = {
  195.       if (ctx.statements == Nil)
  196.         throw new IllegalStateException("JDBCContex setting error: statements empty!")
  197.       if (ctx.sqlType != SQL_UPDATE) {
  198.         Failure(new IllegalStateException("JDBCContex setting error: sqlType must be 'SQL_UPDATE'!"))
  199.       }
  200.       else {
  201.         if (ctx.batch) {
  202.           if (noReturnKey(ctx)) {
  203.             val usql = SQL(ctx.statements.head)
  204.               .tags(ctx.queryTags: _*)
  205.               .batch(ctx.parameters: _*)
  206.             Try {
  207.               NamedDB(ctx.dbName) localTx { implicit session =>
  208.                 ctx.queryTimeout.foreach(session.queryTimeout(_))
  209.                 usql.apply[Seq]()
  210.                 Seq.empty[Long].to[C]
  211.               }
  212.             }
  213.           } else {
  214.             val usql = new SQLBatchWithGeneratedKey(ctx.statements.head, ctx.parameters, ctx.queryTags)(None)
  215.             Try {
  216.               NamedDB(ctx.dbName) localTx { implicit session =>
  217.                 ctx.queryTimeout.foreach(session.queryTimeout(_))
  218.                 usql.apply[C]()
  219.               }
  220.             }
  221.           }
  223.         } else {
  224.           Failure(new IllegalStateException("JDBCContex setting error: must set batch = true !"))
  225.         }
  226.       }
  227.     }
  228.      private def singleTxUpdateWithReturnKey[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(
  229.        implicit cbf: CanBuildFrom[Nothing, Long, C[Long]]): Try[C[Long]] = {
  230.        val Some(key) :: xs = ctx.returnGeneratedKey
  231.        val params: Seq[Any] = ctx.parameters match {
  232.          case Nil => Nil
  233.          case p@_ => p.head
  234.        }
  235.        val usql = new SQLUpdateWithGeneratedKey(ctx.statements.head, params, ctx.queryTags)(key)
  236.        Try {
  237.          NamedDB(ctx.dbName) localTx { implicit session =>
  238.            session.fetchSize(ctx.fetchSize)
  239.            ctx.queryTimeout.foreach(session.queryTimeout(_))
  240.            val result = usql.apply()
  241.            Seq(result).to[C]
  242.          }
  243.        }
  244.      }
  246.       private def singleTxUpdateNoReturnKey[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(
  247.         implicit cbf: CanBuildFrom[Nothing, Long, C[Long]]): Try[C[Long]] = {
  248.       val params: Seq[Any] = ctx.parameters match {
  249.         case Nil => Nil
  250.         case p@_ => p.head
  251.       }
  252.       val before = ctx.preAction match {
  253.         case None => pstm: PreparedStatement => {}
  254.         case Some(f) => f
  255.       }
  256.       val after = ctx.postAction match {
  257.         case None => pstm: PreparedStatement => {}
  258.         case Some(f) => f
  259.       }
  260.       val usql = new SQLUpdate(ctx.statements.head,params,ctx.queryTags)(before)(after)
  261.       Try {
  262.         NamedDB(ctx.dbName) localTx {implicit session =>
  263.           session.fetchSize(ctx.fetchSize)
  264.           ctx.queryTimeout.foreach(session.queryTimeout(_))
  265.           val result = usql.apply()
  266.           Seq(result.toLong).to[C]
  267.         }
  268.       }
  270.     }
  272.     private def singleTxUpdate[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(
  273.       implicit cbf: CanBuildFrom[Nothing, Long, C[Long]]): Try[C[Long]] = {
  274.       if (noReturnKey(ctx))
  275.         singleTxUpdateNoReturnKey(ctx)
  276.       else
  277.         singleTxUpdateWithReturnKey(ctx)
  278.     }
  280.     private def noReturnKey(ctx: JDBCContext): Boolean = {
  281.       if (ctx.returnGeneratedKey != Nil) {
  282.         val k :: xs = ctx.returnGeneratedKey
  283.          k match {
  284.           case None => true
  285.           case Some(k) => false
  286.         }
  287.       } else true
  288.     }
  290.     def noActon: PreparedStatement=>Unit = pstm => {}
  292.     def multiTxUpdates[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(
  293.       implicit cbf: CanBuildFrom[Nothing, Long, C[Long]]): Try[C[Long]] = {
  294.         Try {
  295.           NamedDB(ctx.dbName) localTx { implicit session =>
  296.             session.fetchSize(ctx.fetchSize)
  297.             ctx.queryTimeout.foreach(session.queryTimeout(_))
  298.             val keys: Seq[Option[Any]] = ctx.returnGeneratedKey match {
  299.               case Nil => Seq.fill(ctx.statements.size)(None)
  300.               case k@_ => k
  301.             }
  302.             val sqlcmd = ctx.statements zip ctx.parameters zip keys
  303.             val results = sqlcmd.map { case ((stm, param), key) =>
  304.               key match {
  305.                 case None =>
  306.                   new SQLUpdate(stm, param, Nil)(noActon)(noActon).apply().toLong
  307.                 case Some(k) =>
  308.                   new SQLUpdateWithGeneratedKey(stm, param, Nil)(k).apply().toLong
  309.               }
  310.             }
  311.             results.to[C]
  312.           }
  313.         }
  314.      }
  317.     def jdbcTxUpdates[C[_] <: TraversableOnce[_]](ctx: JDBCContext)(
  318.       implicit cbf: CanBuildFrom[Nothing, Long, C[Long]]): Try[C[Long]] = {
  319.       if (ctx.statements == Nil)
  320.         throw new IllegalStateException("JDBCContex setting error: statements empty!")
  321.       if (ctx.sqlType != SQL_UPDATE) {
  322.         Failure(new IllegalStateException("JDBCContex setting error: sqlType must be 'SQL_UPDATE'!"))
  323.       }
  324.       else {
  325.         if (!ctx.batch) {
  326.           if (ctx.statements.size == 1)
  327.             singleTxUpdate(ctx)
  328.           else
  329.             multiTxUpdates(ctx)
  330.         } else
  331.           Failure(new IllegalStateException("JDBCContex setting error: must set batch = false !"))
  333.       }
  334.     }
  336.   }

JDBCEngineDemo.scala

  1. import java.io.File
  2. import java.io.FileOutputStream
  3. import java.io.InputStream
  4. import jdbccontext._
  5. import configdbs._
  6. import org.joda.time._
  7. import scala.util._
  8. import JDBCEngine._
  10. import scalikejdbc._
  11. object CrudDemo extends App {
  12.   ConfigDBsWithEnv("dev").setup('h2)
  13.   ConfigDBsWithEnv("dev").loadGlobalSettings()
  15.   val dropSQL: String ="""
  16.       drop table members
  17.     """
  19.   val createSQL: String ="""
  20.     create table members (
  21.       id serial not null primary key,
  22.       name varchar(30) not null,
  23.       description varchar(1000),
  24.       birthday date,
  25.       created_at timestamp not null,
  26.       picture blob
  27.     )"""
  29.   var ctx = JDBCContext('h2)
  30.     try {
  31.       ctx = ctx.setDDLCommand(dropSQL)
  32.         .appendDDLCommand(createSQL)
  33.     }
  34.     catch {
  35.        case e: Exception => println(e.getMessage)
  36.     }
  38.   val resultCreateTable = jdbcExcuteDDL(ctx)
  40.   resultCreateTable match {
  41.     case Success(msg) => println(msg)
  42.     case Failure(err) => println(s"${err.getMessage}")
  43.   }
  45.   val insertSQL = "insert into members(name,birthday,description,created_at,picture) values (?, ?, ?, ?, ?)"
  46.   val dateCreated = DateTime.now
  48.   import java.io.FileInputStream
  50.   val picfile = new File("/users/tiger/Nobody.png")
  51.   val fis = new FileInputStream(picfile)
  53.   ctx = JDBCContext('h2)
  54.   try {
  55.     ctx = ctx.setBatchCommand(insertSQL).appendBatchParameters(
  56.       "John",new LocalDate("2008-03-01"),"youngest user",dateCreated,None).appendBatchParameters(
  57.       "peter", None, "no birth date", dateCreated, fis)
  58.       .appendBatchParameters(
  59.         "susan", None, "no birth date", dateCreated, None)
  60.       .setBatchReturnGeneratedKeyOption(JDBCContext.RETURN_GENERATED_KEYVALUE)
  61.   }
  62.   catch {
  63.     case e: Exception => println(e.getMessage)
  64.   }
  66.   var resultInserts = jdbcBatchUpdate(ctx)
  68.   resultInserts match {
  69.     case Success(msg) => println(msg)
  70.     case Failure(err) => println(s"${err.getMessage}")
  71.   }
  74.   val updateSQL = "update members set description = ? where id < ?"
  75.   ctx = JDBCContext('h2)
  76.   try {
  77.      ctx = ctx.setUpdateCommand(JDBCContext.RETURN_GENERATED_KEYVALUE,insertSQL,
  78.        "max", None, "no birth date", dateCreated, None)
  79.        .appendUpdateCommand(JDBCContext.RETURN_UPDATED_COUNT, updateSQL, "id++", 10)
  80.       .appendUpdateCommand(JDBCContext.RETURN_UPDATED_COUNT,"delete members where id = 1")
  81.   }
  82.   catch {
  83.     case e: Exception => println(e.getMessage)
  84.   }
  85.   var resultUpdates = jdbcTxUpdates[Vector](ctx)
  87.   resultUpdates match {
  88.     case Success(msg) => println(msg)
  89.     case Failure(err) => println(s"${err.getMessage}")
  90.   }
  93.   //data model
  94.   case class Member(
  95.                      id: Long,
  96.                      name: String,
  97.                      description: Option[String] = None,
  98.                      birthday: Option[LocalDate] = None,
  99.                      createdAt: DateTime,
  100.                      picture: InputStream)
  102.   //data row converter
  103.   val toMember = (rs: WrappedResultSet) => Member(
  104.     id = rs.long("id"),
  105.     name = rs.string("name"),
  106.     description = rs.stringOpt("description"),
  107.     birthday = rs.jodaLocalDateOpt("birthday"),
  108.     createdAt = rs.jodaDateTime("created_at"),
  109.     picture = rs.binaryStream("picture")
  110.   )
  112.   ctx = JDBCContext('h2)
  113.   ctx = ctx.setQueryCommand("select * from members").setQueryTimeout(Some(1000))
  115.   val vecMember: Vector[Member] = jdbcQueryResult[Vector,Member](ctx,toMember)
  117.   val buffer = new Array[Byte](1024)
  119.   vecMember.foreach {row =>
  120.     println(s"id: ${row.id} name: ${row.name}")
  121.     println(s"name: ${row.name}")
  122.     if (row.picture == null)
  123.       println("picture empty")
  124.     else {
  125.       val fname = s"/users/tiger/pic${row.id}.png"
  126.       val file = new File(fname)
  127.       val output = new FileOutputStream(file)
  129.       println(s"saving picture to $fname")
  131.       row.picture.available()
  132.       while (row.picture.read(buffer) > 0) {
  133.         output.write(buffer)
  134.       }
  136.       output.close()
  138.     }
  139.   }
  141. }

 

 

 

 

 

 

 

 

 

 

 

版权声明：本文为tiger-xc原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。
本文链接：https://www.cnblogs.com/tiger-xc/p/8431245.html