Rust | Axum と SQLx と PostgreSQL で REST API を作る

2022-10-012023-10-31

URLをコピーしました！

Axum と SQLx を使って、REST API の CRUD を実装してみました。

Axum と SQLx

axum-ddd-explicit-architecture

Todo アプリを作成しました。（ベタですが…笑）

ソースコードは Github に置いてあります。

Axum の実装例を見たい場合は todo-driver 、SQLx の実装例を見たい場合は todo-adapter を中心に見ていただくと良いかと思います！

あくまで一例として見てください…人(-ω-`*)

GitHub

GitHub – codemountains/axum-ddd-explicit-architecture: An example of web application (REST API) by u… An example of web application (REST API) by using Rust and Axum with DDD, Clean Architecture, Explicit Architecture. – codemountains/axum-ddd-explicit-architect…

Axum

Axum は、 tokio チーム製の Web サーバーを実装するためのクレートです。

Axum の特徴を簡単に挙げておきます。

非同期ランタイムに tokio が使われているため、tokio のバージョン管理から解放されること
マクロレスで実装されていること

SQLx

SQLx は、Rust で実装された非同期 SQL クレートです。

SQLx の特徴を簡単に挙げておきます。

非同期処理に対応
Diesel とは異なり、SQLx は ORM ではない
データベースに依存せず、PostgreSQL、MySQL、SQLite、および MSSQL のサポートしている
様々なランタイムで動作する（async-std、tokio、actix）

Axum

todo-driver/Cargo.toml

[dependencies]
todo-kernel = { path = "../todo-kernel" }
todo-app = { path = "../todo-app" }
todo-adapter = { path = "../todo-adapter" }
anyhow = "1.0.58"
axum = "0.5.13"
dotenv = "0.15.0"
http-body = "0.4.5"
serde = { version = "1.0", features = ["derive"] }
tracing = "0.1.35"
tracing-subscriber = { version = "0.3.15", features = ["env-filter"] }
tokio = { version = "1.20.0", features = ["full"] }
tower = "0.4.13"
tower-http = { version = "0.3.4", features = ["cors"] }
thiserror = "1.0.35"
validator = { version = "0.16.0", features = ["derive"] }

Router とサーバーの起動

Routerを生成し、routeに URL パスを定義いていきます。

また、それぞれの HTTP メソッドに該当する関数に実行するハンドラ関数を引数として渡しています。

そして、layerには DI コンテナとして振る舞うModulesをExtensionとして渡して、各ハンドラ関数から Modulesを呼び出せるようにします。

todo-driver/src/startup/mod.rs

pub async fn startup(modules: Arc<Modules>) {
    let hc_router = Router::new()
        .route("/", get(hc))
        .route("/postgres", get(hc_postgres));

    let todo_router = Router::new()
        .route("/", get(find_todo).post(create_todo))
        .route(
            "/:id",
            get(get_todo)
                .patch(update_todo)
                .put(upsert_todo)
                .delete(delete_todo),
        );

    let app = Router::new()
        .nest("/v1/hc", hc_router)
        .nest("/v1/todos", todo_router)
        .layer(Extension(modules));

    let addr = SocketAddr::from(init_addr());
    tracing::info!("Server listening on {}", addr);

    axum::Server::bind(&addr)
        .serve(app.into_make_service())
        .await
        .unwrap_or_else(|_| panic!("Server cannot launch."));
}

todo-driver/src/module/mod.rs

use std::sync::Arc;
use todo_adapter::modules::{RepositoriesModule, RepositoriesModuleExt};
use todo_adapter::persistence::postgres::Db;
use todo_adapter::repository::health_check::HealthCheckRepository;
use todo_app::usecase::health_check::HealthCheckUseCase;
use todo_app::usecase::todo::TodoUseCase;

pub struct Modules {
    health_check_use_case: HealthCheckUseCase,
    todo_use_case: TodoUseCase<RepositoriesModule>,
}

pub trait ModulesExt {
    type RepositoriesModule: RepositoriesModuleExt;

    fn health_check_use_case(&self) -> &HealthCheckUseCase;
    fn todo_use_case(&self) -> &TodoUseCase<Self::RepositoriesModule>;
}

impl ModulesExt for Modules {
    type RepositoriesModule = RepositoriesModule;

    fn health_check_use_case(&self) -> &HealthCheckUseCase {
        &self.health_check_use_case
    }

    fn todo_use_case(&self) -> &TodoUseCase<Self::RepositoriesModule> {
        &self.todo_use_case
    }
}

impl Modules {
    pub async fn new() -> Self {
        let db = Db::new().await;

        let repositories_module = Arc::new(RepositoriesModule::new(db.clone()));

        let health_check_use_case = HealthCheckUseCase::new(HealthCheckRepository::new(db));
        let todo_use_case = TodoUseCase::new(repositories_module.clone());

        Self {
            health_check_use_case,
            todo_use_case,
        }
    }
}

URLクエリパラメータを受け取る

GET /v1/todos で実行される関数では、Query(query): Query<TodoQuery>でクエリパラメータを受け取ることができます。

Extension(modules): Extension<Arc<Modules>>が引数として渡されているため、Modulesを利用することができます。

そこから、ユースケースに定義した関数を実行します。

todo-driver/src/routes/todo.rs

pub async fn find_todo(
    Query(query): Query<TodoQuery>,
    Extension(modules): Extension<Arc<Modules>>,
) -> Result<impl IntoResponse, impl IntoResponse> {
    let resp = modules.todo_use_case().find_todo(query.into()).await;

    match resp {
        Ok(tv_list) => match tv_list {
            Some(tv) => {
                let todos = tv.into_iter().map(|t| t.into()).collect();
                let json = JsonTodoList::new(todos);
                Ok((StatusCode::OK, Json(json)))
            }
            None => {
                let json = JsonTodoList::new(vec![]);
                Ok((StatusCode::OK, Json(json)))
            }
        },
        Err(err) => {
            error!("Unexpected error: {:?}", err);

            if err.to_string() == *"`statusCode` is invalid." {
                let json =
                    JsonErrorResponse::new("invalid_request".to_string(), vec![err.to_string()]);
                Err((StatusCode::BAD_REQUEST, Json(json)))
            } else {
                let json = JsonErrorResponse::new(
                    "server_error".to_string(),
                    vec!["INTERNAL SERVER ERROR".to_string()],
                );
                Err((StatusCode::INTERNAL_SERVER_ERROR, Json(json)))
            }
        }
    }
}

なお、構造体への変換に Serde の Deserialize が必要です。

todo-driver/src/model/todo.rs

#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct TodoQuery {
    pub status: Option<String>,
}

impl From<TodoQuery> for SearchTodoCondition {
    fn from(tq: TodoQuery) -> Self {
        Self {
            status_code: tq.status,
        }
    }
}

URL パスパラメータ・JSON リクエストの受け取る

PATCH /v1/todos/:id で実行される関数では、URLパスパラメータをPath(id): Pathで、JSON リクエストをValidatedRequest(source): ValidatedRequest<JsonUpdateTodoContents>でそれぞれ受け取ることができます。

バリデーションについては、後述します。

todo-driver/src/routes/todo.rs

pub async fn update_todo(
    Path(id): Path<String>,
    ValidatedRequest(source): ValidatedRequest<JsonUpdateTodoContents>,
    Extension(modules): Extension<Arc<Modules>>,
) -> Result<impl IntoResponse, impl IntoResponse> {
    match source.validate(id) {
        Ok(todo) => {
            let resp = modules.todo_use_case().update_todo(todo).await;

            resp.map(|tv| {
                info!("Updated todo {}", tv.id);
                let json: JsonTodo = tv.into();
                (StatusCode::OK, Json(json))
            })
            .map_err(|err| {
                error!("{:?}", err);

                if err.to_string() == *"`statusCode` is invalid." {
                    let json = JsonErrorResponse::new(
                        "invalid_request".to_string(),
                        vec![err.to_string()],
                    );
                    (StatusCode::BAD_REQUEST, Json(json))
                } else {
                    let json = JsonErrorResponse::new(
                        "server_error".to_string(),
                        vec!["INTERNAL SERVER ERROR".to_string()],
                    );
                    (StatusCode::INTERNAL_SERVER_ERROR, Json(json))
                }
            })
        }
        Err(errors) => {
            let json = JsonErrorResponse::new("invalid_request".to_string(), errors);
            Err((StatusCode::BAD_REQUEST, Json(json)))
        }
    }
}

バリデーション

バリデーションに関するヘルパー等は、context に定義しています。

todo-driver/src/context/errors.rs

use thiserror::Error;

#[derive(Debug, Error)]
pub enum AppError {
    #[error(transparent)]
    Validation(#[from] validator::ValidationErrors),
    #[error(transparent)]
    JsonRejection(#[from] axum::extract::rejection::JsonRejection),
}

todo-driver/src/context/validate.rs

#[derive(Debug)]
pub struct ValidatedRequest<T>(pub T);

todo-driver/src/context/request_helper.rs

use crate::context::errors::AppError;
use crate::context::validate::ValidatedRequest;
use axum::async_trait;
use axum::extract::{FromRequest, RequestParts};
use axum::{BoxError, Json};
use serde::de::DeserializeOwned;
use validator::Validate;

#[async_trait]
impl<T, B> FromRequest<B> for ValidatedRequest<T>
where
    T: DeserializeOwned + Validate,
    B: http_body::Body + Send,
    B::Data: Send,
    B::Error: Into<BoxError>,
{
    type Rejection = AppError;

    async fn from_request(req: &mut RequestParts<B>) -> Result<Self, Self::Rejection> {
        let Json(value) = Json::<T>::from_request(req).await?;
        value.validate()?;
        Ok(ValidatedRequest(value))
    }
}

validateで返ってきた結果をerrorsという配列に格納して、JSON で返却するためにJsonErrorResponseを定義しています。

型が異なる等で JSON リクエストを定義した構造体に変換する際にエラーが発生する場合、Json::::from_request(req).await?が先に実行されるため、独自に実装したバリデーションのエラーより先にAppError::JsonRejectionエラーが返ります。

todo-driver/src/context/response_helper.rs

use crate::context::errors::AppError;
use axum::http::StatusCode;
use axum::response::{IntoResponse, Response};
use axum::Json;
use serde::Serialize;
use tracing::log::error;

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct JsonErrorResponse {
    error_code: String,
    errors: Vec<String>,
}

impl JsonErrorResponse {
    pub(crate) fn new(error_code: String, errors: Vec<String>) -> Self {
        Self { error_code, errors }
    }
}

impl IntoResponse for AppError {
    fn into_response(self) -> Response {
        match self {
            AppError::Validation(validation_errors) => {
                error!("{:?}", validation_errors);

                let mut messages: Vec<String> = Vec::new();
                let errors = validation_errors.field_errors();
                for (_, v) in errors.into_iter() {
                    for validation_error in v {
                        if let Some(msg) = validation_error.clone().message {
                            messages.push(msg.to_string());
                        }
                    }
                }

                (
                    StatusCode::BAD_REQUEST,
                    Json(JsonErrorResponse::new(
                        "invalid_request".to_string(),
                        messages,
                    )),
                )
            }
            AppError::JsonRejection(rejection) => {
                error!("{:?}", rejection);

                let messages = vec![rejection.to_string()];
                (
                    StatusCode::BAD_REQUEST,
                    Json(JsonErrorResponse::new(
                        "invalid_request".to_string(),
                        messages,
                    )),
                )
            }
        }
        .into_response()
    }
}

validator クレートのマクロを使って、バリデーションを実装しています。

requiredは Option 型のみで使用できます。そのため、別の構造体に詰め替える処理の中でバリデーションを通過した正常値である前提でunwrapを行っています。

titleをOption<String>からStringに変更し、リクエストで null を送信した場合は、AppError::JsonRejectionエラーが返ることになります。

todo-driver/src/model/todo.rs

#[derive(Deserialize, Debug, Validate)]
#[serde(rename_all = "camelCase")]
pub struct JsonCreateTodo {
    #[validate(
        length(min = 1, message = "`title` is empty."),
        required(message = "`title` is null.")
    )]
    pub title: Option<String>,
    #[validate(required(message = "`description` is null."))]
    pub description: Option<String>,
}

impl From<JsonCreateTodo> for CreateTodo {
    fn from(jc: JsonCreateTodo) -> Self {
        CreateTodo {
            title: jc.title.unwrap(),
            description: jc.description.unwrap(),
        }
    }
}

Axum の examples にもバリデーションに関するサンプルがあります。

https://github.com/tokio-rs/axum/tree/main/examples/validator

SQLx

今回、データベースは PostgreSQL を採用しています。

todo-adapter/Cargo.toml

[dependencies]
todo-kernel = { path = "../todo-kernel" }
anyhow = "1.0.58"
async-trait = "0.1.56"
chrono = "0.4.22"
dotenv = "0.15.0"
serde = { version = "1.0.140", features = ["derive"] }
sqlx = { version = "0.6.2", features = ["runtime-tokio-rustls", "postgres", "chrono"] }
tokio = { version = "1.20.0", features = ["full"] }

レコードの作成日時や更新日時を扱いたいので、chrono クレートを使用しています。

PostgreSQL と接続する

コネクションをはるために、Poolを生成します。

todo-adapter/src/persistence/postgres.rs

use sqlx::postgres::PgPoolOptions;
use sqlx::{Pool, Postgres};
use std::env;
use std::sync::Arc;

#[derive(Clone)]
pub struct Db(pub(crate) Arc<Pool<Postgres>>);

impl Db {
    pub async fn new() -> Db {
        let pool = PgPoolOptions::new()
            .max_connections(8)
            .connect(
                &env::var("DATABASE_URL").unwrap_or_else(|_| panic!("DATABASE_URL must be set!")),
            )
            .await
            .unwrap_or_else(|_| {
                panic!("Cannot connect to the database. Please check your configuration.")
            });

        Db(Arc::new(pool))
    }
}

connectの引数に、DB 接続するための URL を渡しています。

DATABASE_URLは、dotenv クレートを使用し .env で管理しています。

sample.env

RUST_LOG=debug
HOST=127.0.0.1
PORT=8080
DATABASE_URL=postgresql://username:password@host:port/database

query_as で SELECT

query_as で SELECT 文を実装します。

SELECT した結果を格納するための構造体を定義しておきます。

ポイントは、#[derive(FromRow)]です。

todo-adapter/src/model/todo.rs

#[derive(FromRow, Debug)]
pub struct StoredTodo {
    pub id: String,
    pub title: String,
    pub description: String,
    pub status_id: String,
    pub status_code: String,
    pub status_name: String,
    pub created_at: DateTime<Utc>,
    pub updated_at: DateTime<Utc>,
}

query_asで上記構造体を結果として受け取ることができますが、SELECT の結果が0件の場合や型の不一致などで構造体への変換に失敗する場合はNoneが返ります。

また、fetch_oneでOption<Struct>、fetch_allでOption<Vec<Struct>>で結果を受け取ることができます。

todo-adapter/src/repository/todo.rs

async fn get(&self, id: &Id<Todo>) -> anyhow::Result<Option<Todo>> {
    let pool = self.db.0.clone();
    let sql = r#"
        select
            t.id as id,
            t.title as title,
            t.description as description,
            ts.id as status_id,
            ts.code as status_code,
            ts.name as status_name,
            t.created_at as created_at,
            t.updated_at as updated_at
        from
            todos as t
            inner join
                todo_statuses as ts
                on ts.id = t.status_id
        where
            t.id = $1
    "#;
    let stored_todo = query_as::<_, StoredTodo>(sql)
        .bind(id.value.to_string())
        .fetch_one(&*pool)
        .await
        .ok();

    match stored_todo {
        Some(st) => Ok(Some(st.try_into()?)),
        None => Ok(None),
    }
}

todo-adapter/src/repository/todo.rs

async fn find(&self, status: Option<TodoStatus>) -> anyhow::Result<Option<Vec<Todo>>> {
    let pool = self.db.0.clone();
    let where_status = if let Some(s) = &status {
        s.id.value.to_string()
    } else {
        "".to_string()
    };

    let mut sql = r#"
        select
            t.id as id,
            t.title as title,
            t.description as description,
            ts.id as status_id,
            ts.code as status_code,
            ts.name as status_name,
            t.created_at as created_at,
            t.updated_at as updated_at
        from
            todos as t
            inner join
                todo_statuses as ts
                on ts.id = t.status_id
        where t.status_id in ($1)
        order by t.created_at asc
    "#
    .to_string();

    if status.is_none() {
        sql = sql.replace("$1", "select id from todo_statuses");
    }

    let stored_todo_list = query_as::<_, StoredTodo>(&sql)
        .bind(where_status)
        .fetch_all(&*pool)
        .await
        .ok();

    match stored_todo_list {
        Some(todo_list) => {
            let todos = todo_list.into_iter().flat_map(|st| st.try_into()).collect();
            Ok(Some(todos))
        }
        None => Ok(None),
    }
}

クエリパラメータでステータスの指定があれば、そのステータスに該当するTodoを取得し、指定がない場合はすべてのTodoを取得する関数として実装しています。

そのため、SQL 文を定義した変数sqlとWhere句where_statusがやや複雑になっています。

ステータスの指定がない場合、Where句の$1が置換され、空文字がbindされますが、エラーは発生しません。

SQL をスマートに組み立てる方法を模索中です。

query で INSERT・UPDATE・DELETE

queryとexecuteで INSERT や UPDATE、 DELETE を実装できます。

todo-adapter/src/repository/todo.rs

async fn insert(&self, source: NewTodo) -> anyhow::Result<Todo> {
    let pool = self.db.0.clone();
    let todo: InsertTodo = source.into();
    let id = todo.id.clone();

    let _ = query("insert into todos (id, title, description) values ($1, $2, $3)")
        .bind(todo.id)
        .bind(todo.title)
        .bind(todo.description)
        .execute(&*pool)
        .await?;

    let sql = r#"
        select
            t.id as id,
            t.title as title,
            t.description as description,
            ts.id as status_id,
            ts.code as status_code,
            ts.name as status_name,
            t.created_at as created_at,
            t.updated_at as updated_at
        from
            todos as t
            inner join
                todo_statuses as ts
                on ts.id = t.status_id
        where
            t.id = $1
    "#;

    let stored_todo = query_as::<_, StoredTodo>(sql)
        .bind(id)
        .fetch_one(&*pool)
        .await?;
    Ok(stored_todo.try_into()?)
}