MySQL #

Database access in Rust is most idiomatic with sqlx — an async crate that’s unique in verifying SQL queries directly at compile time through the query! macro. If you write a query with a wrong column or mismatched type, the program won’t compile, rather than crashing at runtime. This differs from traditional ORMs that map tables to structs and hide SQL behind abstractions — sqlx lets you write native SQL while still getting full type safety. This article covers connections, queries, transactions, and the repository pattern for MySQL using sqlx.

Choosing a Crate for MySQL #

There are three main options for MySQL in Rust:

Crate	Approach	Async	Query safety
`sqlx`	Direct SQL + type checking	Yes (tokio)	Compile-time via macros
`diesel`	ORM with a Rust DSL	No (sync)	Compile-time via DSL
`mysql` / `mysql_async`	Low-level driver	Both	Runtime

This article focuses on sqlx because it offers the best balance of safety, flexibility, and performance for most projects.

Installation #

[dependencies]
sqlx = { version = "0.7", features = ["runtime-tokio-native-tls", "mysql", "macros", "chrono", "uuid"] }
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }
chrono = { version = "0.4", features = ["serde"] }
dotenvy = "0.15"  # for reading .env files

Install sqlx-cli for migration management:

cargo install sqlx-cli --no-default-features --features native-tls,mysql

Connections and Connection Pools #

Always use a connection pool — creating a new connection for every query is very expensive. sqlx uses MySqlPool which manages connections automatically:

use sqlx::mysql::MySqlPoolOptions;
use sqlx::MySqlPool;

async fn buat_pool(database_url: &str) -> Result<MySqlPool, sqlx::Error> {
    MySqlPoolOptions::new()
        .max_connections(20)           // max simultaneous connections
        .min_connections(2)            // minimum standby connections
        .acquire_timeout(std::time::Duration::from_secs(5))   // connection wait limit
        .idle_timeout(std::time::Duration::from_secs(300))    // close idle connections after 5 minutes
        .max_lifetime(std::time::Duration::from_secs(1800))   // maximum connection lifetime
        .connect(database_url)
        .await
}

#[tokio::main]
async fn main() -> Result<(), sqlx::Error> {
    // Get the URL from an environment variable
    let database_url = std::env::var("DATABASE_URL")
        .unwrap_or_else(|_| "mysql://root:***@localhost/contoh".to_string());

    let pool = buat_pool(&database_url).await?;

    // Verify the connection
    sqlx::query("SELECT 1")
        .execute(&pool)
        .await?;

    println!("Connected to MySQL successfully!");
    println!("Pool size: {}/{}", pool.size(), pool.options().get_max_connections());

    Ok(())
}

Creating Tables (Migrations) #

sqlx-cli manages database migrations in a version-control-friendly way:

# Create the database
sqlx database create

# Create a new migration file
sqlx migrate add buat_tabel_pengguna

# Run all pending migrations
sqlx migrate run

# View migration status
sqlx migrate info

The generated migration file (migrations/20240824_buat_tabel_pengguna.sql):

-- migrations/20240824_buat_tabel_pengguna.sql

CREATE TABLE IF NOT EXISTS pengguna (
    id          BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
    nama        VARCHAR(100)    NOT NULL,
    email       VARCHAR(255)    NOT NULL UNIQUE,
    password    VARCHAR(255)    NOT NULL,
    peran       ENUM('user', 'admin', 'moderator') NOT NULL DEFAULT 'user',
    aktif       BOOLEAN         NOT NULL DEFAULT TRUE,
    dibuat_pada DATETIME        NOT NULL DEFAULT CURRENT_TIMESTAMP,
    diperbarui  DATETIME        NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

CREATE TABLE IF NOT EXISTS artikel (
    id              BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
    pengguna_id     BIGINT UNSIGNED NOT NULL,
    judul           VARCHAR(255)    NOT NULL,
    konten          TEXT            NOT NULL,
    diterbitkan     BOOLEAN         NOT NULL DEFAULT FALSE,
    dibuat_pada     DATETIME        NOT NULL DEFAULT CURRENT_TIMESTAMP,
    FOREIGN KEY (pengguna_id) REFERENCES pengguna(id) ON DELETE CASCADE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

CREATE INDEX idx_artikel_pengguna ON artikel(pengguna_id);
CREATE INDEX idx_artikel_diterbitkan ON artikel(diterbitkan, dibuat_pada DESC);

Run migrations from code:

use sqlx::MySqlPool;

async fn jalankan_migrasi(pool: &MySqlPool) -> Result<(), sqlx::Error> {
    sqlx::migrate!("./migrations")
        .run(pool)
        .await?;
    println!("Migration complete");
    Ok(())
}

Structs and Row Mapping #

sqlx maps database rows to Rust structs via the FromRow derive macro:

use chrono::NaiveDateTime;
use serde::{Deserialize, Serialize};
use sqlx::FromRow;

#[derive(Debug, FromRow, Serialize, Deserialize, Clone)]
struct Pengguna {
    pub id: u64,
    pub nama: String,
    pub email: String,
    #[serde(skip_serializing)]  // don't include the password in JSON responses
    pub password: String,
    pub peran: String,
    pub aktif: bool,
    pub dibuat_pada: NaiveDateTime,
    pub diperbarui: NaiveDateTime,
}

#[derive(Debug, FromRow, Serialize)]
struct Artikel {
    pub id: u64,
    pub pengguna_id: u64,
    pub judul: String,
    pub konten: String,
    pub diterbitkan: bool,
    pub dibuat_pada: NaiveDateTime,
}

// Struct for JOIN queries — not all fields from the table
#[derive(Debug, FromRow, Serialize)]
struct ArtikelDenganPenulis {
    pub id: u64,
    pub judul: String,
    pub nama_penulis: String,  // from the pengguna table
    pub dibuat_pada: NaiveDateTime,
}

Queries and Fetching #

Fetching All Rows #

use sqlx::MySqlPool;

async fn ambil_semua_pengguna(pool: &MySqlPool) -> Result<Vec<Pengguna>, sqlx::Error> {
    let pengguna = sqlx::query_as!(
        Pengguna,
        "SELECT id, nama, email, password, peran, aktif, dibuat_pada, diperbarui
         FROM pengguna
         ORDER BY dibuat_pada DESC"
    )
    .fetch_all(pool)
    .await?;

    Ok(pengguna)
}

Fetching a Single Row #

async fn cari_pengguna_by_id(pool: &MySqlPool, id: u64) -> Result<Option<Pengguna>, sqlx::Error> {
    // fetch_optional — returns None if not found (not an error)
    let pengguna = sqlx::query_as!(
        Pengguna,
        "SELECT id, nama, email, password, peran, aktif, dibuat_pada, diperbarui
         FROM pengguna
         WHERE id = ?",
        id  // parameter binding — safe from SQL injection
    )
    .fetch_optional(pool)
    .await?;

    Ok(pengguna)
}

async fn cari_pengguna_by_email(pool: &MySqlPool, email: &str) -> Result<Option<Pengguna>, sqlx::Error> {
    sqlx::query_as!(
        Pengguna,
        "SELECT * FROM pengguna WHERE email = ? AND aktif = TRUE",
        email
    )
    .fetch_optional(pool)
    .await
}

Queries with Many Parameters and Dynamic WHERE #

async fn cari_artikel(
    pool: &MySqlPool,
    pengguna_id: Option<u64>,
    hanya_diterbitkan: bool,
    batas: u32,
    offset: u32,
) -> Result<Vec<Artikel>, sqlx::Error> {
    // For dynamic WHERE, build the query conditionally
    if let Some(uid) = pengguna_id {
        sqlx::query_as!(
            Artikel,
            "SELECT id, pengguna_id, judul, konten, diterbitkan, dibuat_pada
             FROM artikel
             WHERE pengguna_id = ? AND (? = FALSE OR diterbitkan = TRUE)
             ORDER BY dibuat_pada DESC
             LIMIT ? OFFSET ?",
            uid, hanya_diterbitkan, batas, offset
        )
        .fetch_all(pool)
        .await
    } else {
        sqlx::query_as!(
            Artikel,
            "SELECT id, pengguna_id, judul, konten, diterbitkan, dibuat_pada
             FROM artikel
             WHERE (? = FALSE OR diterbitkan = TRUE)
             ORDER BY dibuat_pada DESC
             LIMIT ? OFFSET ?",
            hanya_diterbitkan, batas, offset
        )
        .fetch_all(pool)
        .await
    }
}

JOIN Queries #

async fn ambil_artikel_dengan_penulis(
    pool: &MySqlPool,
) -> Result<Vec<ArtikelDenganPenulis>, sqlx::Error> {
    sqlx::query_as!(
        ArtikelDenganPenulis,
        r#"SELECT
            a.id,
            a.judul,
            p.nama AS nama_penulis,
            a.dibuat_pada
        FROM artikel a
        INNER JOIN pengguna p ON a.pengguna_id = p.id
        WHERE a.diterbitkan = TRUE
        ORDER BY a.dibuat_pada DESC
        LIMIT 20"#
    )
    .fetch_all(pool)
    .await
}

INSERT, UPDATE, DELETE #

use sqlx::MySqlPool;

// Struct for CREATE input — without database-generated fields
#[derive(Debug)]
struct BuatPengguna {
    pub nama: String,
    pub email: String,
    pub password_hash: String,
}

async fn buat_pengguna(
    pool: &MySqlPool,
    input: &BuatPengguna,
) -> Result<u64, sqlx::Error> {
    let hasil = sqlx::query!(
        "INSERT INTO pengguna (nama, email, password) VALUES (?, ?, ?)",
        input.nama,
        input.email,
        input.password_hash
    )
    .execute(pool)
    .await?;

    // last_insert_id() returns the ID of the newly created row
    Ok(hasil.last_insert_id())
}

async fn perbarui_pengguna(
    pool: &MySqlPool,
    id: u64,
    nama: &str,
    email: &str,
) -> Result<bool, sqlx::Error> {
    let hasil = sqlx::query!(
        "UPDATE pengguna SET nama = ?, email = ? WHERE id = ?",
        nama,
        email,
        id
    )
    .execute(pool)
    .await?;

    // rows_affected() > 0 means at least one row was updated
    Ok(hasil.rows_affected() > 0)
}

async fn nonaktifkan_pengguna(
    pool: &MySqlPool,
    id: u64,
) -> Result<bool, sqlx::Error> {
    let hasil = sqlx::query!(
        "UPDATE pengguna SET aktif = FALSE WHERE id = ?",
        id
    )
    .execute(pool)
    .await?;

    Ok(hasil.rows_affected() > 0)
}

async fn hapus_pengguna(
    pool: &MySqlPool,
    id: u64,
) -> Result<bool, sqlx::Error> {
    let hasil = sqlx::query!(
        "DELETE FROM pengguna WHERE id = ?",
        id
    )
    .execute(pool)
    .await?;

    Ok(hasil.rows_affected() > 0)
}

Transactions #

Transactions ensure a group of operations executes atomically — either all succeed or all are rolled back:

use sqlx::{MySqlPool, Transaction, MySql};

async fn transfer_saldo(
    pool: &MySqlPool,
    dari_id: u64,
    ke_id: u64,
    jumlah: f64,
) -> Result<(), sqlx::Error> {
    // Start the transaction
    let mut tx = pool.begin().await?;

    // Check the sender's balance
    let saldo: Option<f64> = sqlx::query_scalar!(
        "SELECT saldo FROM akun WHERE id = ? FOR UPDATE",  // lock the row
        dari_id
    )
    .fetch_optional(&mut *tx)
    .await?;

    let saldo_pengirim = saldo.ok_or_else(|| {
        sqlx::Error::RowNotFound
    })?;

    if saldo_pengirim < jumlah {
        // Auto-rollback when tx is dropped without commit
        return Err(sqlx::Error::Protocol("Saldo tidak cukup".into()));
    }

    // Deduct the sender's balance
    sqlx::query!(
        "UPDATE akun SET saldo = saldo - ? WHERE id = ?",
        jumlah, dari_id
    )
    .execute(&mut *tx)
    .await?;

    // Add to the recipient's balance
    sqlx::query!(
        "UPDATE akun SET saldo = saldo + ? WHERE id = ?",
        jumlah, ke_id
    )
    .execute(&mut *tx)
    .await?;

    // Record the transfer history
    sqlx::query!(
        "INSERT INTO riwayat_transfer (dari_id, ke_id, jumlah) VALUES (?, ?, ?)",
        dari_id, ke_id, jumlah
    )
    .execute(&mut *tx)
    .await?;

    // Commit — if this fails, all operations above are rolled back
    tx.commit().await?;

    println!("Transfer of Rp{:.0} from {} to {} succeeded", jumlah, dari_id, ke_id);
    Ok(())
}

The Repository Pattern #

The repository pattern separates data access logic from business logic — making code more testable and easier to swap database implementations:

use async_trait::async_trait;
use sqlx::MySqlPool;

// Repository trait — defines the contract
#[async_trait]
pub trait PenggunaRepository: Send + Sync {
    async fn cari_by_id(&self, id: u64) -> Result<Option<Pengguna>, sqlx::Error>;
    async fn cari_by_email(&self, email: &str) -> Result<Option<Pengguna>, sqlx::Error>;
    async fn buat(&self, input: &BuatPengguna) -> Result<u64, sqlx::Error>;
    async fn perbarui(&self, id: u64, nama: &str) -> Result<bool, sqlx::Error>;
    async fn hapus(&self, id: u64) -> Result<bool, sqlx::Error>;
    async fn daftar(&self, batas: u32, offset: u32) -> Result<Vec<Pengguna>, sqlx::Error>;
}

// MySQL implementation
pub struct MySqlPenggunaRepository {
    pool: MySqlPool,
}

impl MySqlPenggunaRepository {
    pub fn baru(pool: MySqlPool) -> Self {
        MySqlPenggunaRepository { pool }
    }
}

#[async_trait]
impl PenggunaRepository for MySqlPenggunaRepository {
    async fn cari_by_id(&self, id: u64) -> Result<Option<Pengguna>, sqlx::Error> {
        sqlx::query_as!(
            Pengguna,
            "SELECT id, nama, email, password, peran, aktif, dibuat_pada, diperbarui
             FROM pengguna WHERE id = ?",
            id
        )
        .fetch_optional(&self.pool)
        .await
    }

    async fn cari_by_email(&self, email: &str) -> Result<Option<Pengguna>, sqlx::Error> {
        sqlx::query_as!(
            Pengguna,
            "SELECT id, nama, email, password, peran, aktif, dibuat_pada, diperbarui
             FROM pengguna WHERE email = ?",
            email
        )
        .fetch_optional(&self.pool)
        .await
    }

    async fn buat(&self, input: &BuatPengguna) -> Result<u64, sqlx::Error> {
        let hasil = sqlx::query!(
            "INSERT INTO pengguna (nama, email, password) VALUES (?, ?, ?)",
            input.nama, input.email, input.password_hash
        )
        .execute(&self.pool)
        .await?;
        Ok(hasil.last_insert_id())
    }

    async fn perbarui(&self, id: u64, nama: &str) -> Result<bool, sqlx::Error> {
        let hasil = sqlx::query!(
            "UPDATE pengguna SET nama = ? WHERE id = ?",
            nama, id
        )
        .execute(&self.pool)
        .await?;
        Ok(hasil.rows_affected() > 0)
    }

    async fn hapus(&self, id: u64) -> Result<bool, sqlx::Error> {
        let hasil = sqlx::query!("DELETE FROM pengguna WHERE id = ?", id)
            .execute(&self.pool)
            .await?;
        Ok(hasil.rows_affected() > 0)
    }

    async fn daftar(&self, batas: u32, offset: u32) -> Result<Vec<Pengguna>, sqlx::Error> {
        sqlx::query_as!(
            Pengguna,
            "SELECT id, nama, email, password, peran, aktif, dibuat_pada, diperbarui
             FROM pengguna ORDER BY id DESC LIMIT ? OFFSET ?",
            batas, offset
        )
        .fetch_all(&self.pool)
        .await
    }
}

// A service using the repository — testable with mocks
pub struct LayananPengguna<R: PenggunaRepository> {
    repo: R,
}

impl<R: PenggunaRepository> LayananPengguna<R> {
    pub fn baru(repo: R) -> Self {
        LayananPengguna { repo }
    }

    pub async fn profil(&self, id: u64) -> Result<Pengguna, String> {
        self.repo
            .cari_by_id(id)
            .await
            .map_err(|e| e.to_string())?
            .ok_or_else(|| format!("User ID {} not found", id))
    }

    pub async fn daftar_pengguna(
        &self,
        halaman: u32,
        per_halaman: u32,
    ) -> Result<Vec<Pengguna>, String> {
        let offset = (halaman.saturating_sub(1)) * per_halaman;
        self.repo
            .daftar(per_halaman, offset)
            .await
            .map_err(|e| e.to_string())
    }
}

Integrating with Axum #

A complete pattern: pool as shared state, repository in handlers:

use axum::{extract::{Path, State}, http::StatusCode, response::Json, routing::get, Router};
use std::sync::Arc;

// Application state
#[derive(Clone)]
struct AppState {
    repo: Arc<dyn PenggunaRepository>,
}

// Handler GET /pengguna/:id
async fn handler_profil(
    Path(id): Path<u64>,
    State(state): State<AppState>,
) -> Result<Json<Pengguna>, (StatusCode, String)> {
    state.repo
        .cari_by_id(id)
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?
        .map(Json)
        .ok_or_else(|| (StatusCode::NOT_FOUND, format!("User {} not found", id)))
}

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

    // Run migrations at startup
    sqlx::migrate!("./migrations").run(&pool).await?;

    let state = AppState {
        repo: Arc::new(MySqlPenggunaRepository::baru(pool)),
    };

    let app = Router::new()
        .route("/pengguna/:id", get(handler_profil))
        .with_state(state);

    let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await?;
    println!("Server at http://localhost:3000");
    axum::serve(listener, app).await?;

    Ok(())
}

Summary #

Always use a connection pool — MySqlPoolOptions with max_connections, min_connections, and acquire_timeout. One pool per application, shared via Arc or State.
query! and query_as! verify SQL at compile time — column name or data type errors are caught before the program runs, not at runtime.
Parameter binding with ? — always use parameter binding, never interpolate strings directly into SQL. This structurally prevents SQL injection.
fetch_all for many rows, fetch_one when a row is guaranteed, fetch_optional when it may not exist — pick the right one to avoid surprising errors.
rows_affected() to verify UPDATE/DELETE — an operation returning 0 means no rows were affected, possibly a missing ID.
last_insert_id() to get the ID after INSERT — return it to the caller so the newly created data can be used immediately.
Transactions with pool.begin() and tx.commit() — use them for operations that must be atomic. Transactions auto-rollback when tx is dropped without commit().
The repository pattern for testability — define a repository trait, create a MySQL implementation, and use the trait as a service parameter. During testing, inject a mock.
Run migrations at startup — sqlx::migrate!("./migrations").run(&pool).await? keeps the schema in sync without manual intervention.

← Previous: YAML Next: MSSQL →