How do you generalize connections from a Pool and a Transaction while being able to enter new transactions?

[insomnimus]

I'm trying to abstract away common database operations in a modular fashion, meaning generic functions.
I need a "database" value that

1. lets me run queries against itself
2. lets me enter new transactions if necessary
3. lets me use it multiple times (it being consumed upon first use wouldn't do)

But I also need that "database" type to include &Pool and &mut SqliteConnection (or &mut Transaction, but I can deref manually), via a trait.
1 and 3 seem to be doable. I'm most having trouble with 2 because there's no trait to generalize it with, and my attempts of making one myself weren't successful.

Given those requirements, I'm unable to find a solution.

I was envisioning something similar to below (making up a trait that would encompass all that, called Sql; it's just to demonstrate the point):

async fn get_foo_credits<S: Sql>(mut db: S) -> Result<i32, sqlx::Error> {
	// No transaction
	query_scalar!("SELECT balance FROM foo WHERE foo_id = ?1", id)
		.fetch_one(&mut db)
		.await
}

async fn get_foo_credits_or_insert<S: Sql>(mut db: S, id: i32) -> Result<i32, sqlx::Error> {
	let mut tx = db.begin().await?;
	let credits = match query_scalar!("SELECT credits from foo WHERE foo_id = ?1", id)
		.fetch_optional(&mut *tx)
		.await?
	{
		Some(credits) => Ok(credits),
		None => {
			// Assume we have a `DEFAULT` clause on the foo table
			query_scalar!("INSERT INTO foo (foo_id) VALUES (?1) RETURNING balance", id)
				.fetch_one(&mut *tx)
				.await?
		}
	};

	tx.commit().await?;
	Ok(credits)
}

This example's a bit contrived, and I left out the bit about being able to call it with a Pool as well as a &mut impl Connection / Transaction for brevity. I'm hoping it's not hard to imagine use cases where that would be desired.

So... How do we actually do this kind of thing with sqlx?

[jungnitz]

This is definitely possible. As you indicate, you can create a trait (e.g. Sql) that provides sqlx::Executors and a begin function that creates new transactions. You can then implement this trait for different types of "database objects" (Pool, Connection, Transaction).
Quick untested example (which compiles):

use sqlx::{Connection, PgExecutor, Postgres, Transaction};

trait Sql {
    fn executor(&mut self) -> impl PgExecutor;
    async fn begin(&mut self) -> Result<Transaction<'_, Postgres>, sqlx::Error>;
}

impl Sql for sqlx::PgPool {
    fn executor(&mut self) -> impl PgExecutor {
        &*self
    }
    async fn begin(&mut self) -> Result<Transaction<'_, Postgres>, sqlx::Error> {
        sqlx::PgPool::begin(self).await
    }
}

impl<'c> Sql for Transaction<'c, Postgres> {
    fn executor(&mut self) -> impl PgExecutor {
        &mut **self
    }

    async fn begin(&mut self) -> Result<Transaction<'_, Postgres>, sqlx::Error> {
        Connection::begin(&mut **self).await
    }
}

async fn work(mut pool: sqlx::PgPool) {
    test_tx(&mut pool).await
}

async fn test<S: Sql>(sql: &mut S) {
    sqlx::query("select * from test").execute(sql.executor()).await.unwrap();
}

async fn test_tx<S: Sql>(sql: &mut S) {
    let mut tx = sql.begin().await.unwrap();
    sqlx::query("select * from test2").execute(tx.executor()).await.unwrap();
    test(&mut tx).await;
    tx.commit().await.unwrap();
}