您还需要futures crate：

[dependencies]
futures = { version = "0.3.28", default-features = false }
sqlx = { version = "0.7.2", features = ["runtime-tokio", "postgres"] }

然后你可以做这样的恶作剧：

use futures::stream::StreamExt;
use sqlx::{postgres::PgPoolOptions, Either, PgPool, Row};

#[tokio::main]
async fn main() {
    # ...

    do_query(
        &pool,
        "INSERT INTO todos (description) VALUES ('One'), ('Two') RETURNING id",
    )
    .await;
    do_query(&pool, "DELETE FROM todos").await;
}

async fn do_query(pool: &PgPool, query: &str) {
    let prefix = query.split_whitespace().next().unwrap();

    let mut results = sqlx::query(query).fetch_many(pool);

    while let Some(result) = results.next().await {
        let either = result.unwrap();
        match either {
            Either::Left(res) => {
                let num = res.rows_affected();
                println!("[{prefix}] affected {num} rows");
            }
            Either::Right(row) => {
                let num = row.len();
                println!("[{prefix}] fetched {num} rows");
            }
        }
    }
}

这将返回

[INSERT] fetched 1 rows
[INSERT] fetched 1 rows
[INSERT] affected 2 rows
[DELETE] affected 2 rows