事务

什么是事务?

为什么需要事务?

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

启用事务

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,
}

生成了什么?

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>>;
}

可用方法

基本示例

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
}

逐步说明:

完整示例:转账

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

事务错误类型

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();
 }
}

与软删除一起使用

#[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

另请参阅