SQLx query in handler

[ruizdiazever]

I have not been able to solve this problem although I have seen in the community that it is a recurring problem with no apparent answer!

main.rs

use axum::{extract::Extension,T, Router};
use sqlx::postgres::PgPoolOptions;
use tower_http::trace::TraceLayer;
use axum::{
    error_handling::HandleErrorLayer,
    http::{StatusCode, Uri},
    response::IntoResponse, Server,
};
use dotenvy::dotenv;
use std::{env, error::Error, time::Duration};
use tower::{timeout::TimeoutLayer, ServiceBuilder};
use tracing::{error, info};

mod api;
use api::{health, user, energy};


#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    // Load environment configuration from .env
    dotenv().expect("Set your configuration in a .env file");

    // Init Tracing
    tracing_subscriber::fmt::init();

    // Server adress
    let server_addr = env::var("SERVER").expect("Define SERVER=host:port in your .env");
    let server_addr = server_addr.parse().expect("Define SERVER=host:port in your .env");

    // PostgreSQL 15
    let database_url = env::var("DATABASE_URL").expect("Define DATABASE_URL in your .env");
    let pool = PgPoolOptions::new()
        .max_connections(50)
        .connect(&database_url)
        .await?;

    let app = Router::new()
        .route("/", get(health::root_response))
        .route("/health", get(health::health_check))
        .route("/metrics", post(energy::insert_test_metrics))
        .route("/users", get(user::get_users))
        .route("/user", get(user::get_user))
        .layer(Extension(pool))
        .layer(
            ServiceBuilder::new()
                .layer(TraceLayer::new_for_http())
                .layer(HandleErrorLayer::new(|_| async {
                    StatusCode::REQUEST_TIMEOUT
                }))
                .layer(TimeoutLayer::new(Duration::from_secs(5))),
        )
        .fallback(fallback_handler);

    info!("Launching server: http://{server_addr}/");
    Server::bind(&server_addr)
        .serve(app.into_make_service())
        .with_graceful_shutdown(signal_shutdown())
        .await
        .unwrap();
    Ok(())
}


async fn signal_shutdown() {
    tokio::signal::ctrl_c()
        .await
        .expect("Expect tokio signal ctrl-c");
}


async fn fallback_handler(uri: Uri) -> impl IntoResponse {
    error!("No route for {}", uri);
    (StatusCode::NOT_FOUND, format!("No route for {}", uri))
}

energy.rs

use uuid::Uuid;
use axum::response::Json;
use sqlx::PgPool;
use axum::extract;
use serde::{Deserialize, Serialize};


#[derive(Deserialize, Serialize)]
struct Metric {
    device_id: Uuid,
    watt: f64,
    quality_signal_wifi: f64,
    quality_signal_bluetooth: f64
}


//#[axum::debug_handler]
pub async fn insert_test_metrics(
    axum::Json(metric): axum::Json<Metric>,
    pool: extract::Extension<PgPool>
    ) -> Result<axum::Json<Metric>, axum::Error>  {

    // Query
    sqlx::query!(
        "INSERT INTO test.energy_metric ( device_id, watt, quality_signal_wifi, quality_signal_bluetooth ) values ($1, $2, $3, $4);",
        metric.device_id,
        metric.watt,
        metric.quality_signal_wifi,
        metric.quality_signal_bluetooth
    )
        .execute(&*pool)
        .await
        .map_err(|e| {
                eprintln!("Error to insert the metric in DB: {:?}", e);
                axum::Error::new("Resource not found!");
        }
    );

    Ok(Json(metric))
}

Error

error[E0277]: the trait bound `fn(axum::Json<Metric>, Extension<Pool<Postgres>>) -> impl std::future::Future<Output = Result<axum::Json<Metric>, axum::Error>> {insert_test_metrics}: Handler<_, _, _>` is not satisfied
   --> src/main.rs:40:33
    |
40  |         .route("/metrics", post(energy::insert_test_metrics))
    |                            ---- ^^^^^^^^^^^^^^^^^^^^^^^^^^^ the trait `Handler<_, _, _>` is not implemented for fn item `fn(axum::Json<Metric>, Extension<Pool<Postgres>>) -> impl std::future::Future<Output = Result<axum::Json<Metric>, axum::Error>> {insert_test_metrics}`
    |                            |
    |                            required by a bound introduced by this call
    |
    = note: Consider using `#[axum::debug_handler]` to improve the error message
    = help: the following other types implement trait `Handler<T, S, B>`:
              <MethodRouter<S, B> as Handler<(), S, B>>
              <axum::handler::Layered<L, H, T, S, B, B2> as Handler<T, S, B2>>
note: required by a bound in `post`
   --> /Users/ever/.cargo/registry/src/index.crates.io-6f17d22bba15001f/axum-0.6.19/src/routing/method_routing.rs:407:1
    |
407 | top_level_handler_fn!(post, POST);
    | ^^^^^^^^^^^^^^^^^^^^^^----^^^^^^^
    | |                     |
    | |                     required by a bound in this function
    | required by this bound in `post`
    = note: this error originates in the macro `top_level_handler_fn` (in Nightly builds, run with -Z macro-backtrace for more info)

For more information about this error, try `rustc --explain E0277`.
error: could not compile `api_v2` (bin "api_v2") due to previous error

Cargo.toml

[package]
name = "api_v2"
version = "0.1.0"
edition = "2021"

# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

[dependencies]
async-trait = "0.1.71"
r2d2 = "0.8.10"
axum = { version = "0.6.19", features = ["macros"] }
axum-macros = "0.3.8"
dotenvy = "0.15.7"
futures-util = "0.3.28"
serde = { version = "1.0.173", features = ["derive"] }
serde_json = "1.0.103"
tokio = { version = "1.29.1", features = ["full"] }
tower = { version = "0.4.13", features = ["timeout"] }
tower-http = { version = "0.4.3", features = ["trace"] }
tracing = "0.1.37"
tracing-subscriber = "0.3.17"
uuid = { version = "1.4.1", features = ["v4", "serde"] }
sqlx = { version = "0.7.1", features = ["postgres", "json", "runtime-tokio-rustls", "uuid", "time", "chrono"] }
bcrypt = "0.15.0"
thiserror = "1.0.43"
futures = "0.3.28"
anyhow = "1.0.72"
log = "0.4.19"
rand = "0.8.5"
chrono = "0.4.24"
serde_with = "3.1.0"
validator = { version = "0.16.1", features = ["derive"] }
axum-sqlx-tx = { version = "0.5.0", features = ["postgres", "runtime-tokio-native-tls"] }

[dev-dependencies]
test-context = "0.1.4"

[davidpdrsn]

What does #[axum::debug_handler] say?

[ruizdiazever]

Hi David, first of all it is a pleasure for me to greet you and I admire you very much, I have solved the problem by doing several tests, I leave it below, although it is harcoded it works.

About #[axum::debug_handler]: said the same thing as the compilation error!

use sqlx::Postgres;
use sqlx::Pool;
use uuid::{Uuid, uuid};
use serde::{Deserialize, Serialize};
use axum::extract::Extension;
use axum::http::StatusCode;
use axum::response::IntoResponse;
use serde_json::json;
use axum::response::Json;


#[derive(Deserialize, Serialize, Debug)]
struct Metric {
    device_id: Uuid,
    watt: f64,
    quality_signal_wifi: f64,
    quality_signal_bluetooth: f64,
}


#[axum::debug_handler]
pub async fn insert_test_metrics(Extension(pool): Extension<Pool<Postgres>>) -> impl IntoResponse {
    let metric: Metric = Metric {
        device_id: uuid!("14559db9-ea8e-4dcc-a694-3f02402008f6"),
        watt: 5002.2,
        quality_signal_wifi: 50.2,
        quality_signal_bluetooth: 52.2,
    };

    sqlx::query!(
        "INSERT INTO test.energy_metric (device_id, watt, quality_signal_wifi, quality_signal_bluetooth) VALUES ($1, $2, $3, $4)",
        metric.device_id,
        metric.watt,
        metric.quality_signal_wifi,
        metric.quality_signal_bluetooth
    )
    .execute(&pool)
    .await
    .expect("Unable to insert the record!");

    let output = json!({"info": "Record inserted!"});
    (StatusCode::OK, Json(output))
}

Thanks

[davidpdrsn]

Great! Would you mind explaining what the solution was? In case someone else in the future finds themselves here.

[ruizdiazever]

As I didn't understand the problem very well (I still have a lot to learn about Rust) I can't speak of a universal solution, but the key was to set up the query well thanks to this resource since the examples I found on github did not work with the latest version of axum or sqlx!

[iblacksand]

I may be mistaken, but I believe the error is because of the order of the parameters of insert_test_metrics. I very similar error message, and it was because the Json payload parameter always has to be the last argument. For example, instead of

pub async fn insert_test_metrics(
    axum::Json(metric): axum::Json<Metric>,
    pool: extract::Extension<PgPool>
    ) -> Result<axum::Json<Metric>, axum::Error>  {

it should be

pub async fn insert_test_metrics(
    pool: extract::Extension<PgPool>,
    axum::Json(metric): axum::Json<Metric>
    ) -> Result<axum::Json<Metric>, axum::Error>  {

In the fixed example, the Json parameter is removed completely, so I think this solved the error you saw, rather than the query, since the SQLx query is not related to the error message.

Again I may be mistaken about this specific case, but for people who see the similar error message "the trait `Handler<_, _, _>` is not implemented for fn item ...", you can try making your Json parameter the last argument.

[nickmonad]

If anyone else runs into this issue while using a sqlx::query! macro inside a handler alongside .with_state(...): my issue was not related to the order of the handler arguments (State was first in the list), but rather due to the simple fact that my sqlx::query! macro was failing compilation, since I wasn't running a database with the migrations applied, or had an "offline" data.json for it to check the query against. This manifested as a trait Handler<_, _, _> is not implemented error.

Prior to switching over to .with_state() for my routers, this didn't manifest when using the Extension extractor, as that moved the problem to run-time, so it makes sense I would now see this at compile-time using the State extractor.

Anyway, hope somebody sees this and doesn't waste a couple hours fighting the compiler, like I did...

[jmkng]

This happened to me while using SQLx in a handler, it didn't like me having a lock to a mutex at the same time.

If you have a similar situation, drop the lock early on:

            let handle = {
                let mut links = state.links.lock().unwrap();
                links.remove(&id)
                // unlock ->
            };
            
            if let Some(handle) = handle {
                handle.abort();
                sqlx::query(r#"UPDATE "monitor" SET "active" = false WHERE id = $1"#)
                    .bind(id)
                    .execute(state.pool.as_ref())
                    .await;
            }