如何实现支持事务/直接连接调用的Rust数据库函数？

要解决这个问题，核心是利用sqlx提供的连接状态判断能力，同时将业务逻辑与事务管理解耦，以下是更简洁可靠的实现方式：

1. 重构核心业务逻辑 #

先把与业务相关的代码抽离成独立函数，让它接受通用的Executor类型（sqlx中PgConnection和PgTransaction都实现了这个trait），这样不管是连接还是事务都能直接传入：

async fn find_player_id_and_assign_goal_core(
    &self,
    db: &mut impl sqlx::Executor<'_, Database = sqlx::Postgres>,
    id: &str,
    team_id: &str,
) -> Result<String, Error> {
    // 调用其他服务时直接传Executor，无需Option包装
    let _ = self.team_service.find_team_id(db, team_id).await?;
    let player_id = self.repo.find_player_id(db, id).await?;
    self.goal_service.assign_to_player(db, id).await?;
    
    Ok(player_id)
}

2. 实现事务管理的对外接口 #

在对外暴露的函数中，通过transaction_status()方法判断传入连接的状态，决定是直接使用还是启动新事务，同时处理从连接池获取资源的情况：

async fn find_player_id_and_assign_goal(
    &self,
    db: Option<&mut sqlx::PgConnection>,
    id: &str,
    team_id: &str,
) -> Result<String, Error> {
    match db {
        Some(conn) => match conn.transaction_status() {
            // 连接已处于事务中，直接传入核心函数使用
            sqlx::TransactionStatus::Active => {
                self.find_player_id_and_assign_goal_core(conn, id, team_id).await
            }
            // 连接处于空闲状态，启动新事务执行后提交
            sqlx::TransactionStatus::Idle => {
                let mut tx = conn.begin().await?;
                let player_id = self.find_player_id_and_assign_goal_core(&mut tx, id, team_id).await?;
                tx.commit().await?;
                Ok(player_id)
            }
            // 连接处于无效状态（如已提交/回滚），返回错误
            _ => Err(Error::Database("连接事务状态无效".into())),
        },
        None => {
            // 无传入连接时，从连接池获取并启动事务
            let mut tx = self.pool.begin().await?;
            let player_id = self.find_player_id_and_assign_goal_core(&mut tx, id, team_id).await?;
            tx.commit().await?;
            Ok(player_id)
        }
    }
}

3. 统一其他服务的接口 #

同步修改team_service、repo、goal_service中的相关函数，将参数从Option<&mut PgConnection>改为&mut impl Executor<'_, Database = sqlx::Postgres>，示例如下：

// 以team_service的find_team_id为例
async fn find_team_id(
    &self,
    db: &mut impl sqlx::Executor<'_, Database = sqlx::Postgres>,
    team_id: &str,
) -> Result<(), Error> {
    sqlx::query!("SELECT id FROM teams WHERE id = $1", team_id)
        .execute(db)
        .await?;
    Ok(())
}

关键优势 #

• 解决了无法判断连接是否在事务中的问题，通过transaction_status()精准处理两种场景
• 业务逻辑与事务管理解耦，代码更清晰易维护
• 统一了服务接口，避免冗余的Option包装，提升代码复用性
• 自动处理事务回滚：若核心逻辑执行出错，PgTransaction被drop时会自动回滚，无需手动处理

内容的提问来源于stack exchange，提问作者Fred Hors