entity-derive

具有自动提交/回滚的类型安全多实体事务。

什么是事务？

数据库事务是一种将多个数据库操作组合成单个原子单元的方法。这意味着：

• 全有或全无：要么所有操作都成功，要么没有任何操作被应用
• 自动回滚：如果任何操作失败，所有先前的更改都会自动撤销
• 数据一致性：您的数据库永远不会处于不一致状态

为什么需要事务？

想象一下您正在构建一个银行应用程序，需要在账户之间转账：

1. 从账户A减去100元
2. 向账户B增加100元

没有事务，如果步骤1成功但步骤2失败（网络错误、数据库崩溃等），您刚刚丢失了100元！钱从A中减去了，但从未添加到B。

使用事务，如果步骤2失败，步骤1会自动回滚。钱留在账户A中，就像什么都没发生一样。

启用事务

将transactions属性添加到您的实体：

use entity_derive::Entity;
use uuid::Uuid;

#[derive(Entity)]
#[entity(table = "accounts", transactions)]  // <- 添加这个
pub struct Account {
    #[id]
    pub id: Uuid,

    #[field(create, update, response)]
    pub user_id: Uuid,

    #[field(create, update, response)]
    pub balance: i64,
}

生成了什么？

对于带有#[entity(transactions)]的Account实体，宏会生成：

1. 事务仓库适配器

pub struct AccountTransactionRepo<'t> {
    tx: &'t mut sqlx::Transaction<'static, sqlx::Postgres>,
}

这就像您的常规仓库一样，但所有操作都在事务内部发生。

2. 构建器扩展特征

pub trait TransactionWithAccount<'p> {
    fn with_accounts(self) -> Transaction<'p, PgPool, AccountTransactionRepo<'static>>;
}

这将with_accounts()方法添加到事务构建器中。

可用方法

在事务内部，您可以访问这些方法：

基本示例

use entity_core::prelude::*;

async fn create_account(pool: &PgPool, user_id: Uuid) -> Result<Account, AppError> {
    Transaction::new(pool)           // 1. 开始构建事务
        .with_accounts()              // 2. 添加Account仓库
        .run(|mut ctx| async move {   // 3. 执行操作
            let account = ctx.accounts().create(CreateAccountRequest {
                user_id,
                balance: 0,
            }).await?;

            Ok(account)               // 4. 返回结果（自动提交）
        })
        .await
}

逐步说明：

1. Transaction::new(pool) — 使用数据库池创建新的事务构建器
2. .with_accounts() — 将Account仓库添加到事务上下文
3. .run(|mut ctx| async move { ... }) — 在事务内执行操作
4. Ok(account) — 返回Ok提交事务。返回Err回滚事务。

完整示例：转账

此示例展示了事务的全部功能：

use entity_core::prelude::*;
use uuid::Uuid;

#[derive(Debug)]
pub enum TransferError {
    Database(sqlx::Error),
    AccountNotFound(Uuid),
    InsufficientFunds { available: i64, requested: i64 },
}

impl From<sqlx::Error> for TransferError {
    fn from(e: sqlx::Error) -> Self {
        TransferError::Database(e)
    }
}

impl From<TransactionError<sqlx::Error>> for TransferError {
    fn from(e: TransactionError<sqlx::Error>) -> Self {
        TransferError::Database(e.into_inner())
    }
}

/// 原子地在两个账户之间转账。
///
/// 如果任何步骤失败，所有更改都会自动回滚。
pub async fn transfer(
    pool: &PgPool,
    from_id: Uuid,
    to_id: Uuid,
    amount: i64,
) -> Result<(), TransferError> {
    Transaction::new(pool)
        .with_accounts()
        .run(|mut ctx| async move {
            // 步骤1：获取源账户
            let from = ctx.accounts()
                .find_by_id(from_id)
                .await?
                .ok_or(TransferError::AccountNotFound(from_id))?;

            // 步骤2：检查是否有足够的钱
            if from.balance < amount {
                return Err(TransferError::InsufficientFunds {
                    available: from.balance,
                    requested: amount,
                });
            }

            // 步骤3：获取目标账户
            let to = ctx.accounts()
                .find_by_id(to_id)
                .await?
                .ok_or(TransferError::AccountNotFound(to_id))?;

            // 步骤4：从源账户减去
            // 如果这成功但步骤5失败，这将被回滚
            ctx.accounts().update(from_id, UpdateAccountRequest {
                balance: Some(from.balance - amount),
                user_id: None,
            }).await?;

            // 步骤5：添加到目标账户
            ctx.accounts().update(to_id, UpdateAccountRequest {
                balance: Some(to.balance + amount),
                user_id: None,
            }).await?;

            // 所有操作成功 - 事务将提交
            Ok(())
        })
        .await
}

不同场景下会发生什么：

一个事务中的多个实体

您可以原子地操作多个实体：

#[derive(Entity)]
#[entity(table = "accounts", transactions)]
pub struct Account {
    #[id]
    pub id: Uuid,
    #[field(create, update, response)]
    pub balance: i64,
}

#[derive(Entity)]
#[entity(table = "transfer_logs", transactions)]
pub struct TransferLog {
    #[id]
    pub id: Uuid,
    #[field(create, response)]
    pub from_account_id: Uuid,
    #[field(create, response)]
    pub to_account_id: Uuid,
    #[field(create, response)]
    pub amount: i64,
    #[auto]
    #[field(response)]
    pub created_at: DateTime<Utc>,
}

async fn transfer_with_logging(
    pool: &PgPool,
    from_id: Uuid,
    to_id: Uuid,
    amount: i64,
) -> Result<TransferLog, AppError> {
    Transaction::new(pool)
        .with_accounts()      // 添加Account仓库
        .with_transfer_logs() // 添加TransferLog仓库
        .run(|mut ctx| async move {
            // 更新余额
            let from = ctx.accounts().find_by_id(from_id).await?
                .ok_or(AppError::NotFound)?;

            ctx.accounts().update(from_id, UpdateAccountRequest {
                balance: Some(from.balance - amount),
            }).await?;

            let to = ctx.accounts().find_by_id(to_id).await?
                .ok_or(AppError::NotFound)?;

            ctx.accounts().update(to_id, UpdateAccountRequest {
                balance: Some(to.balance + amount),
            }).await?;

            // 创建日志条目 - 全部在同一事务中！
            let log = ctx.transfer_logs().create(CreateTransferLogRequest {
                from_account_id: from_id,
                to_account_id: to_id,
                amount,
            }).await?;

            Ok(log)
        })
        .await
}

如果日志创建失败，两个账户更新都会回滚！

错误处理

自动回滚

从闭包返回的任何错误都会触发回滚：

Transaction::new(pool)
    .with_accounts()
    .run(|mut ctx| async move {
        ctx.accounts().update(id, dto).await?;  // 成功

        // 某些验证失败
        if amount < 0 {
            return Err(AppError::InvalidAmount);  // <- 触发回滚！
        }

        // 这永远不会执行，上面的更新会被撤销
        ctx.accounts().update(other_id, other_dto).await?;

        Ok(())
    })
    .await

事务错误类型

TransactionError枚举告诉您出了什么问题：

use entity_core::transaction::TransactionError;

let result = Transaction::new(pool)
    .with_accounts()
    .run(|mut ctx| async move { /* ... */ })
    .await;

match result {
    Ok(value) => {
        println!("成功：{:?}", value);
    }
    Err(e) => {
        if e.is_begin() {
            println!("启动事务失败");
        } else if e.is_operation() {
            println!("操作失败：{}", e);
        } else if e.is_commit() {
            println!("提交失败");
        } else if e.is_rollback() {
            println!("回滚失败");
        }

        // 获取内部数据库错误
        let db_error: sqlx::Error = e.into_inner();
    }
}

与软删除一起使用

事务遵守soft_delete属性：

#[derive(Entity)]
#[entity(table = "documents", transactions, soft_delete)]
pub struct Document {
    #[id]
    pub id: Uuid,

    #[field(create, response)]
    pub title: String,

    #[field(skip)]
    pub deleted_at: Option<DateTime<Utc>>,  // soft_delete必需
}

async fn archive_document(pool: &PgPool, id: Uuid) -> Result<bool, AppError> {
    Transaction::new(pool)
        .with_documents()
        .run(|mut ctx| async move {
            // 这设置deleted_at = NOW()而不是DELETE
            let deleted = ctx.documents().delete(id).await?;
            Ok(deleted)
        })
        .await
}

最佳实践

1. 保持事务简短

错误示例：长时间运行的事务

Transaction::new(pool)
    .with_accounts()
    .run(|mut ctx| async move {
        let account = ctx.accounts().find_by_id(id).await?;

        // 不要这样做：在事务内调用外部API
        let rate = external_api.get_exchange_rate().await?;  // <- 慢！

        ctx.accounts().update(id, dto).await?;
        Ok(())
    })
    .await

正确示例：在外部进行慢操作

// 在启动事务之前获取外部数据
let rate = external_api.get_exchange_rate().await?;

Transaction::new(pool)
    .with_accounts()
    .run(|mut ctx| async move {
        ctx.accounts().update(id, UpdateAccountRequest {
            balance: Some(calculate_new_balance(rate)),
        }).await?;
        Ok(())
    })
    .await

2. 不要为单个操作使用事务

不必要：

Transaction::new(pool)
    .with_users()
    .run(|mut ctx| async move {
        ctx.users().find_by_id(id).await  // 只有一个操作！
    })
    .await

更好：使用常规仓库

pool.find_by_id(id).await  // 不需要事务

3. 正确处理所有错误

始终使用?传播错误：

Transaction::new(pool)
    .with_accounts()
    .run(|mut ctx| async move {
        let result = ctx.accounts().update(id, dto).await;

        // 不要这样做：吞掉错误
        if let Err(e) = result {
            log::error!("更新失败：{}", e);
            // 事务不会正确回滚！
        }

        // 这样做：传播错误
        ctx.accounts().update(id, dto).await?;  // <- 使用?

        Ok(())
    })
    .await

常见模式

检查然后更新

Transaction::new(pool)
    .with_products()
    .run(|mut ctx| async move {
        let product = ctx.products().find_by_id(id).await?
            .ok_or(AppError::NotFound)?;

        if product.stock < quantity {
            return Err(AppError::OutOfStock);
        }

        ctx.products().update(id, UpdateProductRequest {
            stock: Some(product.stock - quantity),
            ..Default::default()
        }).await?;

        Ok(product)
    })
    .await

创建多个相关记录

Transaction::new(pool)
    .with_orders()
    .with_order_items()
    .run(|mut ctx| async move {
        // 创建父记录
        let order = ctx.orders().create(CreateOrderRequest {
            customer_id,
            status: "pending".to_string(),
        }).await?;

        // 创建子记录
        for item in items {
            ctx.order_items().create(CreateOrderItemRequest {
                order_id: order.id,
                product_id: item.product_id,
                quantity: item.quantity,
            }).await?;
        }

        Ok(order)
    })
    .await

另请参阅

• [[属性|属性参考]] — 完整的属性文档
• [[钩子|生命周期钩子]] — 在操作前/后运行代码
• [[命令|命令]] — CQRS命令模式
• [[事件|事件]] — 跟踪实体变更