sqlx（PostgreSQL と MySQL）

wakuwaku はリレーショナルデータベースと sqlx を通じて通信します。sqlx のコネクションプールは データソース SqlxSource<DB> になり、データソースは Db<S, C> ハンドルに格納されます。Db ハンドルがクエリを実行し、ケイパビリティ C がどのクエリを実行してよいかを決めます。クエリは普通の Rust の値です。構造体が、何に触れるか（Query）と sqlx 上でどう実行されるか（Execute）を記述します。

Query、Execute、エンティティ、ケイパビリティにまだ馴染みがない場合は、先に クエリとケイパビリティ を読んでください。このページでは sqlx に固有の内容を扱います。

バックエンドを有効にするリンクをコピー

sqlx 連携は wakuwaku::entity::sqlx にあり、wakuwaku クレートの 2 つのフィーチャーのどちらかで有効になります。

[dependencies]
wakuwaku = { version = "1.0.0-rc3", features = ["sqlx-pg"] }

どちらのフィーチャーも entity を有効にし、sqlx を wakuwaku::deps::sqlx として再エクスポートします。フィーチャーは加算的なので、両方のデータベースと通信するサービスは両方を有効にできます。

有効になっている sqlx のフィーチャーリンクをコピー

wakuwaku は sqlx 0.9 に default-features = false で依存し、自分が必要とするものだけを有効にします。

• runtime-tokio
• postgres（sqlx-pg の場合）または mysql（sqlx-mysql の場合）

それ以外はすべて無効です。特に wakuwaku::deps::sqlx には TLS、derive マクロやクエリマクロ、マイグレーション、追加のカラム型がありません。必要な場合は sqlx のフィーチャーを自分で有効にします。

これらのフィーチャーを wakuwaku 経由で有効にすることはできません。転送するフィーチャーがないためです。代わりに、同じ 0.9 のバージョン要件で sqlx を直接の依存関係として追加します。

[dependencies]
wakuwaku = { version = "1.0.0-rc3", features = ["sqlx-pg"] }
sqlx = { version = "0.9", default-features = false, features = ["tls-rustls", "macros", "migrate"] }

Cargo は semver 互換のクレートを 1 回だけビルドし、依存元が要求するフィーチャーを統合します。ビルドに含まれる sqlx は 1 つのままで、あなたが指定したフィーチャーが有効になり、wakuwaku::deps::sqlx::PgPool と sqlx::PgPool は同じ型です。マクロにも直接の依存関係が必要です。#[derive(FromRow)] や query! は ::sqlx:: で始まるパスに展開されるので、クレートに sqlx という名前の依存関係がないと解決できません。

このページの残りでは、wakuwaku::deps::sqlx に最初から含まれる機能だけを使います。

プールと Db ハンドルを作るリンクをコピー

いつも通り sqlx でプールを作り、SqlxSource::new で包み、Db::new でケイパビリティを与えます。

use wakuwaku::deps::sqlx::{self, postgres::PgPoolOptions};
use wakuwaku::entity::sqlx::backend::{PgSource, SqlxSource};
use wakuwaku::entity::{Db, ReadWrite};

pub async fn connect(url: &str) -> Result<Db<PgSource, ReadWrite>, sqlx::Error> {
    let pool = PgPoolOptions::new()
        .max_connections(10)
        .connect(url)
        .await?;
    Ok(Db::new(SqlxSource::new(pool)))
}

ハンドルについて知っておくべきこと:

• クエリごとにコネクションを取得します。 Db を通じて実行したクエリはプールからコネクションを 1 つ取り、オートコミットモードで実行されます。複数の文をアトミックに実行するには トランザクション を書きます。
• プールは非公開のままです。 SqlxSource はプールを外に返さないので、プールに生の SQL を流してケイパビリティを迂回することはできません。マイグレーションなどの準備は、プールを包む 前に 実行してください。
• ハンドルのクローンは安価です。 SqlxSource と Db は Clone で、クローンは同じプールを共有します。これは ServiceCreation::Dep が期待する性質です。ServiceBuilder を参照してください。

前の節の追加フィーチャーを有効にすると、マイグレーション付きのセットアップは次のようになります。

use wakuwaku::deps::sqlx::postgres::PgPoolOptions;
use wakuwaku::entity::sqlx::backend::{PgSource, SqlxSource};
use wakuwaku::entity::{Db, ReadWrite};

pub async fn connect(url: &str) -> Result<Db<PgSource, ReadWrite>, Box<dyn std::error::Error>> {
    let pool = PgPoolOptions::new().connect(url).await?;
    // The pool is still yours here: run migrations and other setup first.
    sqlx::migrate!("./migrations").run(&pool).await?;
    Ok(Db::new(SqlxSource::new(pool)))
}

クエリを書くリンクをコピー

sqlx 向けのクエリは 2 つのトレイトを実装します。

• wakuwaku::entity の Query は エフェクト、つまりどのエンティティを読み書きするかを宣言します。
• Execute<SqlxSource<DB>> は実行方法を記述します。execute は &mut DB::Connection（&mut PgConnection または &mut MySqlConnection）を受け取ります。これは sqlx の executor なので、本体は普通の sqlx の呼び出しで書けます。type Error はたいてい sqlx::Error です。自分の判断で拒否することもあるクエリは独自のエラー型を使います。

use wakuwaku::deps::sqlx::{self, PgConnection, Postgres};
use wakuwaku::entity::sqlx::backend::SqlxSource;
use wakuwaku::entity::{Entity, Execute, Query, Read, Write};

/// The `users` table, as a type.
pub struct User;
impl Entity for User {}

/// A row of `users`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct UserRow {
    pub id: i64,
    pub name: String,
}

/// Look a user up by id.
pub struct FindUser {
    pub id: i64,
}

impl Query for FindUser {
    type Effect = Read<User>;
}

impl Execute<SqlxSource<Postgres>> for FindUser {
    type Output = Option<UserRow>;
    type Error = sqlx::Error;

    async fn execute(self, conn: &mut PgConnection) -> Result<Self::Output, sqlx::Error> {
        let row: Option<(i64, String)> =
            sqlx::query_as("SELECT id, name FROM users WHERE id = $1")
                .bind(self.id)
                .fetch_optional(conn)
                .await?;
        Ok(row.map(|(id, name)| UserRow { id, name }))
    }
}

/// Rename a user. Returns whether the user existed.
pub struct RenameUser {
    pub id: i64,
    pub name: String,
}

impl Query for RenameUser {
    type Effect = Write<User>;
}

impl Execute<SqlxSource<Postgres>> for RenameUser {
    type Output = bool;
    type Error = sqlx::Error;

    async fn execute(self, conn: &mut PgConnection) -> Result<bool, sqlx::Error> {
        let done = sqlx::query("UPDATE users SET name = $2 WHERE id = $1")
            .bind(self.id)
            .bind(self.name)
            .execute(conn)
            .await?;
        Ok(done.rows_affected() == 1)
    }
}

Query の部分は両方のデータベースで同じで、変わるのは Execute の部分（プレースホルダーとコネクション型）だけです。移植性のある SQL で書いたクエリは、両方のソースに対して Execute を実装できます。

ここで行をタプルにデコードしているのは、タプルが derive フィーチャーなしで sqlx の FromRow を実装しているためです。derive を有効にすれば #[derive(sqlx::FromRow)] struct UserRow { … } と書いて、sqlx::query_as::<_, UserRow>(…) を使えます。

execute の中の conn は、プールのコネクション（クエリを単独で実行する場合）か、開いているトランザクションのコネクション（トランザクションのステップが tx.run(query) で実行する場合）のどちらかです。同じクエリ型をどちらの場所でも使えます。トランザクションの中で実行するには、クエリのエラーが AsSqlxError を実装している必要があります。sqlx::Error は実装済みです。

クエリを実行するリンクをコピー

Db<S, C> は、S 上で実行でき、かつエフェクトを C がカバーするすべてのクエリ Q について kanau の Processor<Q> を実装します。クエリの実行は process の呼び出しで、エラーはクエリの Error、ここでは sqlx::Error です。

use crate::queries::{FindUser, RenameUser};
use wakuwaku::deps::kanau::processor::Processor;
use wakuwaku::deps::sqlx;
use wakuwaku::entity::sqlx::backend::PgSource;
use wakuwaku::entity::{Db, ReadWrite};

pub async fn rename(db: &Db<PgSource, ReadWrite>) -> Result<(), sqlx::Error> {
    if let Some(user) = db.process(FindUser { id: 1 }).await? {
        let renamed = db.process(RenameUser { id: user.id, name: "Ada".into() }).await?;
        assert!(renamed);
    }
    Ok(())
}

プライマリとレプリカリンクをコピー

書き込みはプライマリへ、読み取りはレプリカへ送る構成はよくあります。wakuwaku では、プールごとに別のハンドルと別のケイパビリティを与えます。

• プライマリ上の Db<PgSource, ReadWrite> はどんなクエリも実行できます。
• レプリカ上の Db<PgSource, ReadOnly> は、エフェクトが読み取りだけのクエリしか実行できません。

レプリカのハンドルに書き込みを送ると、データベースの実行時エラーではなくコンパイルエラーになります。

async fn on_replica(replica: &Db<PgSource, ReadOnly>) {
    // error[E0277]: the trait bound `ReadOnly: CanWrite<User>` is not satisfied
    let _ = replica.process(RenameUser { id: 1, name: "Ada".into() }).await;
}

コンポーネントはケイパビリティの名前を出すべきではありません。ハンドルについてジェネリックにし、実行するクエリを Processor 境界として列挙します。こうするとケイパビリティは、コンポーネントを具体的なハンドルに結び付ける場所で 1 回だけ検査されます。

use crate::queries::{FindUser, UserRow};
use wakuwaku::ServiceBuilder;
use wakuwaku::ServiceCreation;
use wakuwaku::deps::kanau::processor::Processor;
use wakuwaku::deps::sqlx::{self, postgres::PgPoolOptions};
use wakuwaku::entity::sqlx::backend::{PgSource, SqlxSource};
use wakuwaku::entity::{Db, ReadOnly, ReadWrite};

/// Reads user profiles. Generic over the handle: it only lists the queries it runs.
pub struct Profiles<D> {
    db: D,
}

impl<D> Profiles<D>
where
    D: Processor<FindUser, Output = Option<UserRow>, Error = sqlx::Error>,
{
    pub async fn display_name(&self, id: i64) -> Result<Option<String>, sqlx::Error> {
        Ok(self.db.process(FindUser { id }).await?.map(|user| user.name))
    }
}

/// Wired to the replica: the capability is checked here, once.
impl ServiceCreation for Profiles<Db<PgSource, ReadOnly>> {
    type Dep = Db<PgSource, ReadOnly>;

    fn create_service(db: Self::Dep) -> Self {
        Self { db }
    }
}

pub async fn build(primary_url: &str, replica_url: &str) -> Result<(), sqlx::Error> {
    let primary = PgPoolOptions::new().connect(primary_url).await?;
    let replica = PgPoolOptions::new().connect(replica_url).await?;

    let services = ServiceBuilder::new(Db::<PgSource, ReadWrite>::new(SqlxSource::new(primary)))
        .push(Db::<PgSource, ReadOnly>::new(SqlxSource::new(replica)));

    let replica_db = services.provide::<Db<PgSource, ReadOnly>, _>().clone();
    let profiles = Profiles::create_service(replica_db);
    let _ = profiles.display_name(1).await?;
    Ok(())
}

2 つのハンドルは型が異なるので、ServiceBuilder は両方を保持でき、provide はそれぞれを型で見つけます。実際のサービスではビルダーを grpc_service、cron_job、iggy_consumer に渡し、それらが create_service を呼び出します。複数のハンドルからコンポーネントを組み立てるなら compose_grpc_service、compose_cron_job、compose_iggy_consumer を使います。ServiceBuilder を参照してください。単体テストでは、Profiles は Processor<FindUser> を実装する任意のフェイクを受け取れるので、データベースはまったく不要です。テスト を参照してください。

一部のエンティティだけを書き込めるサービスは、ReadWrite の代わりに独自のケイパビリティ（たとえば「すべて読めるが、書けるのは Session だけ」）を定義できます。クエリとケイパビリティ を参照してください。

API リファレンスリンクをコピー

• SqlxSource、SqlxBackend
• Db、Query、Execute
• wakuwaku::deps