我在上期讨论里已经成功的创建了一个简单的Slick项目,然后又尝试使用了一些最基本的功能。Slick是一个FRM(Functional Relational Mapper),是为fp编程提供的scala SQL Query集成环境,可以让编程人员在scala编程语言里用函数式编程模式来实现对数据库操作的编程。在这篇讨论里我想以函数式思考模式来加深了解Slick。我对fp编程模式印象最深的就是类型匹配:从参数类型和返回结果类型来了解函数功能。所以上面我所指的函数式思考方式主要是从Slick函数的类型匹配角度来分析函数所起的作用和具体使用方式。
我们先了解一下建表过程:
import slick.driver.H2Driver.api._
object slick201 {
//projection case classes 表列模版
case class Coffee(
id: Option[Long]
,name: String
,sup_ID: Int
,price: Double
,grade: Grade
,total: Int
)
case class Supplier(
id: Option[Int]
,name: String
,address: String
,website: Option[String]
)
//自定义字段
abstract class Grade(points: Int)
object Grade {
case object Premium extends Grade(2)
case object Quality extends Grade(1)
case object Bestbuy extends Grade(0)
def fromInt(p: Int) = p match {
case 2 => Premium
case 1 => Quality
case 0 => Bestbuy
}
def toInt(g: Grade) = g match {
case Premium => 2
case Quality => 1
case Bestbuy => 0
}
implicit val customColumn: BaseColumnType[Grade] =
MappedColumnType.base[Grade,Int](Grade.toInt, Grade.fromInt)
}
//schema 表行结构定义
class Coffees(tag: Tag) extends Table[Coffee](tag, "COFFEES") {
def id = column[Long]("COF_ID", O.AutoInc, O.PrimaryKey)
def name = column[String]("COF_NAME")
def price = column[Double]("COF_PRICE")
def supID = column[Int]("COF_SUP")
def grade = column[Grade]("COF_GRADE", O.Default(Grade.Bestbuy))
def total = column[Int]("COF_TOTAL", O.Default(0))
def * = (id.?,name,supID,price,grade,total) <> (Coffee.tupled, Coffee.unapply)
def supplier = foreignKey("SUP_FK",supID,suppliers)(_.id,onUpdate = ForeignKeyAction.Restrict, onDelete = ForeignKeyAction.Cascade)
def nameidx = index("NM_IX",name,unique = true)
}
val coffees = TableQuery[Coffees]
class Suppliers(tag: Tag) extends Table[Supplier](tag, "SUPPLIERS") {
def id = column[Int]("SUP_ID", O.PrimaryKey, O.AutoInc)
def name = column[String]("SUP_NAME")
def address = column[String]("SUP_ADDR", O.Default("-"))
def website = column[Option[String]]("SUP_WEB")
def * = (id.?, name, address, website) <> (Supplier.tupled, Supplier.unapply)
def addidx = index("ADDR_IX",(name,address),unique = true)
}
val suppliers = TableQuery[Suppliers]
}
我尽量把经常会遇到的情况如:定义字段、建索引、默认值、自定义字段等都作了尝试。coffees和suppliers代表了最终的数据表Query,def * 定义了这个Query的默认返回结果字段。
所有的定义都是围绕着表行(Table Row)结构进行的,包括:表属性及操作(Table member methods)、字段(Column)、字段属性(ColumnOptions)。表行定义操作方法基本都在slick.lifted.AbstractTable里、表属性定义在slick.model命名空间里、而大部分的帮助支持函数都在slick.lifted命名空间的其它对象里。
表行的实际类型如下:
abstract class Table[T](_tableTag: Tag, _schemaName: Option[String], _tableName: String) extends AbstractTable[T](_tableTag, _schemaName, _tableName) { table => ...}
/** The profile-independent superclass of all table row objects.
* @tparam T Row type for this table. Make sure it matches the type of your `*` projection. */
abstract class AbstractTable[T](val tableTag: Tag, val schemaName: Option[String], val tableName: String) extends Rep[T] {...}
上面我们使用了模版对应表行定义方式,所有列都能和模版case class对应。那么在定义projection def * 时就需要使用<>函数:
def <>[R : ClassTag](f: (U => R), g: (R => Option[U])) = new MappedProjection[R, U](shape.toNode(value), MappedScalaType.Mapper(g.andThen(_.get).asInstanceOf[Any => Any], f.asInstanceOf[Any => Any], None), implicitly[ClassTag[R]])
f,g是两个case class <> Tuple转换函数。在上面的例子里我们提供的是tupled和unapply,效果就是这样的:
Coffee.tupled
//res2: ((Option[Long], String, Int, Double, Grade, Int)) => Coffee = <function1>
Coffee.unapply _
//res3: Coffee => Option[(Option[Long], String, Int, Double, Grade, Int)] = <function1>
res2 >>> 把tuple: (...)转成coffee,res2 >>> 把coffee转成Option[(...)]
TableQuery[T]继承了Query[T]:slick.lifted.Query.scala
/** Represents a database table. Profiles add extension methods to TableQuery
* for operations that can be performed on tables but not on arbitrary
* queries, e.g. getting the table DDL. */
class TableQuery[E <: AbstractTable[_]](cons: Tag => E) extends Query[E, E#TableElementType, Seq] {...}
...
sealed trait QueryBase[T] extends Rep[T]
/** An instance of Query represents a query or view, i.e. a computation of a
* collection type (Rep[Seq[T]]). It is parameterized with both, the mixed
* type (the type of values you see e.g. when you call map()) and the unpacked
* type (the type of values that you get back when you run the query).
*
* Additional extension methods for queries containing a single column are
* defined in [[slick.lifted.SingleColumnQueryExtensionMethods]].
*/
sealed abstract class Query[+E, U, C[_]] extends QueryBase[C[U]] { self =>...}
好了,来到了Query才算真正进入主题。Query可以说是Slick最核心的类型了。所有针对数据库的读写操作都是通过Query产生SQL语句发送到数据库实现的。Query是个函数式类型,即高阶类型Query[A]。A代表生成SQL语句的元素,通过转变A可以实现不同的SQL语句构建。不同功能的Query包括读取(retreive)、插入(insert)、更新(update)、删除(delete)都是通过Query变形(transformation)实现的。所有Query操作函数的款式:Query[A] => Query[B],是典型的函数式编程方式,也是scala集合操作函数款式。我们先从数据读取Query开始,因为上面我们曾经提到过可以通过map来决定新的结果集结构(projection):
val q1 = coffees.result
q1.statements.head
//res0: String = select "COF_ID", "COF_NAME", "COF_SUP", "COF_PRICE", "COF_GRADE", "COF_TOTAL" from "COFFEES"
val q2 = coffees.map(r => (r.id, r.name)).result
q2.statements.head
//res1: String = select "COF_ID", "COF_NAME" from "COFFEES"
val q3 = (for (c <- coffees) yield(c.id,c.name)).result
q3.statements.head
//res2: String = select "COF_ID", "COF_NAME" from "COFFEES"
map[A,B](Q[A])(A=>B]):Q[B], flatMap[A,B](Q[A])(A => Q[B]):Q[B]
所以不同的SQL语句基本上是通过Query[A] => Query[B]这种对高阶类型内嵌元素进行转变的函数式操作方式实现的。下面是一个带筛选条件的Query:
val q = coffees.filter(_.price > 100.0).map(r => (r.id,r.name)).result
q.statements.head
//res3: String = select "COF_ID", "COF_NAME" from "COFFEES" where "COF_PRICE" > 100.0
val q4 = coffees.filter(_.price > 100.0).take(4).map(_.name).result
q4.statements.head
//res4: String = select "COF_NAME" from "COFFEES" where "COF_PRICE" > 100.0 limit 4
val q5 = coffees.sortBy(_.id.desc.nullsFirst).map(_.name).drop(3).result
q5.statements.head
//res5: String = select "COF_NAME" from "COFFEES" order by "COF_ID" desc nulls first limit -1 offset 3
再复杂一点的Query,比如说join两个表:
val q6 = for {
(c,s) <- coffees join suppliers on (_.supID === _.id)
} yield(c.id,c.name,s.name)
q6.result.statements.head
//res6: String = select x2."COF_ID", x2."COF_NAME", x3."SUP_NAME" from "COFFEES" x2, "SUPPLIERS" x3 where x2."COF_SUP" = x3."SUP_ID"
val q7 = for {
c <- coffees
s <- suppliers.filter(c.supID === _.id)
} yield(c.id,c.name,s.name)
q7.result.statements.head
//res7: String = select x2."COF_ID", x2."COF_NAME", x3."SUP_NAME" from "COFFEES" x2, "SUPPLIERS" x3 where x2."COF_SUP" = x3."SUP_ID"
还有汇总类型的Query:
coffees.map(_.price).max.result.statements.head
//res10: String = select max("COF_PRICE") from "COFFEES"
coffees.map(_.total).sum.result.statements.head
//res11: String = select sum("COF_TOTAL") from "COFFEES"
coffees.length.result.statements.head
//res12: String = select count(1) from "COFFEES"
coffees.filter(_.price > 100.0).exists.result.statements.head
//res13: String = select exists(select "COF_TOTAL", "COF_NAME", "COF_SUP", "COF_ID", "COF_PRICE", "COF_GRADE" from "COFFEES" where "COF_PRICE" > 100.0)
Query是个monad,它可以实现函数组合(functional composition)。如上所示:所有Query操作函数都是Query[A]=>Query[B]形式的。由于Query[A]里面的A类型是Rep[T]类型,是SQL语句组件类型。典型函数如flatMap的调用方式是:flatMap{a => MakeQuery(a ...)},可以看到下一个Query的构成可能依赖a值,而a的类型是表行或列定义。所以Query的函数组合就是SQL语句的组合,最终结果是产生目标SQL语句。
Slick处理数据的方式是通过组合相应的SQL语句后发送给数据库去运算的,相关SQL语句的产生当然是通过Query来实现的:
val qInsert = coffees += Coffee(Some(0),"American",101,56.0,Grade.Bestbuy,0)
qInsert.statements.head
//res10: String = insert into "COFFEES" ("COF_NAME","COF_SUP","COF_PRICE","COF_GRADE","COF_TOTAL") values (?,?,?,?,?)
val qInsert2 = coffees.map{r => (r.name, r.supID, r.price)} += ("Columbia",101,102.0)
qInsert2.statements.head
//res11: String = insert into "COFFEES" ("COF_NAME","COF_SUP","COF_PRICE") values (?,?,?)
val qInsert3 = (suppliers.map{r => (r.id,r.name)}).
returning(suppliers.map(_.id)) += (101,"The Coffee Co.,")
qInsert3.statements.head
//res12: String = insert into "SUPPLIERS" ("SUP_NAME") values (?)
val qDelete = coffees.filter(_.price === 0.0).delete
qDelete.statements.head
//res17: String = delete from "COFFEES" where "COFFEES"."COF_PRICE" = 0.0
val qUpdate = for (c <- coffees if (c.name === "American")) yield c.price
qUpdate.update(10.0).statements.head
//res18: String = update "COFFEES" set "COF_PRICE" = ? where "COFFEES"."COF_NAME" = 'American'
DBIOAction类型款式如下:
sealed trait DBIOAction[+R, +S <: NoStream, -E <: Effect] extends Dumpable {
...}
package object dbio {
/** Simplified type for a streaming [[DBIOAction]] without effect tracking */
type StreamingDBIO[+R, +T] = DBIOAction[R, Streaming[T], Effect.All]
/** Simplified type for a [[DBIOAction]] without streaming or effect tracking */
type DBIO[+R] = DBIOAction[R, NoStream, Effect.All]
val DBIO = DBIOAction
}
def result: DriverAction[R, S, Effect.Read] = {...}
def delete: DriverAction[Int, NoStream, Effect.Write] = {...}
def update(value: T): DriverAction[Int, NoStream, Effect.Write] = {...}
def += (value: U): DriverAction[SingleInsertResult, NoStream, Effect.Write] = {...}
/** Run an Action asynchronously and return the result as a Future. */
final def run[R](a: DBIOAction[R, NoStream, Nothing]): Future[R] = runInternal(a, false)
DBIOAction只是对数据库操作动作的描述,不是实际的读写,所以DBIOAction可以进行组合。所谓组合的意思实际上就是把几个动作连续起来。DBIOAction的函数组件除monad通用的map、flatMap、sequence等,还包括了andThen、zip等合并操作函数,andThen可以返回最后一个动作结果、zip在一个pair里返回两个动作的结果。因为DBIOAction是monad,所以for-comprehension应该是最灵活、最强大的组合方式了。我们来试试用上面Query产生的动作来进行一些组合示范:
val initSupAction = suppliers.schema.create andThen qInsert3
val createCoffeeAction = coffees.schema.create
val insertCoffeeAction = qInsert zip qInsert2
val initSupAndCoffee = for {
_ <- initSupAction
_ <- createCoffeeAction
(i1,i2) <- insertCoffeeAction
} yield (i1,i2)
我们可以任意组合这些操作步骤,因为它们的返回结果类型都是DBIOAction[R]:一个free monad。大多数时间这些动作都是按照一定的流程顺序组合的。可能有些时候下一个动作需要依赖上一个动作产生的结果,这个时候用for-comprehension是最适合的了:
//先选出所有ESPRESSO开头的coffee名称,然后逐个删除
val delESAction = (for {
ns <- coffees.filter(_.name.startsWith("ESPRESSO")).map(_.name).result
_ <- DBIO.seq(ns.map(n => coffees.filter(_.name === n).delete): _*)
} yield ()).transactionally
//delESAction: slick.dbio.DBIOAction[Unit,slick.dbio.NoStream,slick.dbio.Effect.Read ...
//对一个品种价格升10%
def raisePriceAction(i: Long, np: Double, pc: Double) =
(for(c <- coffees if (c.id === i)) yield c.price).update(np * pc)
//raisePriceAction: raisePriceAction[](val i: Long,val np: Double,val pc: Double) => slick.driver.H2Driver.DriverAction[Int,slick.dbio.NoStream,slick.dbio.Effect.Write]
//对所有价格<100的coffee加价
val updatePriceAction = (for {
ips <- coffees.filter(_.price < 100.0).map(r => (r.id, r.price)).result
_ <- DBIO.seq{ips.map { ip => raisePriceAction(ip._1, ip._2, 110.0)}: _* }
} yield()).transactionally
//updatePriceAction: slick.dbio.DBIOAction[Unit,slick.dbio.NoStream,slick.dbio.Effect.Read ...
DBIO.successful(Supplier(Some(102),"Coffee Company","",None))
//res19: slick.dbio.DBIOAction[Supplier,slick.dbio.NoStream,slick.dbio.Effect] = SuccessAction(Supplier(Some(102),Coffee Company,,None))
DBIO.failed(new Exception("oh my god..."))
//res20: slick.dbio.DBIOAction[Nothing,slick.dbio.NoStream,slick.dbio.Effect] = FailureAction(java.lang.Exception: oh my god...)
//主要示范事后处理机制,不必理会功能的具体目的是否有任何意义
qInsert.andFinally(qDelete)
//res21: slick.dbio.DBIOAction[Int,slick.dbio.NoStream,slick.dbio.Effect.Write with slick.dbio.Effect.Write] = slick.dbio.SynchronousDatabaseAction$$anon$6@1d46b337
updatePriceAction.cleanUp (
{ case Some(e) => initSupAction; DBIO.failed(new Exception("oh my..."))
case _ => qInsert3
}
,true
)
//res22: slick.dbio.DBIOAction[Unit,slick.dbio.NoStream,slick.dbio.Effect.Read ...
raisePriceAction(101,10.0,110.0).asTry
//res23: slick.dbio.DBIOAction[scala.util.Try[Int],slick.dbio.NoStream,slick.dbio.Effect.Write] = slick.dbio.SynchronousDatabaseAction$$anon$9@60304a44
/** Run an Action asynchronously and return the result as a Future. */
final def run[R](a: DBIOAction[R, NoStream, Nothing]): Future[R] = runInternal(a, false)
import slick.jdbc.meta.MTable
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration.Duration
import scala.concurrent.{Await, Future}
import scala.util.{Success,Failure}
val db = Database.forURL("jdbc:h2:mem:test1;DB_CLOSE_DELAY=-1", driver="org.h2.Driver")
def recreateCoffeeTable: Future[Unit] = {
db.run(MTable.getTables("Coffees")).flatMap {
case tables if tables.isEmpty => db.run(coffees.schema.create).andThen {
case Success(_) => println("coffee table created")
case Failure(e) => println(s"failed to create! ${e.getMessage}")
}
case _ => db.run((coffees.schema.drop andThen coffees.schema.create)).andThen {
case Success(_) => println("coffee table recreated")
case Failure(e) => println(s"failed to recreate! ${e.getMessage}")
}
}
}
import slick.driver.H2Driver.api._
object slick201 {
//projection case classes 表列模版
case class Coffee(
id: Option[Long]
,name: String
,sup_ID: Int
,price: Double
,grade: Grade
,total: Int
)
case class Supplier(
id: Option[Int]
,name: String
,address: String
,website: Option[String]
)
//自定义字段
abstract class Grade(points: Int)
object Grade {
case object Premium extends Grade(2)
case object Quality extends Grade(1)
case object Bestbuy extends Grade(0)
def fromInt(p: Int) = p match {
case 2 => Premium
case 1 => Quality
case 0 => Bestbuy
}
def toInt(g: Grade) = g match {
case Premium => 2
case Quality => 1
case Bestbuy => 0
}
implicit val customColumn: BaseColumnType[Grade] =
MappedColumnType.base[Grade,Int](Grade.toInt, Grade.fromInt)
}
//schema 表行结构定义
class Coffees(tag: Tag) extends Table[Coffee](tag, "COFFEES") {
def id = column[Long]("COF_ID", O.AutoInc, O.PrimaryKey)
def name = column[String]("COF_NAME")
def price = column[Double]("COF_PRICE")
def supID = column[Int]("COF_SUP")
def grade = column[Grade]("COF_GRADE", O.Default(Grade.Bestbuy))
def total = column[Int]("COF_TOTAL", O.Default(0))
def * = (id.?,name,supID,price,grade,total) <> (Coffee.tupled, Coffee.unapply)
def supplier = foreignKey("SUP_FK",supID,suppliers)(_.id,onUpdate = ForeignKeyAction.Restrict, onDelete = ForeignKeyAction.Cascade)
def nameidx = index("NM_IX",name,unique = true)
}
val coffees = TableQuery[Coffees]
class Suppliers(tag: Tag) extends Table[Supplier](tag, "SUPPLIERS") {
def id = column[Int]("SUP_ID", O.PrimaryKey, O.AutoInc)
def name = column[String]("SUP_NAME")
def address = column[String]("SUP_ADDR", O.Default("-"))
def website = column[Option[String]]("SUP_WEB")
def * = (id.?, name, address, website) <> (Supplier.tupled, Supplier.unapply)
def addidx = index("ADDR_IX",(name,address),unique = true)
}
val suppliers = TableQuery[Suppliers]
class Bars(tag: Tag) extends Table[(Int,String)](tag,"BARS") {
def id = column[Int]("BAR_ID",O.AutoInc,O.PrimaryKey)
def name = column[String]("BAR_NAME")
def * = (id, name)
}
val bars = TableQuery[Bars]
Coffee.tupled
//res2: ((Option[Long], String, Int, Double, Grade, Int)) => Coffee = <function1>
Coffee.unapply _
//res3: Coffee => Option[(Option[Long], String, Int, Double, Grade, Int)] = <function1>
val q1 = coffees.result
q1.statements.head
//res0: String = select "COF_ID", "COF_NAME", "COF_SUP", "COF_PRICE", "COF_GRADE", "COF_TOTAL" from "COFFEES"
val q2 = coffees.map(r => (r.id, r.name)).result
q2.statements.head
//res1: String = select "COF_ID", "COF_NAME" from "COFFEES"
val q3 = (for (c <- coffees) yield(c.id,c.name)).result
q3.statements.head
//res2: String = select "COF_ID", "COF_NAME" from "COFFEES"
val q = coffees.filter(_.price > 100.0).map(r => (r.id,r.name)).result
q.statements.head
//res3: String = select "COF_ID", "COF_NAME" from "COFFEES" where "COF_PRICE" > 100.0
val q4 = coffees.filter(_.price > 100.0).take(4).map(_.name).result
q4.statements.head
//res4: String = select "COF_NAME" from "COFFEES" where "COF_PRICE" > 100.0 limit 4
val q5 = coffees.sortBy(_.id.desc.nullsFirst).map(_.name).drop(3).result
q5.statements.head
//res5: String = select "COF_NAME" from "COFFEES" order by "COF_ID" desc nulls first limit -1 offset 3
val q6 = for {
(c,s) <- coffees join suppliers on (_.supID === _.id)
} yield(c.id,c.name,s.name)
q6.result.statements.head
//res6: String = select x2."COF_ID", x2."COF_NAME", x3."SUP_NAME" from "COFFEES" x2, "SUPPLIERS" x3 where x2."COF_SUP" = x3."SUP_ID"
val q7 = for {
c <- coffees
s <- suppliers.filter(c.supID === _.id)
} yield(c.id,c.name,s.name)
q7.result.statements.head
//res7: String = select x2."COF_ID", x2."COF_NAME", x3."SUP_NAME" from "COFFEES" x2, "SUPPLIERS" x3 where x2."COF_SUP" = x3."SUP_ID"
coffees.map(_.price).max.result.statements.head
//res10: String = select max("COF_PRICE") from "COFFEES"
coffees.map(_.total).sum.result.statements.head
//res11: String = select sum("COF_TOTAL") from "COFFEES"
coffees.length.result.statements.head
//res12: String = select count(1) from "COFFEES"
coffees.filter(_.price > 100.0).exists.result.statements.head
//res13: String = select exists(select "COF_TOTAL", "COF_NAME", "COF_SUP", "COF_ID", "COF_PRICE", "COF_GRADE" from "COFFEES" where "COF_PRICE" > 100.0)
val qInsert = coffees += Coffee(Some(0),"American",101,56.0,Grade.Bestbuy,0)
qInsert.statements.head
//res14: String = insert into "COFFEES" ("COF_NAME","COF_SUP","COF_PRICE","COF_GRADE","COF_TOTAL") values (?,?,?,?,?)
val qInsert2 = coffees.map{r => (r.name, r.supID, r.price)} += ("Columbia",101,102.0)
qInsert2.statements.head
//res15: String = insert into "COFFEES" ("COF_NAME","COF_SUP","COF_PRICE") values (?,?,?)
val qInsert3 = (suppliers.map{r => (r.id,r.name)}).
returning(suppliers.map(_.id)) += (101,"The Coffee Co.,")
qInsert3.statements.head
//res16: String = insert into "SUPPLIERS" ("SUP_NAME") values (?)
val qDelete = coffees.filter(_.price === 0.0).delete
qDelete.statements.head
//res17: String = delete from "COFFEES" where "COFFEES"."COF_PRICE" = 0.0
val qUpdate = for (c <- coffees if (c.name === "American")) yield c.price
qUpdate.update(10.0).statements.head
//res18: String = update "COFFEES" set "COF_PRICE" = ? where "COFFEES"."COF_NAME" = 'American'
val initSupAction = suppliers.schema.create andThen qInsert3
val createCoffeeAction = coffees.schema.create
val insertCoffeeAction = qInsert zip qInsert2
val initSupAndCoffee = for {
_ <- initSupAction
_ <- createCoffeeAction
(i1,i2) <- insertCoffeeAction
} yield (i1,i2)
//先选出所有ESPRESSO开头的coffee名称,然后逐个删除
val delESAction = (for {
ns <- coffees.filter(_.name.startsWith("ESPRESSO")).map(_.name).result
_ <- DBIO.seq(ns.map(n => coffees.filter(_.name === n).delete): _*)
} yield ()).transactionally
//delESAction: slick.dbio.DBIOAction[Unit,slick.dbio.NoStream,slick.dbio.Effect.Read with slick.dbio.Effect.Write with slick.dbio.Effect.Transactional] = CleanUpAction(AndThenAction(Vector(slick.driver.JdbcActionComponent$StartTransaction$@6e76c850, FlatMapAction(slick.driver.JdbcActionComponent$QueryActionExtensionMethodsImpl$$anon$1@2005bce5,<function1>,scala.concurrent.impl.ExecutionContextImpl@245036ad))),<function1>,true,slick.dbio.DBIOAction$sameThreadExecutionContext$@294c4c1d)
//对一个品种价格升10%
def raisePriceAction(i: Long, np: Double, pc: Double) =
(for(c <- coffees if (c.id === i)) yield c.price).update(np * pc)
//raisePriceAction: raisePriceAction[](val i: Long,val np: Double,val pc: Double) => slick.driver.H2Driver.DriverAction[Int,slick.dbio.NoStream,slick.dbio.Effect.Write]
//对所有价格<100的coffee加价
val updatePriceAction = (for {
ips <- coffees.filter(_.price < 100.0).map(r => (r.id, r.price)).result
_ <- DBIO.seq{ips.map { ip => raisePriceAction(ip._1, ip._2, 110.0)}: _* }
} yield()).transactionally
//updatePriceAction: slick.dbio.DBIOAction[Unit,slick.dbio.NoStream,slick.dbio.Effect.Read with slick.dbio.Effect.Write with slick.dbio.Effect.Transactional] = CleanUpAction(AndThenAction(Vector(slick.driver.JdbcActionComponent$StartTransaction$@6e76c850, FlatMapAction(slick.driver.JdbcActionComponent$QueryActionExtensionMethodsImpl$$anon$1@49c8a41f,<function1>,scala.concurrent.impl.ExecutionContextImpl@245036ad))),<function1>,true,slick.dbio.DBIOAction$sameThreadExecutionContext$@294c4c1d)
DBIO.successful(Supplier(Some(102),"Coffee Company","",None))
//res19: slick.dbio.DBIOAction[Supplier,slick.dbio.NoStream,slick.dbio.Effect] = SuccessAction(Supplier(Some(102),Coffee Company,,None))
DBIO.failed(new Exception("oh my god..."))
//res20: slick.dbio.DBIOAction[Nothing,slick.dbio.NoStream,slick.dbio.Effect] = FailureAction(java.lang.Exception: oh my god...)
//示范事后处理机制,不必理会功能的具体目的
qInsert.andFinally(qDelete)
//res21: slick.dbio.DBIOAction[Int,slick.dbio.NoStream,slick.dbio.Effect.Write with slick.dbio.Effect.Write] = slick.dbio.SynchronousDatabaseAction$$anon$6@1d46b337
updatePriceAction.cleanUp (
{ case Some(e) => initSupAction; DBIO.failed(new Exception("oh my..."))
case _ => qInsert3
}
,true
)
//res22: slick.dbio.DBIOAction[Unit,slick.dbio.NoStream,slick.dbio.Effect.Read with slick.dbio.Effect.Write with slick.dbio.Effect.Transactional with slick.dbio.Effect.Write] = CleanUpAction(CleanUpAction(AndThenAction(Vector(slick.driver.JdbcActionComponent$StartTransaction$@6e76c850, FlatMapAction(slick.driver.JdbcActionComponent$QueryActionExtensionMethodsImpl$$anon$1@1f7aad00,<function1>,scala.concurrent.impl.ExecutionContextImpl@245036ad))),<function1>,true,slick.dbio.DBIOAction$sameThreadExecutionContext$@294c4c1d),<function1>,true,scala.concurrent.impl.ExecutionContextImpl@245036ad)
raisePriceAction(101,10.0,110.0).asTry
//res23: slick.dbio.DBIOAction[scala.util.Try[Int],slick.dbio.NoStream,slick.dbio.Effect.Write] = slick.dbio.SynchronousDatabaseAction$$anon$9@60304a44
import slick.jdbc.meta.MTable
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration.Duration
import scala.concurrent.{Await, Future}
import scala.util.{Success,Failure}
val db = Database.forURL("jdbc:h2:mem:test1;DB_CLOSE_DELAY=-1", driver="org.h2.Driver")
def recreateCoffeeTable: Future[Unit] = {
db.run(MTable.getTables("Coffees")).flatMap {
case tables if tables.isEmpty => db.run(coffees.schema.create).andThen {
case Success(_) => println("coffee table created")
case Failure(e) => println(s"failed to create! ${e.getMessage}")
}
case _ => db.run((coffees.schema.drop andThen coffees.schema.create)).andThen {
case Success(_) => println("coffee table recreated")
case Failure(e) => println(s"failed to recreate! ${e.getMessage}")
}
}
}
}