MySQL #

MySQL adalah salah satu database relasional paling populer di dunia, digunakan secara luas mulai dari aplikasi web sederhana hingga sistem enterprise skala besar. Python menyediakan beberapa library untuk berinteraksi dengan MySQL, dan memahami cara menggunakannya dengan benar — termasuk parameterized query untuk mencegah SQL injection, manajemen transaksi, dan connection pooling — adalah fondasi penting dalam membangun aplikasi yang andal dan aman.

Instalasi #

Library resmi yang direkomendasikan untuk koneksi MySQL dari Python adalah mysql-connector-python yang dibuat oleh Oracle:

pip install mysql-connector-python

Alternatif populer lainnya adalah PyMySQL (pure Python, tanpa dependency C):

pip install pymysql

mysql-connector-python adalah pilihan yang lebih stabil untuk produksi karena didukung langsung oleh Oracle. PyMySQL lebih mudah diinstal di lingkungan yang tidak mendukung kompilasi C extension. Artikel ini menggunakan mysql-connector-python, tapi API dasarnya kompatibel dengan DB-API 2.0 yang diikuti keduanya.

Membuat Koneksi #

Koneksi ke MySQL membutuhkan beberapa parameter wajib: host, user, password, dan nama database. Jangan pernah menyimpan nilai-nilai ini langsung di kode — gunakan environment variable.

import mysql.connector
import os

# ANTI-PATTERN: hardcode kredensial di kode
conn = mysql.connector.connect(
    host="localhost",
    user="root",
    password="password123",  # ✗ -- jangan lakukan ini
    database="myapp"
)

# BENAR: baca dari environment variable
conn = mysql.connector.connect(
    host=os.getenv("DB_HOST", "localhost"),
    port=int(os.getenv("DB_PORT", "3306")),
    user=os.getenv("DB_USER"),
    password=os.getenv("DB_PASSWORD"),
    database=os.getenv("DB_NAME"),
    charset="utf8mb4",          # dukung emoji dan karakter Unicode penuh
    use_unicode=True,
    autocommit=False            # kelola transaksi secara eksplisit
)

print("Koneksi berhasil:", conn.is_connected())
conn.close()

Koneksi dengan Context Manager #

Cara yang lebih rapi adalah membungkus koneksi dalam fungsi dan menggunakan try/finally untuk memastikan koneksi selalu ditutup:

import mysql.connector
from contextlib import contextmanager
import os

def get_connection():
    return mysql.connector.connect(
        host=os.getenv("DB_HOST", "localhost"),
        user=os.getenv("DB_USER", "root"),
        password=os.getenv("DB_PASSWORD", ""),
        database=os.getenv("DB_NAME", "myapp"),
        charset="utf8mb4",
        autocommit=False
    )

@contextmanager
def db_connection():
    conn = get_connection()
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

# Penggunaan
with db_connection() as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT VERSION()")
    versi = cursor.fetchone()
    print("Versi MySQL:", versi[0])

Membuat Tabel #

Sebelum melakukan operasi data, buat tabel yang dibutuhkan. Gunakan IF NOT EXISTS agar script aman dijalankan berulang kali.

import mysql.connector

def create_tables(conn):
    cursor = conn.cursor()
    
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS pengguna (
            id          INT AUTO_INCREMENT PRIMARY KEY,
            nama        VARCHAR(100) NOT NULL,
            email       VARCHAR(150) NOT NULL UNIQUE,
            usia        INT,
            aktif       BOOLEAN DEFAULT TRUE,
            dibuat_pada TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    """)
    
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS produk (
            id       INT AUTO_INCREMENT PRIMARY KEY,
            nama     VARCHAR(200) NOT NULL,
            harga    DECIMAL(15, 2) NOT NULL,
            stok     INT DEFAULT 0,
            kategori VARCHAR(100)
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    """)
    
    conn.commit()
    cursor.close()
    print("Tabel berhasil dibuat.")

with db_connection() as conn:
    create_tables(conn)

Operasi CRUD #

Insert Data #

import mysql.connector

def tambah_pengguna(conn, nama: str, email: str, usia: int) -> int:
    cursor = conn.cursor()
    
    # ANTI-PATTERN: string formatting langsung -- rentan SQL Injection
    query = f"INSERT INTO pengguna (nama, email) VALUES ('{nama}', '{email}')"  # ✗
    
    # BENAR: gunakan parameterized query dengan placeholder %s
    query = "INSERT INTO pengguna (nama, email, usia) VALUES (%s, %s, %s)"
    cursor.execute(query, (nama, email, usia))
    
    id_baru = cursor.lastrowid
    cursor.close()
    return id_baru

# Insert banyak baris sekaligus (lebih efisien)
def tambah_banyak_pengguna(conn, daftar_pengguna: list[tuple]) -> int:
    cursor = conn.cursor()
    query = "INSERT INTO pengguna (nama, email, usia) VALUES (%s, %s, %s)"
    cursor.executemany(query, daftar_pengguna)
    jumlah = cursor.rowcount
    cursor.close()
    return jumlah

with db_connection() as conn:
    id1 = tambah_pengguna(conn, "Budi Santoso", "[email protected]", 28)
    print(f"Pengguna baru ID: {id1}")
    
    data_baru = [
        ("Sari Dewi", "[email protected]", 25),
        ("Andi Prasetyo", "[email protected]", 32),
        ("Rina Marlina", "[email protected]", 29),
    ]
    jumlah = tambah_banyak_pengguna(conn, data_baru)
    print(f"{jumlah} pengguna ditambahkan.")

SQL Injection adalah salah satu celah keamanan paling umum dan berbahaya. Jangan pernah membangun query SQL dengan string formatting atau concatenation menggunakan input dari pengguna. Selalu gunakan parameterized query dengan placeholder %s dan tuple nilai terpisah.

Read Data #

def ambil_semua_pengguna(conn) -> list[dict]:
    # dictionary=True -- row dikembalikan sebagai dict, bukan tuple
    cursor = conn.cursor(dictionary=True)
    cursor.execute("SELECT * FROM pengguna WHERE aktif = TRUE ORDER BY dibuat_pada DESC")
    hasil = cursor.fetchall()
    cursor.close()
    return hasil

def ambil_pengguna_by_id(conn, pengguna_id: int) -> dict | None:
    cursor = conn.cursor(dictionary=True)
    cursor.execute("SELECT * FROM pengguna WHERE id = %s", (pengguna_id,))
    hasil = cursor.fetchone()
    cursor.close()
    return hasil

def cari_pengguna(conn, kata_kunci: str) -> list[dict]:
    cursor = conn.cursor(dictionary=True)
    # LIKE dengan parameterized query -- tambahkan % di nilai, bukan di query
    cursor.execute(
        "SELECT * FROM pengguna WHERE nama LIKE %s OR email LIKE %s",
        (f"%{kata_kunci}%", f"%{kata_kunci}%")
    )
    hasil = cursor.fetchall()
    cursor.close()
    return hasil

with db_connection() as conn:
    semua = ambil_semua_pengguna(conn)
    for p in semua:
        print(f"[{p['id']}] {p['nama']} — {p['email']}")
    
    satu = ambil_pengguna_by_id(conn, 1)
    if satu:
        print(f"Ditemukan: {satu['nama']}")
    
    hasil_cari = cari_pengguna(conn, "budi")
    print(f"Hasil pencarian: {len(hasil_cari)} pengguna")

Update Data #

def update_pengguna(conn, pengguna_id: int, nama: str, usia: int) -> bool:
    cursor = conn.cursor()
    query = "UPDATE pengguna SET nama = %s, usia = %s WHERE id = %s"
    cursor.execute(query, (nama, usia, pengguna_id))
    berhasil = cursor.rowcount > 0
    cursor.close()
    return berhasil

def nonaktifkan_pengguna(conn, pengguna_id: int) -> bool:
    cursor = conn.cursor()
    cursor.execute("UPDATE pengguna SET aktif = FALSE WHERE id = %s", (pengguna_id,))
    berhasil = cursor.rowcount > 0
    cursor.close()
    return berhasil

with db_connection() as conn:
    ok = update_pengguna(conn, 1, "Budi Santoso Wijaya", 29)
    print("Update berhasil:", ok)

Delete Data #

def hapus_pengguna(conn, pengguna_id: int) -> bool:
    cursor = conn.cursor()
    cursor.execute("DELETE FROM pengguna WHERE id = %s", (pengguna_id,))
    berhasil = cursor.rowcount > 0
    cursor.close()
    return berhasil

with db_connection() as conn:
    ok = hapus_pengguna(conn, 99)
    print("Hapus berhasil:", ok)

Transaksi #

Transaksi memastikan serangkaian operasi database dieksekusi secara atomik — semua berhasil atau semua dibatalkan. Ini krusial untuk operasi yang melibatkan beberapa tabel atau beberapa baris yang harus konsisten.

def transfer_stok(conn, dari_produk_id: int, ke_produk_id: int, jumlah: int) -> None:
    cursor = conn.cursor()
    
    try:
        # Cek stok tersedia
        cursor.execute("SELECT stok FROM produk WHERE id = %s FOR UPDATE", (dari_produk_id,))
        baris = cursor.fetchone()
        
        if not baris or baris[0] < jumlah:
            raise ValueError(f"Stok tidak mencukupi. Tersedia: {baris[0] if baris else 0}")
        
        # Kurangi stok sumber
        cursor.execute(
            "UPDATE produk SET stok = stok - %s WHERE id = %s",
            (jumlah, dari_produk_id)
        )
        
        # Tambah stok tujuan
        cursor.execute(
            "UPDATE produk SET stok = stok + %s WHERE id = %s",
            (jumlah, ke_produk_id)
        )
        
        conn.commit()
        print(f"Transfer {jumlah} unit berhasil.")
        
    except Exception as e:
        conn.rollback()
        print(f"Transaksi dibatalkan: {e}")
        raise
    finally:
        cursor.close()

Connection Pool #

Membuat koneksi baru ke database setiap request adalah operasi yang mahal. Connection pool mempertahankan sekumpulan koneksi yang siap pakai dan mendaur ulangnya.

import mysql.connector
from mysql.connector import pooling
import os

# Buat pool saat aplikasi start
connection_pool = pooling.MySQLConnectionPool(
    pool_name="myapp_pool",
    pool_size=5,                    # jumlah koneksi yang dipertahankan
    pool_reset_session=True,
    host=os.getenv("DB_HOST", "localhost"),
    user=os.getenv("DB_USER", "root"),
    password=os.getenv("DB_PASSWORD", ""),
    database=os.getenv("DB_NAME", "myapp"),
    charset="utf8mb4"
)

def get_pooled_connection():
    return connection_pool.get_connection()

# Penggunaan -- koneksi otomatis dikembalikan ke pool saat close()
def ambil_pengguna_pool(pengguna_id: int) -> dict | None:
    conn = get_pooled_connection()
    try:
        cursor = conn.cursor(dictionary=True)
        cursor.execute("SELECT * FROM pengguna WHERE id = %s", (pengguna_id,))
        return cursor.fetchone()
    finally:
        conn.close()  # dikembalikan ke pool, bukan benar-benar ditutup

Penanganan Error #

import mysql.connector
from mysql.connector import Error, errorcode

def tambah_pengguna_aman(conn, nama: str, email: str) -> int | None:
    cursor = conn.cursor()
    try:
        cursor.execute(
            "INSERT INTO pengguna (nama, email) VALUES (%s, %s)",
            (nama, email)
        )
        conn.commit()
        return cursor.lastrowid
    
    except Error as e:
        conn.rollback()
        
        if e.errno == errorcode.ER_DUP_ENTRY:
            print(f"Email '{email}' sudah terdaftar.")
        elif e.errno == errorcode.ER_BAD_NULL_ERROR:
            print("Ada field wajib yang kosong.")
        else:
            print(f"Error database [{e.errno}]: {e.msg}")
        
        return None
    finally:
        cursor.close()

Ringkasan #

Parameterized query wajib — selalu gunakan %s sebagai placeholder dan tuple nilai terpisah; jangan pernah format string SQL langsung dari input pengguna.
cursor(dictionary=True) — gunakan agar baris hasil query dikembalikan sebagai dict (akses via nama kolom) bukan tuple (akses via indeks).
charset="utf8mb4" — gunakan charset ini (bukan utf8) agar mendukung emoji dan seluruh karakter Unicode dengan benar.
Jangan hardcode kredensial — baca host, user, password dari environment variable, bukan langsung di kode.
commit() dan rollback() eksplisit — atur autocommit=False dan kelola transaksi secara manual untuk kontrol penuh atas atomicity.
executemany() — gunakan untuk insert banyak baris sekaligus; jauh lebih efisien daripada loop execute() satu per satu.
Connection pool — gunakan MySQLConnectionPool di aplikasi web/server agar tidak membuat koneksi baru setiap request.
cursor.lastrowid — ambil ID baris yang baru saja di-insert tanpa perlu query tambahan.
cursor.rowcount — cek apakah UPDATE atau DELETE benar-benar mengubah baris, bukan hanya berjalan tanpa efek.

← Sebelumnya: YAML Berikutnya: MSSQL →