标签: doobie

我试图抽象插入不同类型的对象到类似结构的SQL表.这是我正在尝试做的事情:

class TableAccess[A : Meta](table: String) {
  def insert(key: String, a: A): ConnectionIO[Unit] = {
    (fr"insert into " ++ Fragment.const(table) ++ fr" values ($key, $a);").update.run.map(_ => ())
  }
}

但我得到这个编译错误:

[error] diverging implicit expansion for type doobie.util.param.Param[A]
[error] starting with method fromMeta in object Param
[error]     (fr"insert into " ++ Fragment.const(table) ++ fr" values ($key, $a);").update.run.map(_ => ())

我在文档中找到的只有:

doobie允许您使用Meta实例插入任何类型的值(及其选项),其中包括...

但在这种情况下似乎还不够; 我需要什么样的类型类/进口/转换？

Doobie的书说,从存储库层返回ConnectionIO是一个很好的做法.它提供了链接调用并在一个事务中执行它们的能力.很好,很清楚.

现在让我们假设我们正在开发REST API服务,我们的场景是:

1. 在数据库中查找对象
2. 使用此对象执行一些异步操作(使用cats.effect.IO或monix.eval.Task).
3. 将对象存储在数据库中.

我们希望在1个事务中执行所有这些步骤.问题是没有自然变换,我们transactor.trans()在2 monad工作 - Task和ConnectionIO.那是不可能的.

问题是 - 如何将doobie ConnectionIO与1个组合中的任何效果monad 混合,例如我们在1个事务中工作并且能够在世界末尾提交/回滚所有数据库突变？

谢谢!

UPD:小例子

def getObject: ConnectionIO[Request]                      = ???
def saveObject(obj: Request): ConnectionIO[Request]       = ???
def processObject(obj: Request): monix.eval.Task[Request] = ???

val transaction:??? = for {
    obj       <- getObject             //ConnectionIO[Request]
    processed <- processObject(obj)    //monix.eval.Task[Request]
    updated   <- saveObject(processed) //ConnectionIO[Request]
  } yield updated

UPD2:@ oleg-pyzhcov提供的正确答案是将效果数据类型提升到ConnectionIO这样:

def getObject: ConnectionIO[Request]                      = ???
def saveObject(obj: Request): ConnectionIO[Request]       = ???
def processObject(obj: Request): monix.eval.Task[Request] …

如何在 Doobie 中读/写时间戳？

我有一个包含时间戳字段的记录类。当我尝试将其写入数据库或使用 doobie 读取时，出现错误Cannot find or construct a Read instance for type。

case class ExampleRecord(data: String, created_at: Timestamp)

val create = sql"create table if not exists example_ts (data TEXT NOT NULL, created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP)".update.run
val insert = Update[ExampleRecord]("insert into example_ts (data, created_at) values (?, ?)")
  .updateMany(List(
    ExampleRecord("one", Timestamp.valueOf(LocalDateTime.now())),
    ExampleRecord("two", Timestamp.valueOf(LocalDateTime.now()))
  ))
val select = sql"select data, created_at from example_ts".query[ExampleRecord].stream

val app = for {
  _ <- create.transact(xa).compile.drain
  _ <- insert.transact(xa).compile.drain
  _ <- …

我正在尝试将项目从 cats-effect 2 迁移到 cats-effect 3，我正在使用 doobie 与数据库交互。以前我可以ConnectionIO按照描述的IO那样升级，但是升级后我没有找到任何实现，如何使用 CE3 实现相同的效果？LiftIO[ConnectionIO]

我正在将以下 10 行 Python 代码移植到 Scala：

import psycopg2

def execute(user, password, database, host, port, *queries):
    connection = psycopg2.connect(user=user, password=password, host=host, port=port, database=database)
    cursor = connection.cursor()
    for sql in queries:
        print(sql)
        cursor.execute(sql)
    connection.commit()
    cursor.close()
    connection.close()

我有以下等效的 Scala 代码：

def execute(user: String, password: String, database: String, host: String, port: Int, queries: String*): Unit = {
  ???
}    

我想在针对数据库的单个事务中执行（并打印）一堆 SQL 语句（假设它是 Postgres）并完成。

我如何使用doobie做到这一点？

笔记：

1. 我无法将接口更改为我的execute()（包括我无法添加类型或隐式参数）。它必须接受字符串用户、密码等和一个可变参数，queries: String*从而保持与 Python 相同的接口。

2. 还请提及所有需要的进口

Doobie可以select *使用case类来方便且正确地传递参数，但是我看不到如何使用updateand 进行类似的工作insert。

例如，给定这样的案例类：

case class Course(
  sku: String,
  title: String,
  id: Id,
  price: Int,
  instructorid: Id,
  groupid: Id,
  shortdescription: String = "",
  transcript: String = "",
  project_home: String = "",
  repository: String = "",
  category: String = "",
  image: String = "",
  privacy: String = "",
  language: String = "",
  keywords: String = "",
  goals: String = "",
  instructionallevel: String = "",
  audience: String = "",
  studenttasks: String =  "",
  sections: String = …

我正在使用 doobie 查询一些数据，一切正常，如下所示：

case class Usuario(var documento: String, var nombre: String, var contrasena: String)

  def getUsuario(doc: String) =
     sql"""SELECT documento, nombre, contrasena FROM "Usuario" WHERE "documento" = $doc"""
      .query[Usuario]
      .option
      .transact(xa)
      .unsafeRunSync()

但是，如果我声明一个具有类型限制的函数，如下所示：

 def getOption[T](f: Fragment): Option[T] = {
    f.query[T]
     .option
     .transact(xa)
     .unsafeRunSync()

}

我收到了这些错误：

Error:(42, 12) Cannot find or construct a Read instance for type:
  T
This can happen for a few reasons, but the most common case is that a data
member somewhere within this type doesn't have …

我在 ZIO 应用程序中使用 Doobie，有时会出现死锁（应用程序完全冻结）。如果我仅在一个内核上运行我的应用程序，或者达到与数据库的最大并行连接数，就会发生这种情况。

我的代码看起来像：

def mkTransactor(cfg: DatabaseConfig): RManaged[Blocking, Transactor[Task]] =
    ZIO.runtime[Blocking].toManaged_.flatMap { implicit rt =>
      val connectEC = rt.platform.executor.asEC
      val transactEC = rt.environment.get.blockingExecutor.asEC

      HikariTransactor
        .fromHikariConfig[Task](
          hikari(cfg),
          connectEC,
          Blocker.liftExecutionContext(transactEC)
        )
        .toManaged
    }

  private def hikari(cfg: DatabaseConfig): HikariConfig = {
    val config = new com.zaxxer.hikari.HikariConfig

    config.setJdbcUrl(cfg.url)
    config.setSchema(cfg.schema)
    config.setUsername(cfg.user)
    config.setPassword(cfg.pass)

    config
  }

或者，我在 Hikari ( config.setLeakDetectionThreshold(10000L))上设置了泄漏检测参数，但出现泄漏错误不是由于处理数据库查询所花费的时间。

我正在使用 Doobie，在我发现的示例中，它使用unsafeRunSync，例如：

sql"select name from country"
  .query[String]    // Query0[String]
  .to[List]         // ConnectionIO[List[String]]
  .transact(xa)     // IO[List[String]]
  .unsafeRunSync    // List[String]
  .take(5)          // List[String]
  .foreach(println)

在底层，该函数的实现如下：

final def unsafeRunSync(): A = unsafeRunTimed(Duration.Inf).get

在文档中，我发现“请注意，此函数用于测试；它永远不应该出现在您的主线生产代码中！”。我想知道如果它unsafeRunSync在幕后使用这个功能，那么在生产中使用是否可以？

另外，如果不使用，如何设置执行超时unsafeRunTimed？

是否可以使用Doobie并行运行多个查询？

我有以下（伪）查询：

def prepareForQuery(input: String): ConnectionIO[Unit] = ???
val gettAllResults: ConnectionIO[List[(String, BigDecimal)]] = ???
def program(input : String) : ConnectionIO[List[(String, BigDecimal)]] = for{
    _ <- prepareForQuery(input)
    r <- gettAllResults
  } yield r

我尝试了以下内容：

import doobie._
import doobie.implicits._
import cats.implicits._
val xa = Transactor.fromDataSource[IO](myDataSource)
val result = (program(i1),program(i2)).parMapN{case (a,b) => a ++ b}
val rs = result.transact(xa).unsafeRunSync

但是，找不到的NonEmptyParallel实例ConnectionIO。

错误：（107，54）找不到参数p的隐式值：cats.NonEmptyParallel [doobie.ConnectionIO，F] val结果=（program（i1），program（i2））。parMapN {case（a，b）= > a ++ b}

我是否缺少明显的东西或尝试无法完成的事情？谢谢