SQLite + Salvo RESTful CRUD 完整项目

基于上面 rusqlite，做一套可直接跑的 HTTP API，包含：连接池、用户增删改查、CORS、错误返回、JSON序列化。

Cargo.toml

[package]
name = "salvo_sqlite_crud"
version = "0.1.0"
edition = "2021"

[dependencies]
salvo = { version = "0.72", features = ["full"] }
rusqlite = { version = "0.29", features = ["bundled"] }
r2d2 = "0.8"
r2d2‑rusqlite = "0.19"
serde = { version = "1.0", features = ["derive"] }
anyhow = "1.0"
thiserror = "1.0"
tokio = { version = "1.0", features = ["full"] }

src/main.rs

use anyhow::{Context, Result};
use r2d2::Pool;
use r2d2_rusqlite::RusqliteConnectionManager;
use rusqlite::{params, Row};
use salvo::cors::Cors;
use salvo::http::StatusCode;
use salvo::prelude::*;
use serde::{Deserialize, Serialize};
use thiserror::Error;

// ---------------- 模型定义 ----------------
#[derive(Debug, Serialize)]
struct User {
    id: i64,
    name: String,
    email: String,
    age: Option<i32>,
}

impl User {
    fn from_row(row: &Row) -> rusqlite::Result<Self> {
        Ok(Self {
            id: row.get(0)?,
            name: row.get(1)?,
            email: row.get(2)?,
            age: row.get(3)?,
        })
    }
}

// 创建用户请求体
#[derive(Debug, Deserialize)]
struct CreateUserReq {
    name: String,
    email: String,
    age: Option<i32>,
}

#[derive(Debug, Deserialize)]
struct UpdateUserReq {
    name: Option<String>,
    email: Option<String>,
    age: Option<i32>,
}

// 统一API返回格式
#[derive(Serialize)]
struct ApiResp<T> {
    code: i32,
    msg: String,
    data: Option<T>,
}

// 业务错误
#[derive(Error, Debug)]
enum AppError {
    #[error("数据库错误:{0}")]
    Db(#[from] rusqlite::Error),
    #[error("连接池获取失败")]
    Pool,
    #[error("用户不存在")]
    NotFound,
    #[error("{0}")]
    Anyhow(#[from] anyhow::Error),
}

impl AppError {
    fn into_response<T>(self) -> ApiResp<T> {
        match self {
            AppError::NotFound => ApiResp {
                code: 404,
                msg: self.to_string(),
                data: None,
            },
            _ => ApiResp {
                code: 500,
                msg: self.to_string(),
                data: None,
            },
        }
    }
}

// ---------------- DB工具 ----------------
type DbPool = Pool<RusqliteConnectionManager>;

fn init_schema(pool: &DbPool) -> Result<()> {
    let conn = pool.get().map_err(|_| AppError::Pool)?;
    conn.execute(
        "CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            email TEXT UNIQUE NOT NULL,
            age INTEGER
        )",
        (),
    )
    .context("建表失败")?;
    Ok(())
}

// ---------------- Handler ----------------
#[handler]
async fn create_user(req: &mut Request, depot: &mut Depot, res: &mut Response) {
    let pool = depot.obtain::<DbPool>().unwrap();
    let body: CreateUserReq = match req.parse_json().await {
        Ok(v) => v,
        Err(e) => {
            res.render(Json(ApiResp::<()> {
                code: 400,
                msg: format!("请求解析错误:{}", e),
                data: None,
            }));
            return;
        }
    };

    let r = || -> Result<User, AppError> {
        let conn = pool.get().map_err(|_| AppError::Pool)?;
        conn.execute(
            "INSERT INTO users(name,email,age) VALUES (?1,?2,?3)",
            params![body.name, body.email, body.age],
        )?;
        let uid = conn.last_insert_rowid();
        let mut stmt = conn.prepare("SELECT id,name,email,age FROM users WHERE id=?1")?;
        let user = stmt.query_row(params![uid], User::from_row)?;
        Ok(user)
    };

    match r() {
        Ok(u) => res.render(Json(ApiResp {
            code: 0,
            msg: "ok".into(),
            data: Some(u),
        })),
        Err(e) => res.render(Json(e.into_response::<User>())),
    }
}

#[handler]
async fn list_users(depot: &mut Depot, res: &mut Response) {
    let pool = depot.obtain::<DbPool>().unwrap();
    let r = || -> Result<Vec<User>, AppError> {
        let conn = pool.get().map_err(|_| AppError::Pool)?;
        let mut stmt = conn.prepare("SELECT id,name,email,age FROM users ORDER BY id")?;
        let iter = stmt.query_map((), User::from_row)?;
        let mut list = Vec::new();
        for item in iter {
            list.push(item?);
        }
        Ok(list)
    };

    match r() {
        Ok(list) => res.render(Json(ApiResp {
            code: 0,
            msg: "ok".into(),
            data: Some(list),
        })),
        Err(e) => res.render(Json(e.into_response::<Vec<User>>())),
    }
}

#[handler]
async fn get_user(id: PathParam<i64>, depot: &mut Depot, res: &mut Response) {
    let pool = depot.obtain::<DbPool>().unwrap();
    let uid = id.into_inner();
    let r = || -> Result<User, AppError> {
        let conn = pool.get().map_err(|_| AppError::Pool)?;
        let mut stmt = conn.prepare("SELECT id,name,email,age FROM users WHERE id=?1")?;
        match stmt.query_row(params![uid], User::from_row) {
            Ok(u) => Ok(u),
            Err(rusqlite::Error::QueryReturnedNoRows) => Err(AppError::NotFound),
            Err(e) => Err(AppError::Db(e)),
        }
    };

    match r() {
        Ok(u) => res.render(Json(ApiResp {
            code: 0,
            msg: "ok".into(),
            data: Some(u),
        })),
        Err(e) => res.render(Json(e.into_response::<User>())),
    }
}

#[handler]
async fn update_user(
    id: PathParam<i64>,
    req: &mut Request,
    depot: &mut Depot,
    res: &mut Response,
) {
    let pool = depot.obtain::<DbPool>().unwrap();
    let uid = id.into_inner();
    let body: UpdateUserReq = match req.parse_json().await {
        Ok(v) => v,
        Err(e) => {
            res.render(Json(ApiResp::<()> {
                code: 400,
                msg: format!("请求解析错误:{}", e),
                data: None,
            }));
            return;
        }
    };

    let r = || -> Result<User, AppError> {
        let conn = pool.get().map_err(|_| AppError::Pool)?;
        let mut sql = String::from("UPDATE users SET ");
        let mut args = Vec::new();
        let mut cols = Vec::new();

        if let Some(name) = &body.name {
            cols.push("name = ?");
            args.push(name);
        }
        if let Some(email) = &body.email {
            cols.push("email = ?");
            args.push(email);
        }
        if body.age.is_some() {
            cols.push("age = ?");
            args.push(&body.age);
        }
        sql.push_str(&cols.join(","));
        sql.push_str(" WHERE id = ?");
        args.push(&Some(uid.to_string()));

        let mut stmt = conn.prepare(&sql)?;
        stmt.execute(rusqlite::params_from_iter(args))?;

        let mut stmt = conn.prepare("SELECT id,name,email,age FROM users WHERE id=?1")?;
        match stmt.query_row(params![uid], User::from_row) {
            Ok(u) => Ok(u),
            Err(rusqlite::Error::QueryReturnedNoRows) => Err(AppError::NotFound),
            Err(e) => Err(AppError::Db(e)),
        }
    };

    match r() {
        Ok(u) => res.render(Json(ApiResp {
            code: 0,
            msg: "ok".into(),
            data: Some(u),
        })),
        Err(e) => res.render(Json(e.into_response::<User>())),
    }
}

#[handler]
async fn delete_user(id: PathParam<i64>, depot: &mut Depot, res: &mut Response) {
    let pool = depot.obtain::<DbPool>().unwrap();
    let uid = id.into_inner();
    let r = || -> Result<(), AppError> {
        let conn = pool.get().map_err(|_| AppError::Pool)?;
        let affected = conn.execute("DELETE FROM users WHERE id=?1", params![uid])?;
        if affected == 0 {
            return Err(AppError::NotFound);
        }
        Ok(())
    };

    match r() {
        Ok(_) => res.render(Json(ApiResp::<()> {
            code: 0,
            msg: "删除成功".into(),
            data: None,
        })),
        Err(e) => res.render(Json(e.into_response::<()>())),
    }
}

#[tokio::main]
async fn main() -> Result<()> {
    // sqlite连接池
    let manager = RusqliteConnectionManager::new("app.db");
    let pool = Pool::builder()
        .max_size(8)
        .build(manager)
        .context("创建连接池失败")?;

    init_schema(&pool)?;
    println!("数据库初始化完成");

    let router = Router::new()
        .hoop(Cors::new().allow_any_origin())
        .hoop(salvo::affix::inject(pool))
        .post("/users", create_user)
        .get("/users", list_users)
        .get("/users/:id", get_user)
        .patch("/users/:id", update_user)
        .delete("/users/:id", delete_user);

    let acceptor = TcpListener::new("127.0.0.1:8000").bind().await;
    Server::new(acceptor).serve(router).await;
    Ok(())
}

API 接口列表

方法	路径	说明
POST	/users	创建用户
GET	/users	全部用户
GET	/users/:id	单个用户
PATCH	/users/:id	更新用户
DELETE	/users/:id	删除用户

测试示例 POST /users

{
    "name":"李四",
    "email":"lisi@test.com",
    "age":28
}

运行：

cargo run

访问 http://127.0.0.1:8000

重点注意

1. rusqlite 是同步阻塞，放在 Salvo handler 内部执行，由 tokio 运行时处理；生产环境大SQL建议扔 tokio::spawn_blocking。
2. Connection 不能跨线程，全部靠 r2d2 连接池拿连接。
3. r2d2‑rusqlite 连接池大小不要设置太大，SQLite本身并发写有限。
4. SQLite 写锁全局，高并发写场景需要注意。