Function accepting any Executor

[mareq]

The code below pretty much says it (or use the link to the Rust playground if you prefer that). The pool has Executor implemented on immutable reference, transaction has it implemented on mutable reference (I do not know about others - connection?). I am trying to write single function that accepts them all, similar to e.g. QueryAs::fetch_one.

I am running into problems with borrow checker when I need to run more than single query inside such function. The E type below indeed is a reference in both cases (pool, transaction) and therefore it should be borrowed, but I do not seem to be able to tell it to the compiler.

Any help anyone please?

use sqlx::{postgres::PgPoolOptions, Executor, Postgres};
use std::{borrow::Borrow, error::Error};

// CREATE TABLE customers (
//  customer_id serial PRIMARY KEY,
//  name VARCHAR UNIQUE,
//  email VARCHAR NOT NULL,
//  active bool NOT NULL DEFAULT TRUE
// );
//
// INSERT INTO
//     customers (name, email)
// VALUES
//     ('Alice', 'alice@example.tld'),
//     ('Bob', 'bob@example.tld'),
//     ('Carol', 'carol@example.tld'),
//     ('Dave', 'dave@example.tld')
// ;

#[derive(Clone, Debug, sqlx::FromRow)]
pub struct Customer {
    pub customer_id: i32,
    pub name: Option<String>,
    pub email: String,
    pub active: bool,
}

// In theory, all that is missing is to tell the compiler that the `E` is indeed *some kind of* borrow
// (for pool, it is immutable reference, for transaction it is mutable reference)
pub async fn run_two_queries<'e, E>(db_executor: E) -> Result<(), Box<dyn Error>>
where
    E: Executor<'e, Database = Postgres>,
{
    let c: Customer = sqlx::query_as("SELECT * FROM customers WHERE customer_id = $1")
        .bind(1_i32)
        .fetch_one(db_executor)
        .await?;
    println!("{:#?}", c);

    // REQUIREMENT: the function can run multiple subsequent queries (without complaints from borrow checker)
    //
    // Obviously, it makes no sense to split these particular queries into the two parts. That is not the point.
    // There are plenty of real-life situations where it makes lots of sense to run multiple queries inside
    // the same function. Such real-life scenario would only unnecessarily complicate this example though without
    // addeding any additional value.
    //
    // The code below results in the following error:
    // error[E0382]: use of moved value: `db_executor`
    //   --> src/main.rs:66:20
    //    |
    // 30 | pub async fn run_two_queries<'e, E>(db_executor: E) -> Result<(), Box<dyn Error>>
    //    |                                     ----------- move occurs because `db_executor` has type `E`, which does not implement the `Copy` trait
    // ...
    // 36 |         .fetch_one(db_executor)
    //    |                    ----------- value moved here
    // ...
    // 66 |         .fetch_one(db_executor)
    //    |                    ^^^^^^^^^^^ value used here after move
    //    |
    // help: consider further restricting this bound
    //    |
    // 32 |     E: Executor<'e, Database = Postgres> + Copy,
    //    |                                          ^^^^^^
    let c: Customer = sqlx::query_as("SELECT * FROM customers WHERE customer_id = $1")
        .bind(1_i32)
        .fetch_one(db_executor)
        .await?;
    println!("{:#?}", c);

    Ok(())
}

#[async_std::main]
async fn main() -> Result<(), Box<dyn Error>> {
    pretty_env_logger::init_timed();

    let pool = PgPoolOptions::new()
        .max_connections(5)
        .connect("postgres://postgres:postgres@localhost:5432/sandbox")
        .await?;

    // REQUIREMENT: the function accepts any `sqlx::Executor` (including `Pool`)
    println!("{:?}", run_two_queries(&pool).await?);

    // REQUIREMENT: the function accepts any `sqlx::Executor` (including `Transaction`)
    let mut tx = pool.begin().await?;
    println!("{:?}", run_two_queries(&mut tx).await?);

    // NICE TO HAVE: the function accepts any other type that
    // would be accepted e.g. by `sqlx::query::Query::fetch`

    Ok(())
}

[mareq]

I tried to tell the compiler that E is Borrow, but to no avail - I am not sure if it is just that I do not know how to say that properly, or if it is simply not possible to say that E already is borrowed - can't write &E, because we need both &mut E (transaction) and &E (pool) in the same time.

I am thinking about this solution:

• put the pool, transaction, etc. into enum mq::sqlx::Executor<sqlx::Database>,
• create my own versions of sqlx::query::Query (and all its siblings like QueryAs) and implementing execute, fetch, fetch_one, etc. functions on them such that they accept the mq::sqlx::Executor enum,
• implement From<sqlx::query::Query> conversion for my custom types mq::sqlx::query::Query (analogous for all siblings).

Another alternative would be:

• put the pool, transaction, etc. into enum mq::sqlx::Executor<sqlx::Database>,
• creae trait mq::sqlx::query::GenericQuery (and all siblings like GenericQueryAs, etc.) with generic_execute, generic_fetch, generic_fetch_one etc. functions accepting the above enum ,
• implement the above traits for sqlx::query::Query (and siblings).

Yet another alternative:

• put the pool, transaction, etc. into enum mq::sqlx::Executor<sqlx::Database>,
• implement trait sqlx::Executor for &mut mq::sqlx::Executor<sqlx::Database>.

I like the last of the three alternatives the most ATM, but I did not think about them too much (they are more-less just momentary thoughts), so I may have missed some problems (e.g. I can't get rid of the feeling that the mut part of the third alternative will bring some head-aches) making such solutions impossible/undesirable.

Also, both of them look like a lot of typing, which should really be part of sqlx, not of some mq::sqlx made-up custom thingie used in my single private project and then forgotten in a drawer. So I guess the better question would be about making some changes like that to sqlx itself. That may allow for some more elegant solution, because Query and siblings can be modified directly that way. OTOH, I am not sure how much will for this kind of changes there would be, because they would most probably introduce some backward-incompatibility (same as e.g. the change from &mut &Poll to &Pool).

EDIT: It occured to me that there is also third alternative, so I added it (and adjusted the comment at the bottom accordingly).

[mareq]

@mehcode @abonander @danielakhterov @jplatte Any recommendations/comments please?

[mehcode]

We're aware of how.. frustrating it can be when trying to abstract code across multiple points.

Most of the time, I tend to make functions take &mut PgConnection. When you do this, &mut tx (where tx is PgTransaction) will coerce into &mut PgConnection. This is due to the Deref<Target = PgConnection> on PgTransaction.

---

In the next branch ("soon" to be 0.6) you'll be able to write this:

async fn some_function(mut executor: impl PgExecutor) {
  // [...]
     .fetch_one(&mut executor)
}

and directly pass &PgPool or &mut PgConnection, etc. into the function.

[HamishWHC]

Not sure why this works

    pub async fn get_current_binary(
        &self,
        tournament_id: i64,
        conn: &mut sqlx::SqliteConnection,
    ) -> Result<Option<Binary>> {
        Ok(sqlx::query_as!(
            Binary,
            "SELECT * FROM binaries WHERE user_id=? AND tournament_id=? AND compile_result='success' ORDER BY created_at DESC",
            self.id,
            tournament_id
        )
        .fetch_optional(conn)
        .await?)
    }

    pub async fn get_profile(
        &self,
        tournament_id: i64,
        conn: &mut sqlx::SqliteConnection,
    ) -> Result<UserProfile> {
        let ts = self
            .get_tournament_stats_summary(tournament_id, conn)
            .await?;

        // Must be successfully compiled to be the user's current binary (even if the user is viewing their own profile, as this binary is used in games).
        let binary = self.get_current_binary(tournament_id, conn).await?;

        Ok(UserProfile {
            current_binary: match binary {
                Some(binary) => Some(BinaryResponse {
                    stats_summary: binary.get_stats_summary(conn).await?,
                    binary,
                }),
                None => None,
            },
            current_tournament_stats_summary: self
                .get_tournament_stats_summary(tournament_id, conn)
                .await?,
            user: self.clone(),
        })
    }

but not this

    pub async fn get_profile(
        &self,
        tournament_id: i64,
        conn: &mut sqlx::SqliteConnection, // move occurs because `conn` has type `&mut SqliteConnection`, which does not implement the `Copy` trait.
    ) -> Result<UserProfile> {
        let ts = self
            .get_tournament_stats_summary(tournament_id, conn)
            .await?;

        // Must be successfully compiled to be the user's current binary (even if the user is viewing their own profile, as this binary is used in games).
        let binary = sqlx::query_as!(
            Binary,
            "SELECT * FROM binaries WHERE user_id=? AND tournament_id=? AND compile_result='success' ORDER BY created_at DESC",
            self.id,
            tournament_id
        )
        .fetch_optional(conn) // Value moved here.
        .await?;

        Ok(UserProfile {
            current_binary: match binary {
                Some(binary) => Some(BinaryResponse {
                    stats_summary: binary.get_stats_summary(conn).await?, // Value borrowed here after move.
                    binary,
                }),
                None => None,
            },
            current_tournament_stats_summary: self
                .get_tournament_stats_summary(tournament_id, conn)
                .await?,
            user: self.clone(),
        })
    }

The second version get a compiler error about not being able to copy an &mut (see comments for where).

Would you happened to have encountered this before?

[jplatte]

Yes, you need to reborrow the connection manually (&mut *conn) in places where the &mut Connection is used as the argument to a function generic over that argument since the compiler doesn't do automatic reborrowing in generic contexts.

[HamishWHC]

Oh awesome. Knew it was a reborrow issue but kept looking for a trait/method that would help, instead of just trying to reborrow! Thanks for your help!

[0xpr03]

I get warnings about lifetimes for this when using a MySqlExecutor (I just want to allow transactions)

[Victor-N-Suadicani]

In the next branch ("soon" to be 0.6) you'll be able to write this:

async fn some_function(mut executor: impl PgExecutor) {
  // [...]
     .fetch_one(&mut executor)
}

and directly pass &PgPool or &mut PgConnection, etc. into the function.

Is there any update on this? I want to be able to both give a pool, connection or transaction. For the transaction case, I need to commit afterwards so I can't transfer ownership. But I can't use a mutable reference to the transaction, as that can only be used for a single query. I need to perform multiple queries. This is quite frustrating.

[jblachly]

WHEW

I came here to post the exact same question.

My goal was to be able to pass both PgPool as well as Transaction.

For others who may stumble across this, I was able to solve this although for reasons I don't understand:

This does not work for both (only for PgPool):

pub async fn project_locked_g1<'e, E>(
    db: &'e E,
    project_id: Uuid
) -> Result<bool, sqlx::Error>
where
    &'e E: sqlx::Executor<'e, Database = sqlx::Postgres>
{
    let qstr = "SELECT locked FROM projects WHERE projects.id = $1;";
    let result: bool = sqlx::query_scalar(qstr)
        .bind(project_id)
        .fetch_one(db)
        .await?;
    
    Ok(result)
}

While this version, differing subtly only in parameter typing and trait bounds, does work:

pub async fn project_locked_g2<'e, E>(
    db: E,
    project_id: Uuid
) -> Result<bool, sqlx::Error>
where
    E: sqlx::Executor<'e, Database = sqlx::Postgres>
{
...
}

In previously attempts, I had tried to change the Database type constraint(?) to be yet another generic parameter but that was hitting a dead end.

I am not sophisticated enough to understand why this incredibly simple solution works (and especially why removal of lifetime annotation), and would be glad to hear from @mehcode or @jplatte if they are able to comment!

[jplatte]

It's quite simply to explain why the first version doesn't work: &Connection and &Transaction don't implement Executor. The three "groups" of types that implement it are

• &Pool
• &mut Connection
• &mut Transaction

[jblachly]

That makes sense if mut is part of the type itself (and not simply a qualifier) -- thanks

[jakeprobst]

unfortunately the working code here falls over if you try and do more than one query with an executor in a single function:

pub async fn project_locked_g3<'e, E>(
    db: E,
    project_id: Uuid
) -> Result<bool, sqlx::Error>
where
    E: sqlx::Executor<'e, Database = sqlx::Postgres>
{
    let qstr = "SELECT locked FROM projects WHERE projects.id = $1;";
    let result: bool = sqlx::query_scalar(qstr)
        .bind(&project_id)
        .fetch_one(db)
        .await?;
    let result: bool = sqlx::query_scalar(qstr)
        .bind(&project_id)
        .fetch_one(db)
        .await?;
    Ok(result)
}

952 |     db: E,
    |     -- move occurs because `db` has type `E`, which does not implement the `Copy` trait
...
961 |         .fetch_one(db)
    |                    -- value moved here
...
967 |         .fetch_one(db)
    |                    ^^ value used here after move

trying to borrow doesn't help:

961 |         .fetch_one(&db)
    |          --------- ^^^ the trait `Executor<'_>` is not implemented for `&E`
    |          |
    |          required by a bound introduced by this call

[jplatte]

See #1635 (comment) for a solution.

See also #588 for an attempt to make this work with PgExecutor (which failed).

[kvzn]

Any news pls?

[mseele]

Please look at the examples in https://docs.rs/sqlx/latest/sqlx/trait.Acquire.html, that should give you an answer.

[alextes]

This discussion comes so close to what I'm looking for and yet still doesn't bring an end to my hours of digging.

I also want to have functions that accept Transaction<Postgres> and Pool<Postgres>. The reason for the former is that it is great for running tests, and for the latter that in production these functions get executed a lot and can work concurrently.

Executor<'c, Database = Postgres> appears the natural fit but actually is not for the reasons mentioned above. If I understand correctly it kinda says "one ticket to execute a query against the DB" but both are troubling for functions that want multiple executions, and from my tinkering, it appears to result in callers having to handle ticket production i.e. acquiring executors.

Accepting &mut PgConnection seems pretty okay but here is where I got in trouble. I'm experimenting with a pattern where some struct BlockStore comes with a bunch of methods and so holds its own database connection state. struct BlockStore { db: &mut PgConnection } from what I understand is no good because we can't re-use it. At best we could Mutex but we wanted to execute concurrently. Pool is great, but we lose the ease-of-testing where we pass a Transaction that each fn executes against.

My next shot at this (thanks @mseele !) is from the Acquire page it appears the way to do this is to take a generic with trait Acquire. I'll try it out. Either way, this has taken many hours if anyone has ideas on how to make things better maybe I can help implement (:

Bonus: it'd be nice to go one level up still and make a trait for this imagined BlockStore but that story seems even messier, still if anyone is using something like that I'd love to know about it!

[alextes]

Right, enum would work, but no more trait, so your function needs to implement the call twice 🙈 . (I'm sure there are smarter ways to do so but I think it ends up with what was mentioned in the OP, you have a thin layer on top of sqlx that knows what to do with your enum)

[alextes]

New idea, the struct could hold an option to both and provide the one available. doesn't change anything.

[kvzn]

@alextes same situation with you, after a lot of trying, I'm using enum now.

[gak]

Anyone have an example of this enum? I'm having issues implementing it.

[alextes]

quickly wanted to comment that nowadays, i tend to use sqlx::test for testing, which means having your struct rely on Pool works fine as it can pass a full pool to tests. with postgres available locally, creating a new database for each test is v fast.

if you really want to have fns that can take transactions what i'd do nowadays is declare the fn outside the store which holds the Pool state, and have the store acquire transactions or transaction likes in its methods before calling into the standalone testable units. i rarely find this worth the extra boilerplate myself.

[maxwolff]

pub async fn foo<'a, T: sqlx::PgExecutor<'a>>(executor: T) -> Result<()>{...} .

use &pool &mut *connection and &mut *transaction (have to deref bc now only the internal type implements Executor.

https://github.com/launchbadge/sqlx/blob/main/CHANGELOG.md#breaking under 7.0.7

[xpe]

Here I'm going to rewrite what @maxwolff wrote above using the source code (e.g. fetch_optional) as a guide. Here are the typical letters chosen for lifetimes and traits...

pub async fn foo<'c, E: PgExecutor<'c>>(executor: E)

where 'c means connection and E means Executor.

To be clear, the method signature above does not allow execution of multiple queries.

---

For a signature that allows for execution of multiple queries, see: #1635 (comment) . I use it like this...

pub trait PgAcquire<'a>: sqlx::Acquire<'a, Database = Postgres> {}

pub async fn multiple_queries<'a, A: PgAcquire <'a>>(conn: A) -> Result<(), sqlx::Error> {
    let mut conn = conn.acquire().await?;
    sqlx::query!("SELECT...").fetch_one(&mut *conn).await?;
    sqlx::query!("SELECT...").fetch_one(&mut *conn).await?;
    Ok(())
}

[adanielnoel]

Using PgAcquire for functions that run multiple queries works great! PgExecutor is consumed after one query as others noticed.

[kulak]

I just had to face this and it is hitting the limits of my understanding of RUST.

Here is what worked for me:

pub async fn insert_rec<'e>(ex: impl MySqlExecutor<'e>) -> Result<i32, Error> {
    let sql = "insert into `table` (`col`) values(?)";
    sqlx::query(sql).bind(col_val).execute(ex).await.map(|op|op.last_insert_id() as i32)
}

// inside test case:
let db_pool = MySqlPool::connect(database_url.as_str()).await.unwrap();
let mut tx = db_pool.begin().await.unwrap();
let id = insert_rec(tx.as_mut()).await.unwrap();

The context is I have a series of CRUD functions that I need to chain and I'd like to be able to use Transaction or Pool as function argument depending on context.

[xpe]

@kulak Looks like you're using the suggestion from #1136 (comment).

[andrieshiemstra]

I just had to face this and it is hitting the limits of my understanding of RUST.

Here is what worked for me:

pub async fn insert_rec<'e>(ex: impl MySqlExecutor<'e>) -> Result<i32, Error> {
    let sql = "insert into `table` (`col`) values(?)";
    sqlx::query(sql).bind(col_val).execute(ex).await.map(|op|op.last_insert_id() as i32)
}

Thanks!, saved my life after a frustrating day of trying to make a generic exe_query function... now i just have two variants for mysql and for Pg, if anyone could can figure out how to make one like this work:

pub async fn insert_rec<'e>(sql: String, ex: impl Executor<'e>) -> Result<i32, Error> {
     sqlx::query(sql).bind(col_val).execute(ex).await.map(|op|op.last_insert_id() as i32)
}

That would be awesome!

[kulak]

I run into the limit of this approach. Function with this signature cannot take 2 functions with the same signature in it. The borrow checker will not let it go through.

[Henriquelay]

Is there a thread for this but with generic impl trait? I wanna create some sort of repository accessible by the pool or connection variable. Something like:

impl<A> Find<Backup> for A
where
    A: for<'a> Acquire<'a, Database = sqlx::Sqlite>,
{
    #[tracing::instrument(skip(self), err)]
    async fn find_all(self) -> Result<Vec<Backup>, sqlx::Error> {
        sqlx::query_as(
            r#"
        SELECT * 
        FROM "Backups" 
        ORDER BY started_at DESC
        "#,
        )
        .fetch_all(self.acquire().await?)
        .await
    }
 }   

Has anyone face the same issue? Thanks