Rust数据库编程：SQL与NoSQL实践指南

use tokio_postgres::{NoTls, Error};
 
#[tokio::main]
async fn main() -> Result<(), Error> {
    // 连接到数据库
    let (client, connection) = tokio_postgres::connect(
        "host=localhost user=postgres dbname=mydb",
        NoTls,
    ).await?
 
    // 在后台运行连接
    tokio::spawn(async move {
        if let Err(e) = connection.await {
            eprintln!("连接错误：{}", e);
        }
    });
 
    // 创建表
    client.execute(
        "CREATE TABLE IF NOT EXISTS users (
            id SERIAL PRIMARY KEY,
            name VARCHAR NOT NULL,
            email VARCHAR UNIQUE NOT NULL
        )",
        &[],
    ).await?
 
    // 插入数据
    client.execute(
        "INSERT INTO users (name, email) VALUES ($1, $2)",
        &[&"张三", &"zhangsan@example.com"],
    ).await?
 
    // 查询数据
    let rows = client
        .query("SELECT id, name, email FROM users", &[])
        .await?
 
    for row in rows {
        let id: i32 = row.get(0);
        let name: &str = row.get(1);
        let email: &str = row.get(2);
        println!("用户 {}: {} ({})", id, name, email);
    }
 
    Ok(())
}

async fn transfer_money(
    client: &tokio_postgres::Client,
    from_account: i32,
    to_account: i32,
    amount: f64,
) -> Result<(), Error> {
    let transaction = client.transaction().await?
 
    // 检查余额
    let row = transaction
        .query_one(
            "SELECT balance FROM accounts WHERE id = $1",
            &[&from_account],
        )
        .await?
 
    let balance: f64 = row.get(0);
    if balance < amount {
        return Err(Error::new(
            tokio_postgres::error::SqlState::INSUFFICIENT_RESOURCES,
            "余额不足",
        ));
    }
 
    // 更新两个账户
    transaction
        .execute(
            "UPDATE accounts SET balance = balance - $1 WHERE id = $2",
            &[&amount, &from_account],
        )
        .await?
 
    transaction
        .execute(
            "UPDATE accounts SET balance = balance + $1 WHERE id = $2",
            &[&amount, &to_account],
        )
        .await?
 
    // 提交事务
    transaction.commit().await?
 
    Ok(())
}

use mongodb::{Client, options::ClientOptions};
use serde::{Deserialize, Serialize};
 
#[derive(Debug, Serialize, Deserialize)]
struct User {
    #[serde(rename = "_id", skip_serializing_if = "Option::is_none")]
    id: Option<mongodb::bson::oid::ObjectId>,
    name: String,
    email: String,
    age: i32,
}
 
#[tokio::main]
async fn main() -> mongodb::error::Result<()> {
    // 连接到MongoDB
    let mut client_options = ClientOptions::parse(
        "mongodb://localhost:27017"
    ).await?
    client_options.app_name = Some("我的Rust应用".to_string());
    
    let client = Client::with_options(client_options)?
    let db = client.database("mydb");
    let collection = db.collection::<User>("users");
 
    // 插入文档
    let user = User {
        id: None,
        name: "李四".to_string(),
        email: "lisi@example.com".to_string(),
        age: 30,
    };
 
    let insert_result = collection.insert_one(user, None).await?
    println!("插入的文档ID：{}", insert_result.inserted_id);
 
    // 查询文档
    use mongodb::bson::{doc, Document};
    let filter = doc! { "age": { "$gt": 20 } };
    
    let mut cursor = collection.find(filter, None).await?
    while let Some(result) = cursor.try_next().await? {
        println!("找到用户：{:?}", result);
    }
 
    Ok(())
}

use mongodb::bson::{doc, Document};
 
async fn advanced_queries(
    collection: &mongodb::Collection<Document>
) -> mongodb::error::Result<()> {
    // 复杂查询
    let filter = doc! {
        "age": { "$gte": 20, "$lte": 50 },
        "interests": { "$in": ["编程", "读书"] }
    };
 
    // 聚合管道
    let pipeline = vec![
        doc! {
            "$match": {
                "age": { "$gte": 20 }
            }
        },
        doc! {
            "$group": {
                "_id": "$city",
                "avg_age": { "$avg": "$age" },
                "count": { "$sum": 1 }
            }
        },
        doc! {
            "$sort": { "count": -1 }
        }
    ];
 
    let mut cursor = collection.aggregate(pipeline, None).await?
    while let Some(result) = cursor.try_next().await? {
        println!("聚合结果：{:?}", result);
    }
 
    Ok(())
}

use redis::AsyncCommands;
 
async fn cache_operations() -> redis::RedisResult<()> {
    let client = redis::Client::open("redis://127.0.0.1/")?;
    let mut con = client.get_async_connection().await?
 
    // 字符串操作
    con.set("key", "value").await?
    let value: String = con.get("key").await?
    println!("获取的值：{}", value);
 
    // 列表操作
    con.rpush("list", "item1").await?
    con.rpush("list", "item2").await?
    let items: Vec<String> = con.lrange("list", 0, -1).await?
 
    // 哈希表操作
    con.hset("user:1", "name", "王五").await?
    con.hset("user:1", "age", 25).await?
    let name: String = con.hget("user:1", "name").await?
 
    // 设置过期时间
    con.expire("key", 60).await? // 60秒后过期
 
    Ok(())
}

use bb8_postgres::PostgresConnectionManager;
use tokio_postgres::NoTls;
 
async fn connection_pool_example() -> Result<(), Box<dyn std::error::Error>> {
    let manager = PostgresConnectionManager::new(
        "host=localhost user=postgres dbname=mydb"
            .parse()?
        NoTls,
    );
 
    let pool = bb8::Pool::builder()
        .max_size(20)
        .build(manager)
        .await?
 
    // 使用连接池
    let conn = pool.get().await?
    let rows = conn
        .query("SELECT * FROM users WHERE id = $1", &[&1])
        .await?
 
    for row in rows {
        println!("用户名：{}", row.get::<_, String>("name"));
    }
 
    Ok(())
}