ライブラリ

Doobie

概要

DoobieはScala向けの関数型データベースアクセスライブラリで、「型安全性とコンポーザビリティの完璧な融合」を実現します。手動クエリ記述による型安全APIを特徴とし、クエリフラグメントの組み合わせと操作で大規模クエリ構築を可能にします。隠蔽なしで最適化された特殊クエリ記述が可能な自由度の高いライブラリとして、関数型プログラミング愛好者の定番選択肢となっています。

詳細

Doobie 2025年版は、Cats EffectとCatsエコシステムとの深い統合により、関数型プログラミングの原則に忠実な純関数的データベースアクセスを提供します。JDBCレイヤーの上に構築されながらも、型安全性を最大限に活用したConnectionIOモナドによる複雑なデータベース操作の組み合わせが可能。SQL記述の完全な制御権を開発者に委ね、最適化されたクエリを柔軟に構築できる設計思想を貫いています。

主な特徴

• 型安全なSQL構築: コンパイル時のSQL型チェックと安全なクエリ組み立て
• ConnectionIOモナド: 関数型効果システムによる安全なリソース管理
• フラグメント指向設計: 再利用可能なクエリパーツの組み合わせ
• Cats Effect統合: 非同期処理とエラーハンドリングの完全サポート
• カスタムマッピング: 柔軟な型変換とスキーママッピング機能
• テスト支援機能: クエリ検証とデバッグのための豊富なツール

メリット・デメリット

メリット

• 関数型プログラミングパラダイムとの完璧な統合と一貫性
• SQL記述の完全な制御による最適化クエリの実現
• 型安全性によるランタイムエラーの大幅な削減
• ConnectionIOモナドによる安全で構造化されたデータベース操作
• Cats Effectエコシステムによる豊富な関数型プリミティブ
• クエリフラグメントの再利用性と保守性の向上

デメリット

• 関数型プログラミングの知識が必須で学習コストが高い
• 従来のORM機能（自動テーブル生成等）は提供されない
• Cats/Cats Effectエコシステムへの依存による複雑性増加
• 手動SQL記述による開発速度の低下の可能性
• オブジェクト指向アプローチとの設計思想の違い
• 小規模プロジェクトでは過剰エンジニアリングの傾向

参考ページ

• Doobie 公式ドキュメント
• Doobie GitHub リポジトリ
• Typelevel プロジェクト

書き方の例

プロジェクトセットアップと依存関係

// build.sbt
libraryDependencies ++= Seq(
  // Core doobie依存関係
  "org.tpolecat" %% "doobie-core"      % "1.0.0-RC4",
  
  // データベースドライバー（必要に応じて）
  "org.tpolecat" %% "doobie-h2"        % "1.0.0-RC4",          // H2
  "org.tpolecat" %% "doobie-postgres"  % "1.0.0-RC4",          // PostgreSQL
  "org.tpolecat" %% "doobie-hikari"    % "1.0.0-RC4",          // HikariCP
  
  // テスト支援（オプション）
  "org.tpolecat" %% "doobie-scalatest" % "1.0.0-RC4" % "test"
)

// 基本インポート
import cats._, cats.data._, cats.implicits._
import cats.effect._
import doobie._
import doobie.implicits._

基本的な接続設定とTransactor

import cats.effect.IO
import doobie.util.ExecutionContexts
import cats.effect.unsafe.implicits.global

// PostgreSQL接続例
val xa = Transactor.fromDriverManager[IO](
  driver = "org.postgresql.Driver",
  url = "jdbc:postgresql://localhost:5432/mydb",
  user = "username",
  password = "password",
  logHandler = None // ログを有効にする場合は適切なハンドラーを設定
)

// H2インメモリデータベース接続例
val h2xa = Transactor.fromDriverManager[IO](
  driver = "org.h2.Driver",
  url = "jdbc:h2:mem:test;DB_CLOSE_DELAY=-1",
  user = "sa",
  password = "",
  logHandler = None
)

// HikariCP接続プール使用例
import com.zaxxer.hikari.HikariConfig
import com.zaxxer.hikari.HikariDataSource

val config = new HikariConfig()
config.setJdbcUrl("jdbc:postgresql://localhost:5432/mydb")
config.setUsername("username")
config.setPassword("password")
config.setMaximumPoolSize(10)

val dataSource = new HikariDataSource(config)
val pooledXa = Transactor.fromDataSource[IO](dataSource, ExecutionContexts.synchronous)

ケースクラスとクエリ定義

// データモデル定義
case class User(id: Int, name: String, email: String, age: Option[Int])
case class Post(id: Int, userId: Int, title: String, content: String)

// 基本的なクエリ関数
def findUserById(id: Int): ConnectionIO[Option[User]] =
  sql"SELECT id, name, email, age FROM users WHERE id = $id"
    .query[User]
    .option

def findUsersByName(name: String): ConnectionIO[List[User]] =
  sql"SELECT id, name, email, age FROM users WHERE name LIKE ${s"%$name%"}"
    .query[User]
    .to(List)

def getAllUsers: ConnectionIO[List[User]] =
  sql"SELECT id, name, email, age FROM users"
    .query[User]
    .to(List)

// プログラムの実行
val program: IO[Option[User]] = findUserById(1).transact(xa)
val result = program.unsafeRunSync()
println(result)

// 複数ユーザーの取得
val users: IO[List[User]] = getAllUsers.transact(xa)
users.unsafeRunSync().foreach(println)

データ挿入・更新・削除操作

// データ挿入
def insertUser(name: String, email: String, age: Option[Int]): ConnectionIO[Int] =
  sql"INSERT INTO users (name, email, age) VALUES ($name, $email, $age)"
    .update
    .run

// 生成されたキーを取得する挿入
def insertUserWithGeneratedId(name: String, email: String, age: Option[Int]): ConnectionIO[User] =
  sql"INSERT INTO users (name, email, age) VALUES ($name, $email, $age)"
    .update
    .withUniqueGeneratedKeys("id", "name", "email", "age")

// データ更新
def updateUserEmail(id: Int, newEmail: String): ConnectionIO[Int] =
  sql"UPDATE users SET email = $newEmail WHERE id = $id"
    .update
    .run

// データ削除
def deleteUser(id: Int): ConnectionIO[Int] =
  sql"DELETE FROM users WHERE id = $id"
    .update
    .run

// 実行例
val insertProgram = for {
  userId <- insertUserWithGeneratedId("田中太郎", "[email protected]", Some(30))
  _      <- updateUserEmail(userId.id, "[email protected]")
  user   <- findUserById(userId.id)
} yield user

val result = insertProgram.transact(xa).unsafeRunSync()
println(s"Updated user: $result")

フラグメント指向プログラミングとクエリ組み立て

import doobie.implicits._

// 再利用可能なクエリフラグメント
def baseUserQuery: Fragment = fr"SELECT id, name, email, age FROM users"

def whereIdEquals(id: Int): Fragment = fr"WHERE id = $id"

def whereNameLike(name: String): Fragment = fr"WHERE name LIKE ${s"%$name%"}"

def whereAgeGreaterThan(age: Int): Fragment = fr"WHERE age > $age"

def orderByName: Fragment = fr"ORDER BY name ASC"

def limitResults(limit: Int): Fragment = fr"LIMIT $limit"

// フラグメントを組み合わせた動的クエリ
def findUsersWithFilters(
    nameFilter: Option[String] = None,
    minAge: Option[Int] = None,
    limit: Option[Int] = None
): ConnectionIO[List[User]] = {
  
  val whereClause = List(
    nameFilter.map(name => fr"name LIKE ${s"%$name%"}"),
    minAge.map(age => fr"age > $age")
  ).flatten
  
  val query = baseUserQuery ++ 
    (if (whereClause.nonEmpty) fr"WHERE" ++ whereClause.reduce(_ ++ fr"AND" ++ _) else Fragment.empty) ++
    orderByName ++
    (limit.map(l => fr"LIMIT $l").getOrElse(Fragment.empty))
  
  query.query[User].to(List)
}

// 使用例
val dynamicQuery = findUsersWithFilters(
  nameFilter = Some("田中"),
  minAge = Some(25),
  limit = Some(10)
)

dynamicQuery.transact(xa).unsafeRunSync().foreach(println)

トランザクション処理とエラーハンドリング

import cats.implicits._

// 複雑なトランザクション例
def transferUserPosts(fromUserId: Int, toUserId: Int): ConnectionIO[Int] = for {
  // ユーザーの存在確認
  fromUser <- findUserById(fromUserId).flatMap {
    case Some(user) => user.pure[ConnectionIO]
    case None => 
      val error: Throwable = new RuntimeException(s"User $fromUserId not found")
      error.raiseError[ConnectionIO, User]
  }
  
  toUser <- findUserById(toUserId).flatMap {
    case Some(user) => user.pure[ConnectionIO]
    case None =>
      val error: Throwable = new RuntimeException(s"User $toUserId not found")
      error.raiseError[ConnectionIO, User]
  }
  
  // 投稿の移転
  postsUpdated <- sql"UPDATE posts SET user_id = $toUserId WHERE user_id = $fromUserId"
    .update
    .run
    
} yield postsUpdated

// エラーハンドリング付きプログラム
def safeTransferPosts(fromUserId: Int, toUserId: Int): IO[Either[String, Int]] = {
  transferUserPosts(fromUserId, toUserId)
    .transact(xa)
    .attempt
    .map {
      case Right(count) => Right(count)
      case Left(error) => Left(s"Transfer failed: ${error.getMessage}")
    }
}

// 実行例
val transferResult = safeTransferPosts(1, 2).unsafeRunSync()
transferResult match {
  case Right(count) => println(s"Successfully transferred $count posts")
  case Left(error) => println(s"Error: $error")
}

バッチ処理とストリーミング

import fs2._

// バッチ挿入
def insertUsers(users: List[User]): ConnectionIO[Int] = {
  val sql = "INSERT INTO users (name, email, age) VALUES (?, ?, ?)"
  Update[(String, String, Option[Int])](sql)
    .updateMany(users.map(u => (u.name, u.email, u.age)))
}

// ストリーミング処理
def streamAllUsers: Stream[ConnectionIO, User] =
  sql"SELECT id, name, email, age FROM users"
    .query[User]
    .stream

def processUsersInBatches: IO[Unit] = {
  streamAllUsers
    .chunkN(100) // 100件ずつ処理
    .evalMap { chunk =>
      val users = chunk.toList
      // 各バッチの処理ロジック
      IO.println(s"Processing batch of ${users.size} users")
    }
    .compile
    .drain
    .transact(xa)
}

// 大量データ処理例
def exportUsersToCSV: IO[Unit] = {
  streamAllUsers
    .map { user =>
      s"${user.id},${user.name},${user.email},${user.age.getOrElse("")}"
    }
    .intersperse("\n")
    .through(fs2.text.utf8.encode)
    .through(fs2.io.file.Files[IO].writeAll(fs2.io.file.Path("users.csv")))
    .compile
    .drain
    .transact(xa)
}

processUsersInBatches.unsafeRunSync()

高度な型マッピングとカスタム変換

import java.time.LocalDateTime
import doobie.Meta

// カスタム型のマッピング
case class UserId(value: Int) extends AnyVal
case class Email(value: String) extends AnyVal

// 暗黙のMetaインスタンス定義
implicit val userIdMeta: Meta[UserId] = Meta[Int].imap(UserId.apply)(_.value)
implicit val emailMeta: Meta[Email] = Meta[String].imap(Email.apply)(_.value)

// 強い型付けを使用したユーザーモデル
case class TypedUser(
  id: UserId,
  name: String,
  email: Email,
  createdAt: LocalDateTime
)

def findTypedUserById(id: UserId): ConnectionIO[Option[TypedUser]] =
  sql"SELECT id, name, email, created_at FROM users WHERE id = ${id.value}"
    .query[TypedUser]
    .option

// 列挙型のマッピング
sealed trait UserStatus
case object Active extends UserStatus
case object Inactive extends UserStatus
case object Suspended extends UserStatus

implicit val userStatusMeta: Meta[UserStatus] = 
  Meta[String].imap {
    case "active" => Active
    case "inactive" => Inactive
    case "suspended" => Suspended
    case other => throw new IllegalArgumentException(s"Unknown status: $other")
  } {
    case Active => "active"
    case Inactive => "inactive"
    case Suspended => "suspended"
  }

case class UserWithStatus(
  id: UserId,
  name: String,
  email: Email,
  status: UserStatus
)

def findActiveUsers: ConnectionIO[List[UserWithStatus]] =
  sql"SELECT id, name, email, status FROM users WHERE status = 'active'"
    .query[UserWithStatus]
    .to(List)

デバッグとテスト支援

import doobie.util.log.LogHandler

// SQLロギング設定
val logHandler = LogHandler.jdkLogHandler

val xaWithLogging = Transactor.fromDriverManager[IO](
  driver = "org.h2.Driver",
  url = "jdbc:h2:mem:test;DB_CLOSE_DELAY=-1",
  user = "sa",
  password = "",
  logHandler = Some(logHandler)
)

// YOLO モード（開発・デバッグ用）
val y = xa.yolo
import y._

// クエリの即座実行とテスト
findUserById(1).quick.unsafeRunSync()
getAllUsers.check.unsafeRunSync() // クエリの型チェック

// テスト用のヘルパー関数
def setupTestData: ConnectionIO[Unit] = for {
  _ <- sql"CREATE TABLE IF NOT EXISTS users (id SERIAL PRIMARY KEY, name VARCHAR(255), email VARCHAR(255), age INTEGER)".update.run
  _ <- sql"DELETE FROM users".update.run
  _ <- insertUser("テストユーザー1", "[email protected]", Some(25))
  _ <- insertUser("テストユーザー2", "[email protected]", Some(30))
  _ <- insertUser("テストユーザー3", "[email protected]", None)
} yield ()

// テスト実行
val testProgram = for {
  _ <- setupTestData
  users <- getAllUsers
  _ <- IO.println(s"Test users: ${users.length}")
} yield users

testProgram.transact(xa).unsafeRunSync()