MSSQL #

Microsoft SQL Server (MSSQL) adalah database relasional enterprise yang banyak digunakan di lingkungan korporat, terutama pada stack teknologi berbasis Windows dan .NET. Python dapat berinteraksi dengan MSSQL melalui library pyodbc yang menggunakan standar ODBC, membuatnya fleksibel untuk terhubung ke berbagai versi SQL Server — baik yang di-host lokal, di server Windows, maupun di Azure SQL Database. Memahami perbedaan sintaks antara MSSQL dan database lain (seperti placeholder ? alih-alih %s) adalah kunci untuk menghindari bug yang sulit dilacak saat berpindah platform.

Instalasi #

pip install pyodbc

Selain library Python, kamu juga perlu menginstal ODBC Driver for SQL Server yang sesuai dengan sistem operasi:

# macOS (menggunakan Homebrew)
brew install msodbcsql18

# Linux (Ubuntu/Debian)
curl https://packages.microsoft.com/keys/microsoft.asc | sudo apt-key add -
curl https://packages.microsoft.com/config/ubuntu/22.04/prod.list \
    | sudo tee /etc/apt/sources.list.d/mssql-release.list
sudo apt-get update
sudo ACCEPT_EULA=Y apt-get install -y msodbcsql18

# Windows: unduh dan instal dari
# https://learn.microsoft.com/sql/connect/odbc/download-odbc-driver-for-sql-server

Versi ODBC Driver yang direkomendasikan saat ini adalah ODBC Driver 18 for SQL Server. Periksa versi yang terinstal di sistem kamu dengan menjalankan odbcinst -q -d di Linux/macOS, atau lihat di “ODBC Data Sources” di Windows.

Membuat Koneksi #

MSSQL menggunakan connection string berbasis ODBC yang sedikit berbeda dari library database lain. Ada dua skenario umum: SQL Server Authentication (username + password) dan Windows Authentication (tanpa password, menggunakan akun Windows saat ini).

import pyodbc
import os

# ANTI-PATTERN: hardcode kredensial langsung di kode
conn = pyodbc.connect(
    "DRIVER={ODBC Driver 18 for SQL Server};"
    "SERVER=localhost;"
    "DATABASE=myapp;"
    "UID=sa;"
    "PWD=password123"  # ✗ -- jangan lakukan ini
)

# BENAR: baca dari environment variable
def build_connection_string() -> str:
    server   = os.getenv("MSSQL_HOST", "localhost")
    database = os.getenv("MSSQL_DB", "myapp")
    user     = os.getenv("MSSQL_USER", "sa")
    password = os.getenv("MSSQL_PASSWORD", "")
    driver   = os.getenv("MSSQL_DRIVER", "ODBC Driver 18 for SQL Server")
    
    return (
        f"DRIVER={{{driver}}};"
        f"SERVER={server};"
        f"DATABASE={database};"
        f"UID={user};"
        f"PWD={password};"
        "TrustServerCertificate=yes;"  # perlu untuk koneksi lokal/dev
        "Encrypt=yes;"
    )

conn = pyodbc.connect(build_connection_string())
conn.autocommit = False  # kelola transaksi secara eksplisit
print("Koneksi berhasil")
conn.close()

# Windows Authentication (tanpa username/password)
conn_string_windows = (
    "DRIVER={ODBC Driver 18 for SQL Server};"
    "SERVER=localhost\\SQLEXPRESS;"
    "DATABASE=myapp;"
    "Trusted_Connection=yes;"
)
conn = pyodbc.connect(conn_string_windows)

Context Manager untuk Koneksi #

import pyodbc
from contextlib import contextmanager
import os

@contextmanager
def db_connection():
    conn = pyodbc.connect(build_connection_string())
    conn.autocommit = False
    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("SQL Server Version:", versi[0][:50])

Membuat Tabel #

Sintaks DDL MSSQL sedikit berbeda dari MySQL — gunakan IDENTITY untuk auto-increment dan NVARCHAR untuk string Unicode.

def create_tables(conn):
    cursor = conn.cursor()
    
    # Cek apakah tabel sudah ada sebelum membuat
    cursor.execute("""
        IF NOT EXISTS (
            SELECT * FROM sysobjects 
            WHERE name='pengguna' AND xtype='U'
        )
        CREATE TABLE pengguna (
            id          INT IDENTITY(1,1) PRIMARY KEY,
            nama        NVARCHAR(100) NOT NULL,
            email       NVARCHAR(150) NOT NULL,
            usia        INT,
            aktif       BIT DEFAULT 1,
            dibuat_pada DATETIME2 DEFAULT GETDATE(),
            CONSTRAINT UQ_pengguna_email UNIQUE (email)
        )
    """)
    
    cursor.execute("""
        IF NOT EXISTS (
            SELECT * FROM sysobjects 
            WHERE name='produk' AND xtype='U'
        )
        CREATE TABLE produk (
            id       INT IDENTITY(1,1) PRIMARY KEY,
            nama     NVARCHAR(200) NOT NULL,
            harga    DECIMAL(15, 2) NOT NULL,
            stok     INT DEFAULT 0,
            kategori NVARCHAR(100)
        )
    """)
    
    conn.commit()
    cursor.close()
    print("Tabel berhasil dibuat.")

with db_connection() as conn:
    create_tables(conn)

Perbedaan penting dari MySQL:
IDENTITY(1,1) menggantikan AUTO_INCREMENT
NVARCHAR untuk string Unicode (mendukung karakter non-Latin), VARCHAR hanya ASCII
BIT menggantikan BOOLEAN
DATETIME2 menggantikan TIMESTAMP — presisi lebih tinggi
GETDATE() menggantikan CURRENT_TIMESTAMP

Operasi CRUD #

Perbedaan Placeholder — Penting! #

# ANTI-PATTERN: menggunakan %s seperti di MySQL/PostgreSQL
cursor.execute("INSERT INTO pengguna (nama) VALUES (%s)", ("Budi",))  # ✗ -- error di pyodbc

# BENAR: pyodbc menggunakan tanda tanya (?) sebagai placeholder
cursor.execute("INSERT INTO pengguna (nama) VALUES (?)", ("Budi",))   # ✓

Insert Data #

def tambah_pengguna(conn, nama: str, email: str, usia: int) -> int:
    cursor = conn.cursor()
    
    # Gunakan OUTPUT INSERTED.id untuk mendapatkan ID yang baru dibuat
    cursor.execute(
        """
        INSERT INTO pengguna (nama, email, usia)
        OUTPUT INSERTED.id
        VALUES (?, ?, ?)
        """,
        (nama, email, usia)
    )
    
    baris = cursor.fetchone()
    id_baru = baris[0] if baris else None
    cursor.close()
    return id_baru

def tambah_banyak_pengguna(conn, daftar: list[tuple]) -> int:
    cursor = conn.cursor()
    cursor.executemany(
        "INSERT INTO pengguna (nama, email, usia) VALUES (?, ?, ?)",
        daftar
    )
    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),
    ]
    tambah_banyak_pengguna(conn, data_baru)

Read Data #

def ambil_semua_pengguna(conn) -> list[dict]:
    cursor = conn.cursor()
    cursor.execute("""
        SELECT id, nama, email, usia, aktif
        FROM pengguna
        WHERE aktif = 1
        ORDER BY dibuat_pada DESC
    """)
    
    kolom = [desc[0] for desc in cursor.description]
    hasil = [dict(zip(kolom, baris)) for baris in cursor.fetchall()]
    cursor.close()
    return hasil

def ambil_pengguna_by_id(conn, pengguna_id: int) -> dict | None:
    cursor = conn.cursor()
    cursor.execute(
        "SELECT id, nama, email, usia FROM pengguna WHERE id = ?",
        (pengguna_id,)
    )
    baris = cursor.fetchone()
    if not baris:
        return None
    
    kolom = [desc[0] for desc in cursor.description]
    cursor.close()
    return dict(zip(kolom, baris))

# Pagination dengan OFFSET-FETCH (SQL Server 2012+)
def ambil_pengguna_halaman(conn, halaman: int = 1, per_halaman: int = 10) -> list[dict]:
    cursor = conn.cursor()
    offset = (halaman - 1) * per_halaman
    
    cursor.execute("""
        SELECT id, nama, email, usia
        FROM pengguna
        ORDER BY id
        OFFSET ? ROWS
        FETCH NEXT ? ROWS ONLY
    """, (offset, per_halaman))
    
    kolom = [desc[0] for desc in cursor.description]
    hasil = [dict(zip(kolom, baris)) for baris in 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']}")

Update dan Delete #

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

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

Stored Procedure #

Memanggil stored procedure adalah fitur yang sering dipakai di lingkungan enterprise SQL Server. pyodbc mendukungnya melalui sintaks EXEC atau menggunakan callproc pada beberapa driver.

-- Buat stored procedure di SQL Server terlebih dahulu
CREATE PROCEDURE GetPenggunaByEmail
    @Email NVARCHAR(150)
AS
BEGIN
    SELECT id, nama, email, usia
    FROM pengguna
    WHERE email = @Email
END

def panggil_sp_get_pengguna(conn, email: str) -> dict | None:
    cursor = conn.cursor()
    
    # Cara 1: menggunakan EXEC
    cursor.execute("EXEC GetPenggunaByEmail ?", (email,))
    baris = cursor.fetchone()
    
    if not baris:
        return None
    
    kolom = [desc[0] for desc in cursor.description]
    cursor.close()
    return dict(zip(kolom, baris))

with db_connection() as conn:
    pengguna = panggil_sp_get_pengguna(conn, "[email protected]")
    if pengguna:
        print(f"Ditemukan: {pengguna['nama']}")

Transaksi #

def proses_order(conn, pengguna_id: int, produk_id: int, jumlah: int) -> int | None:
    cursor = conn.cursor()
    
    try:
        # Cek stok dengan row-level lock
        cursor.execute(
            "SELECT stok FROM produk WITH (UPDLOCK) WHERE id = ?",
            (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
        cursor.execute(
            "UPDATE produk SET stok = stok - ? WHERE id = ?",
            (jumlah, produk_id)
        )
        
        # Buat record order, ambil ID baru via OUTPUT
        cursor.execute(
            """
            INSERT INTO orders (pengguna_id, produk_id, jumlah, total)
            OUTPUT INSERTED.id
            SELECT ?, ?, ?, (harga * ?) FROM produk WHERE id = ?
            """,
            (pengguna_id, produk_id, jumlah, jumlah, produk_id)
        )
        
        order_id = cursor.fetchone()[0]
        conn.commit()
        return order_id
        
    except Exception as e:
        conn.rollback()
        print(f"Transaksi dibatalkan: {e}")
        return None
    finally:
        cursor.close()

Penanganan Error #

import pyodbc

def tambah_pengguna_aman(conn, nama: str, email: str) -> int | None:
    cursor = conn.cursor()
    try:
        cursor.execute(
            """
            INSERT INTO pengguna (nama, email)
            OUTPUT INSERTED.id
            VALUES (?, ?)
            """,
            (nama, email)
        )
        conn.commit()
        baris = cursor.fetchone()
        return baris[0] if baris else None
    
    except pyodbc.IntegrityError as e:
        conn.rollback()
        # Error 2627: Violation of UNIQUE constraint
        # Error 2601: Cannot insert duplicate key
        if "2627" in str(e) or "2601" in str(e):
            print(f"Email '{email}' sudah terdaftar.")
        else:
            print(f"Integrity error: {e}")
        return None
    
    except pyodbc.Error as e:
        conn.rollback()
        sqlstate = e.args[0]
        print(f"Database error [{sqlstate}]: {e}")
        return None
    
    finally:
        cursor.close()

Ringkasan #

Placeholder ? bukan %s — pyodbc menggunakan tanda tanya ? sebagai placeholder parameter, berbeda dari MySQL (%s) dan PostgreSQL (%s atau $1).
OUTPUT INSERTED.id — gunakan klausa ini untuk mendapatkan ID baris yang baru di-insert, menggantikan lastrowid yang tidak selalu andal di pyodbc.
NVARCHAR bukan VARCHAR — selalu gunakan NVARCHAR untuk kolom string agar mendukung karakter Unicode (termasuk huruf Indonesia beraksen).
IDENTITY(1,1) — padanan AUTO_INCREMENT MySQL di MSSQL untuk kolom auto-increment.
OFFSET-FETCH — gunakan untuk pagination di SQL Server 2012+, menggantikan LIMIT/OFFSET yang ada di MySQL/PostgreSQL.
WITH (UPDLOCK) — gunakan hint ini saat melakukan read-before-update dalam transaksi untuk mencegah race condition.
TrustServerCertificate=yes — tambahkan di connection string untuk koneksi lokal/development; di produksi gunakan sertifikat yang valid.
Jangan hardcode kredensial — selalu baca host, user, password dari environment variable.
autocommit = False — set secara eksplisit setelah koneksi dibuat untuk kontrol penuh atas transaksi.

← Sebelumnya: MySQL Berikutnya: Oracle →