actix-webでmongodbを利用する

rustとvueの勉強としてtodoアプリを作っています．jsonサーバーを作成するwebフレームワークにactix-webを利用しており，todoに対するコメントのデータベースにmongodbを利用しています．
actix-webでmongodbを利用しているサンプルプログラムがあまりなかったので，今回はactix-webでmongodbのrust版ドライバであるmongo-rust-driverを利用してREST apiを作成します．

バージョン

actix-webの安定版ではactix_web::main内でmongo-rust-driverが何故か動かなかった(パニックしてしまった)ため，最新版である4.0.0-beta.9を利用します．安定版で利用したい場合はactix_web::mainの代わりにactix_rt::mainを検討してみて下さい．またactix-webでは非同期に関数を実行するためCargo.tomlには以下のように追加します．

[dependencies]
actix-web = "4.0.0-beta.9"

[dependencies.mongodb]
version = "2.0.0"
default-features = false
features = ["tokio-runtime"]

以降で利用するプログラムは以下のファイル構造をとっています．

src
├ comment_models.rs
├ comment_actions.rs
├ error.rs
├ lib.rs
└ bin
　 └ build_server.rs

データ構造体

利用するデータの構造体は以下であり，comment_models.rsで定義されています．TodoCommentsがmongodbのコレクションに入るドキュメント，TodoCommentがそのうちの一つのコメントであり，UpserTodoCommentFormがinsertとupsertのときに送られてくるjsonのデータ構造です．TodoCommentsはmongodbで利用するためSerialize, Deserializeトレイトを実装し，サーバーでコレクションをcloneして利用するためCloneトレイトを実装しています．UpsertTodoComentFormではjsonデータとして取得するためDeserializeを，テストコードの時にjsonを作成するためSerializeを実装しています．

# [derive(Debug, Serialize, Deserialize, Clone)]
pub struct TodoComment {
    pub comment: String,
    pub user_name: Option<String>,
    pub reply_id: Option<i32>,
    pub timestamp: NaiveDateTime
}

# [derive(Debug, Serialize, Deserialize, Clone)]
pub struct TodoComments {
    pub todo_id: i32,
    pub comments: Vec<TodoComment>
}

# [derive(Debug, Deserialize, Serialize)]
pub struct UpsertTodoCommentForm {  // jsonで渡される構造体
    pub todo_id: i32,
    pub comment_id: i32,
    pub comment: String,
    pub user_name: Option<String>,
    pub reply_id: Option<i32>
}

データのinsert,upsert,delete,get

データのinsertに利用する関数は以下であり, comments_actions.rsで定義されています．actix-webではデータを利用するときはactix_web::web::Dataでラップしますが，actix_web::web::Dataは内部でArcを利用しているため引数をArcでラップしています．この関数はドキュメントのTodoComments部分のcommentsにTodoComment部分を追加するためのものですが，存在しない場合はTodoCommentsごと追加するためにTodoCommentsを引数としています．挿入あるいはコメントを変更したTodoCommentsを返すためにfind_one_and_updateを利用していますが，これはデフォルトでは変更する前のデータを返すためオプションを変更する必要があります．find_one_and_update部分ではTodoCommentに対応する部分を配列に挿入するという処理を行うためmongodb::bson::to_documentを利用してmongodbで利用可能な形にしています．

use mongodb::Collection;
use mongodb::bson::{doc, to_document};
use mongodb::options::{FindOptions, FindOneAndUpdateOptions, ReturnDocument};

use std::sync::Arc;
use anyhow::{Context, anyhow};

use crate::comments_models::TodoComments;
use crate::error::AppError;

pub async fn insert_comment(
    collection: Arc<Collection<TodoComments>>, 
    todo_comments: TodoComments
) -> Result<TodoComments, AppError> {
    let todo_id = todo_comments.todo_id;
    let todo_comment = &todo_comments.comments[0];

    if let None = collection.find_one(doc! {"todo_id": todo_id}, None).await? {  // todo_idが該当するものが存在しなかった場合
        collection.insert_one(&todo_comments, None).await?;
        return Ok(todo_comments);
    } else {
        let fined_one_and_update_options =  FindOneAndUpdateOptions::builder()
            .return_document(Some(ReturnDocument::After))
            .build();
        let res = collection.find_one_and_update(
            doc!{"todo_id": todo_id},
            doc!{"$push":{"comments": to_document(todo_comment).context("bson encode error")?}},
            fined_one_and_update_options
            ).await?;

        match res {
            Some(res_todo_comments) => { return Ok(res_todo_comments); },
            None => { return Err(anyhow!("There is contradiction in logic").into()); }
        }
    }
}

データをupsertする関数は以下です．ほとんどinsertと一緒ですが，引数のcomment_idに対応するインデックスの要素が存在する場合に$set，存在しない場合に$pushしています．

pub async fn upsert_comment(
    collection: Arc<Collection<TodoComments>>, 
    comment_id: i32, 
    todo_comments: TodoComments
) -> Result<TodoComments, AppError> {
    let todo_id = todo_comments.todo_id;
    let todo_comment = &todo_comments.comments[0];
    let comment = &todo_comments.comments[0].comment;

    if let None = collection.find_one(doc! {"todo_id": todo_id}, None).await? {  // todo_idが該当するものが存在しなかった場合
        collection.insert_one(&todo_comments, None).await?;
        return Ok(todo_comments);
    } else {
        let fined_one_and_update_options =  FindOneAndUpdateOptions::builder()
            .return_document(Some(ReturnDocument::After))
            .build();
        match collection.find_one_and_update(
            doc! {"todo_id": todo_id, format!("comments.{}",comment_id):{"$exists":true}},
            doc! { "$set":{format!("comments.{}.comment", comment_id): comment}},
            fined_one_and_update_options.clone()
        ).await? {
            Some(res_todo_comments) => {return Ok(res_todo_comments);},
            None => {
                let res_todo_comments = collection.find_one_and_update(
                    doc! {"todo_id": todo_id},
                    doc!{"$push":{"comments": to_document(todo_comment).context("bson encode error")?}},
                    fined_one_and_update_options.clone()
                ).await?
                .ok_or(anyhow!("There is contradiction in logic"))?;
                return Ok(res_todo_comments);
            }
        }
    }
}

テータをdeleteする関数は以下です．mongodbでインデックスを指定した配列の要素の削除は2段階必要であり，$unsetで要素をnullにした後$pullで要素がnullのものを削除しています．該当するコメントが存在しない場合は独自に定義したAppError::CommentNotFoundErrorを返します．

use mongodb::bson::Bson::Null;

pub async fn delete_comment(
    collection: Arc<Collection<TodoComments>>, 
    todo_id: i32,
    comment_id: i32
) -> Result<(), AppError> {
    if let None = collection.find_one(doc! {"todo_id": todo_id}, None).await? {  // todo_idが該当するものが存在しなかった場合
        return Err(AppError::CommentNotFoundError{todo_id, comment_id});
    } else {
        let res = collection.update_one(
            doc! {"todo_id": todo_id, format!("comments.{}", comment_id):{"$exists":true}},
            doc! {"$unset": {format!("comments.{}", comment_id): 1}},
            None
        ).await?;  // 該当コメントをnullにする
        if res.matched_count == 0 {  // todo_id, comment_idが該当するものが存在しなかった場合
            return Err(AppError::CommentNotFoundError{todo_id, comment_id});
        } else {
            collection.update_one(
                doc! {"todo_id": todo_id},
                doc!{"$pull": {"comments": Null}},
                None
            ).await?;  // nullのコメントを削除する

            collection.delete_one(
                doc! {"todo_id": todo_id, "comments": {"$size": 0}},
                None
            ).await?;  // comments配列の要素が全て削除された場合にドキュメントごと削除
        }

        return Ok(());
    }
}

データをgetする関数は以下です．try_collectでStreamからVecを作成します．

use futures::stream::TryStreamExt;

pub async fn load_todo_comments(
    collection: Arc<Collection<TodoComments>>, 
    todo_ids: Vec<i32>
) -> Result<Vec<TodoComments>, AppError> {
    let find_options = FindOptions::builder().sort(doc! {"todo_id" : 1}).build();
    let cursor = collection.find(
        doc!{"todo_id":{"$in": todo_ids}},
        find_options
    ).await?;

    let todo_comments_vec: Vec<TodoComments> = cursor.try_collect().await?;
    Ok(todo_comments_vec)
}

リクエストハンドラの定義

actix_web::Appにサービスとして登録する関数を定義します．insert, update(upsert), delete, getに対応するリクエストハンドラを定義します．以下はlib.rsで定義されています．

insertに対応するリクエストハンドラは以下となります．リクエストボディにUpsertTodoCommentFormのjsonデータを含めるため，引数に追加しています．actix-webではこのように引数にデータペース，json，クエリストリング， パスパラメータを渡すことで対応するサービスを定義します．内部でどのように処理しているか気になりますが，詳細はこちらにあります．insert_comment_hundlerでは，時間を取得した後insert_comment関数を実行しています．関数が成功した場合はjsonポディを持ったステータスコード200のレスポンスを返し，失敗した場合はErrorInternalServerErrorでエラーをラップして返します．ErrorInternalServerErrorは任意のエラーをステータスコード500のレスポンスに対応するエラーに変換するヘルパー関数であり，エラーの文字列はレスポンスのテキストから参照できます．

use mongodb::Collection;

use actix_web::{get, post, put, delete, web, Error, HttpResponse};
use actix_web::error::{
    ErrorInternalServerError,
    ErrorNotFound,
    ErrorBadRequest
};

# [post("/api/comment")]
pub async fn insert_comment_hundler(
    collection: web::Data<Collection<TodoComments>>,
    inserted_comment_form: web::Json<UpsertTodoCommentForm>
) -> Result<HttpResponse, Error> {
    let collection = collection.into_inner();
    let inserted_comment_form = inserted_comment_form.0;
    let register_datetime = Utc::now().naive_utc();
    let (_, inserted_comment) = convert_comment_upsert_form(inserted_comment_form, register_datetime);
    insert_comment(collection, inserted_comment)
    .await
    .map(|todo_comments|HttpResponse::Ok().json(todo_comments))
    .map_err(|some_err|ErrorInternalServerError(some_err))
}

upsert(update)に対応するリクエストハンドラは以下となります．ほとんどinsertのものと同じです．

# [put("/api/comment")]
pub async fn upsert_comment_hundler(
    collection: web::Data<Collection<TodoComments>>,
    upserted_comment_form: web::Json<UpsertTodoCommentForm>
) -> Result<HttpResponse, Error> {
    let collection = collection.into_inner();
    let upserted_comment_form = upserted_comment_form.0;
    let upserted_datetime = Utc::now().naive_utc();
    let (comment_id, upserted_comment) = convert_comment_upsert_form(upserted_comment_form, upserted_datetime);
    upsert_comment(collection, comment_id, upserted_comment)
    .await
    .map(|todo_comments|HttpResponse::Ok().json(todo_comments))
    .map_err(|some_err|ErrorInternalServerError(some_err))   
}

deleteに対応するリクエストハンドラは以下となります．コメントを指定するtodo_idとcomment_idはパスパラメータ―として指定しています．AppError::CommentNotFoundErrorが起こった場合ErrorNotFondでステータスコード404のレスポンスに対応するエラーに変換します．成功した場合は特にポディに入れるデータもないのでfinishとします．

# [delete("/api/comment/{todo_id}/{comment_id}")]
pub async fn delete_comment_hundler(
    collection: web::Data<Collection<TodoComments>>,
    path: web::Path<(i32, i32)>
) -> Result<HttpResponse, Error> {
    let collection = collection.into_inner();
    let path = path.into_inner();
    let todo_id = path.0;
    let comment_id = path.1;
    delete_comment(collection, todo_id, comment_id)
    .await
    .map(|_|HttpResponse::Ok().finish())
    .map_err(|some_err|{
        match some_err {
            AppError::CommentNotFoundError{todo_id: _, comment_id: _} => ErrorNotFound(some_err),
            _ => ErrorInternalServerError(some_err)
        }
    })
}

getに対するリクエストハンドラは以下となります．取得したいコメントのtodo_idsを配列で指定するため，Stringのtodo_idsを持つ構造体をクエリとして引数に渡します．クエリストリングは/api/comment?todo_ids=3,4,5のように指定できます．

use anyhow::Context;

# [derive(Deserialize, Clone)]
pub struct TodoCommentsQeuery{
    todo_ids: String
}

pub fn parse_todo_ids_str(todo_ids_str: String) -> Result<Vec<i32>, AppError>{
    let todo_ids = todo_ids_str.split(",").map(|id_str|{
        id_str.parse::<i32>().context("url parse_error")
        .map_err(|some_err|some_err.into())
    }).collect::<Result<Vec<i32>, AppError>>()?;
    Ok(todo_ids)
}

# [get("/api/comment")]
pub async fn load_todo_comments_multi_hundler(
    collection: web::Data<Collection<TodoComments>>,
    query: web::Query<TodoCommentsQeuery>
) -> Result<HttpResponse, Error> {
    let collection = collection.into_inner();
    let todo_ids = parse_todo_ids_str(query.todo_ids.clone())
    .map_err(|some_err|ErrorBadRequest(some_err))?;

    load_todo_comments(collection, todo_ids)
    .await
    .map(|todo_comments_vec|HttpResponse::Ok().json(todo_comments_vec))
    .map_err(|some_err|ErrorInternalServerError(some_err))
}

サーバーの作成

以下はbuild_server.rsにあります．
actix_web::main内でサーバーを作成します．mongodbのコレクションはアプリケーションビルダーのapp_dataメソッドで与えます．

use actix_web::{App, HttpServer};
use actix_web::web::Data;
use mongodb::Client;

# [actix_web::main]
async fn main() -> std::io::Result<()> {
    let mongo_db_url = std::env::var("MONGODB_URL").expect("MONGODB_URL must be set");
    let client = Client::with_uri_str(&mongo_db_url).await.expect("Failed to connect client");
    let db = client.database("todo_app");
    let collection = db.collection::<TodoComments>("todo_comments");

    let bind = "127.0.0.1:8080";
    HttpServer::new(move ||{
        App::new()
            .app_data(Data::new(collection.clone()))
            .service(insert_comment_hundler)
            .service(upsert_comment_hundler)
            .service(delete_comment_hundler)
            .service(load_todo_comments_multi_hundler)
    })
    .bind(&bind)?
    .run()
    .await
}

mongodを起動した後に以下を実行すれば，サーバーが起動します．

cargo run --bin build_server