sqlx és Postgres

Az előző leckében, a „JSON API projekt” során felépítettünk egy axum-alapú REST API-t, amely a memóriában tartott állapotot (egy Vec-et vagy HashMap-et Arc mögé zárva) szolgálta ki. Ez tökéletes volt a tanuláshoz, de egy komolyabb alkalmazásban az adatoknak túl kell élniük a szerver újraindítását is – ehhez valódi adatbázis kell. Ebben a leckében megnézzük, hogyan érhetünk el Postgres adatbázist aszinkron, típusbiztos módon az sqlx crate segítségével, és hogyan illesztjük ezt bele a Tokio és axum köré épített szerverünkbe.

Miért kell külön async adatbázis-kliens?

A korábbi leckékben megismertük, hogy a Tokio executor egy kis számú szálon (jellemzően a CPU magok számával megegyező mennyiségen) futtat rengeteg async taskot, és az .await pontoknál engedi át a vezérlést más taskoknak. Ez az egész modell azon a feltételezésen alapul, hogy egy task soha nem blokkolja hosszú ideig a szálat – amint egy Future nem tud továbblépni (pl. hálózati válaszra vár), visszaadja az irányítást az executornak.

Egy hagyományos, szinkron adatbázis-driver ezt a szabályt megsérti: amikor egy lekérdezés lefut, a hívó szál egyszerűen ott várakozik, amíg a válasz megérkezik a hálózaton. Ha ezt egy Tokio worker szálon tesszük meg, az az adott szálon futó összes más taskot is blokkolja, amíg a lekérdezés be nem fejeződik – ez pontosan az, amit az async modellel el akarunk kerülni.

Ezért kell egy olyan adatbázis-klienst használnunk, amely maga is Future-öket ad vissza, és a hálózati I/O-t a Tokio reaktorán keresztül végzi, nem pedig blokkoló szisztémahívásokkal. Ezt biztosítja számunkra az sqlx.

Az sqlx crate és a connection pool

Az sqlx egy async, típusbiztos SQL-kliens Rusthoz, amely Postgres, MySQL és SQLite mellett is működik – nem ORM, hanem közvetlenül SQL-lekérdezéseket futtatunk vele, csak úgy, hogy a válasz Rust struct-okba illeszkedik.

Első lépésként a Cargo.toml-ban be kell állítanunk a megfelelő feature-öket. Az sqlx 0.7-es verziója óta a runtime (runtime-tokio) és a TLS-backend (pl. tls-rustls vagy tls-native-tls) külön feature-ök – mindkettőt meg kell adnunk, különben a crate fordítási hibát dob.

[dependencies]
tokio = { version = "1", features = ["full"] }
axum = "0.8"
sqlx = { version = "0.9", features = [
    "runtime-tokio",
    "tls-rustls",
    "postgres",
    "macros",
    "migrate",
] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"

error: at least one runtime and one TLS backend must be enabled
       (see the sqlx documentation for the full list of features)

Ha ez megvan, a fő belépési pont a PgPool: egy connection pool, azaz kapcsolatok egy készlete, amelyeket az alkalmazásunk megosztva, egymástól függetlenül futó taskok között használ. Ahelyett, hogy minden lekérdezéshez új TCP-kapcsolatot nyitnánk az adatbázis felé (ami lassú lenne), a pool előre nyit néhányat, és kiadja, visszaveszi őket, amikor egy task lekérdezést futtat.

use sqlx::postgres::PgPoolOptions;
use std::env;

#[tokio::main]
async fn main() -> Result<(), sqlx::Error> {
    // A kapcsolati string-et sose írjuk a kódba, jöjjön környezeti változóból.
    let database_url = env::var("DATABASE_URL")
        .expect("DATABASE_URL környezeti változó nincs beállítva");

    let pool = PgPoolOptions::new()
        .max_connections(5)
        .connect(&database_url)
        .await?;

    println!("Sikeres kapcsolódás az adatbázishoz!");

    pool.close().await;
    Ok(())
}

Ez a kódrészlet a megadott feature-ökkel önmagában fordul – futáshoz természetesen szükség van egy futó Postgres szerverre és egy érvényes DATABASE_URL-re, de a cargo build/cargo check a fordítás szintjén nem igényel élő adatbázis-kapcsolatot, hiszen itt még nincs jelen semmilyen SQL-lekérdezés.

A PgPoolOptions::new() egy builder-mintát követő struct, amin láncolva állíthatjuk be a pool paramétereit (max_connections, min_connections, acquire_timeout stb.), végül a connect async metódussal jön létre a valódi kapcsolat-készlet.

Query makrók és fn-ek

Az sqlx-nek két, egymást kiegészítő módja van a lekérdezések futtatására.

Az első a runtime fn-alapú megközelítés: a sqlx::query és sqlx::query_as sima async függvények, amelyeknek stringként adjuk át az SQL-t. Ezek bármikor fordulnak, mert az SQL helyességét csak futásidőben, az adatbázis felé küldve ellenőrzi a driver.

use serde::Serialize;
use sqlx::FromRow;

#[derive(Debug, Serialize, FromRow)]
struct Task {
    id: i64,
    title: String,
    done: bool,
}

async fn fetch_tasks(pool: &sqlx::PgPool) -> Result<Vec<Task>, sqlx::Error> {
    let tasks = sqlx::query_as::<_, Task>("SELECT id, title, done FROM tasks ORDER BY id")
        .fetch_all(pool)
        .await?;

    Ok(tasks)
}

A #[derive(FromRow)] gondoskodik arról, hogy a lekérdezés eredménysorai automatikusan Task értékekké alakuljanak, feltéve, hogy az oszlopnevek megegyeznek a mezőnevekkel.

A második út a compile-time ellenőrzött makró: a sqlx::query! és sqlx::query_as! makrók fordítási időben kapcsolódnak az adatbázishoz (vagy egy előre elmentett „offline” gyorsítótárhoz), és leellenőrzik, hogy a megadott SQL érvényes-e, illetve hogy a visszaadott oszlopok típusa összeillik-e a struct mezőivel.

async fn insert_task(pool: &sqlx::PgPool, title: &str) -> Result<i64, sqlx::Error> {
    let record = sqlx::query!(
        "INSERT INTO tasks (title, done) VALUES ($1, false) RETURNING id",
        title
    )
    .fetch_one(pool)
    .await?;

    Ok(record.id)
}

A kettő közötti választás tervezési döntés: a makrók korábban (fordításkor) elkapják a hibákat, de igénylik az élő sémát a build-hez; a fn-ek rugalmasabbak, de a hibák csak futásidőben derülnek ki.

Migrációk: a séma verziózása

Egy adatbázis-séma (táblák, oszlopok, indexek) idővel változik, és ezt a változást verziózottan, csapatban is reprodukálhatóan kell tudni követni – erre valók a migrációk. Az sqlx ehhez a sqlx-cli parancssori eszközt és a migrate! makrót ajánlja.

A sqlx-cli-vel (cargo install sqlx-cli) egy migrations/ mappában, időbélyeggel ellátott SQL-fájlokban tartjuk a séma-változásokat:

migrations/
├── 20240101120000_create_tasks.sql
└── 20240115093000_add_done_column.sql

Egy migrációs fájl tartalma egyszerű SQL:

-- 20240101120000_create_tasks.sql
CREATE TABLE tasks (
    id BIGSERIAL PRIMARY KEY,
    title TEXT NOT NULL,
    done BOOLEAN NOT NULL DEFAULT false
);

Az sqlx::migrate! makró fordítási időben beágyazza a migrations/ mappa tartalmát a bináris fájlba, így a szerver induláskor önállóan tudja alkalmazni a hiányzó lépéseket – nincs szükség külön deploy-lépésre.

use sqlx::postgres::PgPoolOptions;

#[tokio::main]
async fn main() -> Result<(), sqlx::Error> {
    let pool = PgPoolOptions::new()
        .max_connections(5)
        .connect("postgres://user:pass@localhost/mydb")
        .await?;

    // A migrations/ mappa tartalmát a build ágyazza be, futáskor pedig
    // lefuttatja az addig még nem alkalmazott lépéseket.
    sqlx::migrate!("./migrations").run(&pool).await?;

    println!("Séma naprakész.");
    Ok(())
}

Az sqlx egy külön táblában (_sqlx_migrations) tartja számon, mely migrációk futottak már le, így ez a hívás minden induláskor biztonságosan kiadható – ha nincs új migráció, egyszerűen nem csinál semmit.

A pool megosztása axum State-tel

A korábbi leckében már megismertük az axum State extractort megosztott állapot elérésére. A PgPool maga is olcsón klónozható (a klón csak egy referenciát másol a valódi kapcsolat-készletre), így nyugodtan tehetjük Arc-ba és megoszthatjuk a handlerek között ugyanúgy, ahogy korábban bármilyen más megosztott állapotot.

use axum::{
    extract::State,
    http::StatusCode,
    routing::get,
    Json, Router,
};
use serde::Serialize;
use sqlx::postgres::PgPoolOptions;
use sqlx::PgPool;
use std::sync::Arc;

#[derive(Debug, Serialize, sqlx::FromRow)]
struct Task {
    id: i64,
    title: String,
    done: bool,
}

async fn list_tasks(
    State(pool): State<Arc<PgPool>>,
) -> Result<Json<Vec<Task>>, StatusCode> {
    let tasks = sqlx::query_as::<_, Task>("SELECT id, title, done FROM tasks ORDER BY id")
        .fetch_all(pool.as_ref())
        .await
        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;

    Ok(Json(tasks))
}

fn app(pool: Arc<PgPool>) -> Router {
    Router::new()
        .route("/tasks", get(list_tasks))
        .with_state(pool)
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let pool = PgPoolOptions::new()
        .max_connections(5)
        .connect(&std::env::var("DATABASE_URL")?)
        .await?;

    let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await?;
    axum::serve(listener, app(Arc::new(pool))).await?;

    Ok(())
}

A list_tasks handler visszatérési típusa Result<Json<Vec<Task>>, StatusCode> – ez azért érvényes axum handler-visszatérési érték, mert axum minden olyan típusnak biztosít „HTTP válasszá alakítás” logikát, amelyet ténylegesen visszaadhatunk egy handlerből, mind a sikeres, mind a hiba-ágnak; ezt a mechanizmust majd egy későbbi leckében nézzük meg részletesebben.

Próbáld ki!

Telepíts (vagy indíts Dockerrel) egy lokális Postgres szervert, majd:

1. Hozz létre egy migrations/ mappát egy CREATE TABLE tasks (...) migrációval a fentiek alapján.
2. Írj egy count_tasks(pool: &sqlx::PgPool) -> Result<i64, sqlx::Error> async függvényt, amely a sqlx::query_as (nem makró!) segítségével egy SELECT COUNT(*) FROM tasks lekérdezést futtat, és visszaadja a darabszámot.
3. Hívd meg ezt a függvényt a main-ből a pool létrehozása és a migráció lefuttatása után, és írd ki println!-nel az eredményt.

Ha nincs kéznél Postgres, a gyakorlat első felét (a migrációs fájl megírását) élő adatbázis nélkül is el tudod végezni – csak a futtatáshoz kell majd élő kapcsolat.

Önellenőrzés

Miért nem használhatunk egy hagyományos, szinkron Postgres-drivert egy Tokio-alapú async alkalmazásban anélkül, hogy az a teljesítményt ne rontaná?

Megoldás

Mert egy blokkoló (szinkron) driver a lekérdezés idejére megállítja azt a Tokio worker szálat, amelyen fut, így az azon a szálon ütemezett összes többi task is várni kényszerül – ezzel szétesik az az alapfeltevés, hogy egy task csak `.await` pontokon adja át az irányítást. Az `sqlx` async driver, ezért a lekérdezés ideje alatt is átengedi a vezérlést más taskoknak.

Miért nem lehet mindig a query! és query_as! makrókat használni, függetlenül a projekt állapotától?

Megoldás

Mert ezek fordítási időben ellenőrzik az SQL-t és a típusokat, amihez szükségük van egy elérhető `DATABASE_URL`-re (élő séma) vagy egy előre elkészített offline gyorsítótárra (`.sqlx` mappa, `cargo sqlx prepare` paranccsal generálva). Ha ez hiányzik, a build elakad. A runtime `query`/`query_as` fn-ek ezzel szemben bármikor fordulnak, csak a hibákat futásidőben dobják.

Mi történik, ha egy már lefutott migrációt módosítunk a migrations/ mappában, majd újra elindítjuk a szervert?

Megoldás

Az `sqlx` a `_sqlx_migrations` táblában checksum-ot is tárol minden lefutott migrációhoz. Ha egy már alkalmazott migrációs fájl tartalmát módosítjuk, a checksum nem fog egyezni, és az `sqlx::migrate!(...).run(...)` hívás hibával leáll – szándékosan, hogy elkerülje a séma-inkonzisztenciát. Egy már lefutott migrációt tehát ne módosíts, helyette írj hozzá egy új migrációs fájlt.

Összefoglalás és további lépések

• Async környezetben blokkoló adatbázis-drivert nem használhatunk anélkül, hogy az executor teljesítményét ne rontanánk – erre valók az async-first kliensek, mint az sqlx.
• Az sqlx connection pool-ot (PgPool, PgPoolOptions) használ a kapcsolatok hatékony, megosztott kezelésére; a Cargo.toml-ban a runtime és a TLS-backend feature-öket egyaránt be kell kapcsolni, különben a build már a fordítás elején elhasal.
• Lekérdezést futtathatunk runtime fn-ekkel (sqlx::query, sqlx::query_as – bármikor fordulnak) vagy compile-time ellenőrzött makrókkal (sqlx::query!, sqlx::query_as! – élő sémát vagy offline cache-t igényelnek).
• A migrációk (migrations/ mappa + sqlx::migrate!) verziózzák a séma-változásokat, és a szerver induláskor önállóan tudja alkalmazni a hiányzó lépéseket.
• A PgPool Arc-ba csomagolva tökéletesen illik az axum State extractorába, így az adatbázis-elérés ugyanolyan egyszerűen kerül a handlerekbe, mint bármelyik korábban megismert megosztott állapot.

A következő leckében ráépítünk mindarra, amit itt megismertünk: bevezetjük a JWT-alapú autentikációt, hogy a felhasználók bejelentkezés után biztonságosan érjék el a most elkészített adatbázis-alapú végpontjainkat.

További olvasmányok:

• sqlx – docs.rs
• sqlx migrate modul – docs.rs
• Tokio – tokio.rs