SQL Alchemy #

SQLAlchemy adalah library database Python yang paling lengkap dan paling banyak digunakan — bukan sekadar ORM, tapi sebuah toolkit database dengan dua layer berbeda yang bisa dipakai secara terpisah atau bersama. SQLAlchemy Core memberikan abstraksi SQL yang ekspresif tanpa menyembunyikan query, sementara SQLAlchemy ORM memetakan tabel database ke kelas Python sehingga kamu bekerja dengan objek, bukan SQL mentah. Memahami perbedaan kedua layer ini, cara mendefinisikan relasi, dan cara mengelola session adalah pondasi untuk menggunakan SQLAlchemy secara efektif di proyek nyata.

Instalasi #

pip install sqlalchemy

# Tambahkan driver database yang digunakan:
pip install psycopg2-binary          # PostgreSQL
pip install mysql-connector-python   # MySQL
pip install pyodbc                   # MSSQL
# SQLite sudah termasuk di Python standar

Core vs ORM — Dua Layer SQLAlchemy #

Sebelum menulis kode, penting memahami dua cara menggunakan SQLAlchemy dan kapan memilih masing-masing:

SQLAlchemy Core
  ├── Bekerja dengan tabel, kolom, dan ekspresi SQL secara eksplisit
  ├── Kamu masih "berpikir SQL" tapi dengan Python API
  ├── Performa lebih baik untuk query kompleks dan bulk operation
  └── Cocok untuk: data pipeline, reporting, query SQL yang kompleks

SQLAlchemy ORM
  ├── Memetakan tabel ke kelas Python (model)
  ├── Operasi CRUD via objek Python, bukan SQL langsung
  ├── Fitur: lazy/eager loading, relationship, identity map
  └── Cocok untuk: aplikasi web, API, domain yang kaya logika bisnis

SQLAlchemy 2.0 memperkenalkan API baru yang lebih konsisten antara Core dan ORM. Artikel ini menggunakan SQLAlchemy 2.x — jika kamu masih memakai 1.x, beberapa sintaks berbeda terutama cara menjalankan query dan cara membuat session.

Setup Engine dan Koneksi #

Engine adalah titik masuk ke database — mengelola connection pool dan mengeksekusi query. Satu engine biasanya dibuat satu kali saat aplikasi start.

from sqlalchemy import create_engine, text
import os

# Format URL: dialect+driver://user:password@host:port/database
DATABASE_URL = os.getenv(
    "DATABASE_URL",
    "postgresql+psycopg2://postgres:password@localhost:5432/myapp"
)

# ANTI-PATTERN: hardcode URL database di kode
engine = create_engine("postgresql://postgres:rahasia@localhost/myapp")  # ✗

# BENAR: baca dari environment variable
engine = create_engine(
    DATABASE_URL,
    echo=False,           # True untuk log semua SQL -- berguna saat debug
    pool_size=5,          # jumlah koneksi yang dipertahankan
    max_overflow=10,      # koneksi tambahan saat pool penuh
    pool_pre_ping=True    # cek koneksi sebelum dipakai (cegah stale connection)
)

# Contoh URL untuk berbagai database
# SQLite (file lokal):
engine_sqlite = create_engine("sqlite:///myapp.db")

# MySQL:
engine_mysql = create_engine(
    "mysql+mysqlconnector://user:password@localhost:3306/myapp"
)

# MSSQL:
engine_mssql = create_engine(
    "mssql+pyodbc://sa:password@localhost/myapp"
    "?driver=ODBC+Driver+18+for+SQL+Server&TrustServerCertificate=yes"
)

# Tes koneksi
with engine.connect() as conn:
    hasil = conn.execute(text("SELECT 1"))
    print("Koneksi berhasil:", hasil.fetchone())

Mendefinisikan Model (ORM) #

Model adalah kelas Python yang merepresentasikan tabel database. Sejak SQLAlchemy 2.0, cara modern menggunakan DeclarativeBase dan type annotations.

from sqlalchemy import (
    String, Integer, Boolean, Numeric, Text,
    DateTime, ForeignKey, func
)
from sqlalchemy.orm import (
    DeclarativeBase, Mapped, mapped_column,
    relationship, Session
)
from datetime import datetime
from typing import Optional, List

class Base(DeclarativeBase):
    pass

class Pengguna(Base):
    __tablename__ = "pengguna"
    
    id:          Mapped[int]           = mapped_column(Integer, primary_key=True, autoincrement=True)
    nama:        Mapped[str]           = mapped_column(String(100), nullable=False)
    email:       Mapped[str]           = mapped_column(String(150), nullable=False, unique=True)
    usia:        Mapped[Optional[int]] = mapped_column(Integer)
    aktif:       Mapped[bool]          = mapped_column(Boolean, default=True)
    dibuat_pada: Mapped[datetime]      = mapped_column(DateTime, server_default=func.now())
    
    # Relasi one-to-many ke Order
    orders: Mapped[List["Order"]] = relationship("Order", back_populates="pengguna")
    
    def __repr__(self) -> str:
        return f"<Pengguna(id={self.id}, nama='{self.nama}', email='{self.email}')>"

class Produk(Base):
    __tablename__ = "produk"
    
    id:       Mapped[int]           = mapped_column(Integer, primary_key=True, autoincrement=True)
    nama:     Mapped[str]           = mapped_column(String(200), nullable=False)
    harga:    Mapped[float]         = mapped_column(Numeric(15, 2), nullable=False)
    stok:     Mapped[int]           = mapped_column(Integer, default=0)
    kategori: Mapped[Optional[str]] = mapped_column(String(100))
    
    orders: Mapped[List["Order"]] = relationship("Order", back_populates="produk")

class Order(Base):
    __tablename__ = "orders"
    
    id:          Mapped[int]      = mapped_column(Integer, primary_key=True, autoincrement=True)
    pengguna_id: Mapped[int]      = mapped_column(ForeignKey("pengguna.id"), nullable=False)
    produk_id:   Mapped[int]      = mapped_column(ForeignKey("produk.id"), nullable=False)
    jumlah:      Mapped[int]      = mapped_column(Integer, nullable=False)
    total:       Mapped[float]    = mapped_column(Numeric(15, 2), nullable=False)
    dibuat_pada: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
    
    pengguna: Mapped["Pengguna"] = relationship("Pengguna", back_populates="orders")
    produk:   Mapped["Produk"]   = relationship("Produk", back_populates="orders")

# Buat semua tabel yang belum ada
Base.metadata.create_all(engine)

Session Management #

Session adalah unit kerja ORM — mengelola objek yang dipantau, mengumpulkan perubahan, dan mengirimkan ke database sebagai satu transaksi.

from sqlalchemy.orm import sessionmaker, Session
from contextlib import contextmanager

SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)

@contextmanager
def get_session():
    session = SessionLocal()
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

# ANTI-PATTERN: lupa menutup session
session = SessionLocal()
pengguna = session.get(Pengguna, 1)
# ... lupa session.close() -- koneksi leak!

# BENAR: selalu gunakan context manager
with get_session() as session:
    pengguna = session.get(Pengguna, 1)
    print(pengguna)

Operasi CRUD dengan ORM #

Create #

def tambah_pengguna(nama: str, email: str, usia: int) -> Pengguna:
    with get_session() as session:
        pengguna = Pengguna(nama=nama, email=email, usia=usia)
        session.add(pengguna)
        session.flush()      # kirim ke DB tapi belum commit -- id sudah tersedia
        session.refresh(pengguna)  # reload dari DB
        return pengguna

def tambah_banyak_pengguna(daftar: list[dict]) -> int:
    with get_session() as session:
        objek_list = [Pengguna(**data) for data in daftar]
        session.add_all(objek_list)
        return len(objek_list)

# Penggunaan
p = tambah_pengguna("Budi Santoso", "[email protected]", 28)
print(f"Pengguna baru: {p}")

tambah_banyak_pengguna([
    {"nama": "Sari Dewi",     "email": "[email protected]",  "usia": 25},
    {"nama": "Andi Prasetyo", "email": "[email protected]",  "usia": 32},
])

Read #

from sqlalchemy import select, and_, or_, desc, func

def ambil_pengguna_by_id(pengguna_id: int) -> Pengguna | None:
    with get_session() as session:
        # Cara 1: session.get() -- untuk PK lookup, cek identity map dulu
        return session.get(Pengguna, pengguna_id)

def ambil_semua_pengguna_aktif() -> list[Pengguna]:
    with get_session() as session:
        stmt = (
            select(Pengguna)
            .where(Pengguna.aktif == True)
            .order_by(desc(Pengguna.dibuat_pada))
        )
        return list(session.scalars(stmt))

# Select field tertentu saja
def ambil_nama_email_pengguna() -> list[dict]:
    with get_session() as session:
        stmt = select(Pengguna.id, Pengguna.nama, Pengguna.email)
        hasil = session.execute(stmt).all()
        return [{"id": r.id, "nama": r.nama, "email": r.email} for r in hasil]

# Filter kompleks
def cari_pengguna(kata_kunci: str = None, usia_min: int = None, usia_max: int = None):
    with get_session() as session:
        stmt = select(Pengguna).where(Pengguna.aktif == True)
        
        if kata_kunci:
            stmt = stmt.where(
                or_(
                    Pengguna.nama.ilike(f"%{kata_kunci}%"),
                    Pengguna.email.ilike(f"%{kata_kunci}%")
                )
            )
        if usia_min is not None:
            stmt = stmt.where(Pengguna.usia >= usia_min)
        if usia_max is not None:
            stmt = stmt.where(Pengguna.usia <= usia_max)
        
        stmt = stmt.order_by(Pengguna.nama).limit(50)
        return list(session.scalars(stmt))

# Pagination
def ambil_pengguna_halaman(halaman: int = 1, per_halaman: int = 10) -> list[Pengguna]:
    with get_session() as session:
        offset = (halaman - 1) * per_halaman
        stmt = (
            select(Pengguna)
            .where(Pengguna.aktif == True)
            .order_by(Pengguna.id)
            .offset(offset)
            .limit(per_halaman)
        )
        return list(session.scalars(stmt))

Update #

def update_pengguna(pengguna_id: int, **kwargs) -> bool:
    with get_session() as session:
        pengguna = session.get(Pengguna, pengguna_id)
        if not pengguna:
            return False
        
        # Update atribut yang diberikan
        for key, value in kwargs.items():
            if hasattr(pengguna, key):
                setattr(pengguna, key, value)
        
        return True  # commit otomatis via context manager

# Bulk update lebih efisien dengan Core
from sqlalchemy import update as sa_update

def nonaktifkan_pengguna_bulk(pengguna_ids: list[int]) -> int:
    with get_session() as session:
        stmt = (
            sa_update(Pengguna)
            .where(Pengguna.id.in_(pengguna_ids))
            .values(aktif=False)
        )
        hasil = session.execute(stmt)
        return hasil.rowcount

# Penggunaan
update_pengguna(1, nama="Budi Santoso Wijaya", usia=29)
nonaktifkan_pengguna_bulk([3, 4, 5])

Delete #

from sqlalchemy import delete as sa_delete

def hapus_pengguna(pengguna_id: int) -> bool:
    with get_session() as session:
        pengguna = session.get(Pengguna, pengguna_id)
        if not pengguna:
            return False
        session.delete(pengguna)
        return True

# Bulk delete
def hapus_pengguna_tidak_aktif() -> int:
    with get_session() as session:
        stmt = sa_delete(Pengguna).where(Pengguna.aktif == False)
        hasil = session.execute(stmt)
        return hasil.rowcount

Query Lanjutan #

Join Tabel #

from sqlalchemy.orm import joinedload, selectinload

# Inner join dengan ORM
def ambil_orders_dengan_detail() -> list[dict]:
    with get_session() as session:
        stmt = (
            select(Order, Pengguna.nama, Produk.nama)
            .join(Pengguna, Order.pengguna_id == Pengguna.id)
            .join(Produk, Order.produk_id == Produk.id)
            .order_by(desc(Order.dibuat_pada))
        )
        hasil = session.execute(stmt).all()
        return [
            {
                "order_id": r[0].id,
                "pengguna": r[1],
                "produk":   r[2],
                "jumlah":   r[0].jumlah,
                "total":    float(r[0].total)
            }
            for r in hasil
        ]

# Eager loading relationship -- hindari N+1 query
def ambil_pengguna_dengan_orders() -> list[Pengguna]:
    with get_session() as session:
        stmt = (
            select(Pengguna)
            .options(selectinload(Pengguna.orders))  # load orders sekaligus
            .where(Pengguna.aktif == True)
        )
        pengguna_list = list(session.scalars(stmt))
        
        for p in pengguna_list:
            # orders sudah ter-load, tidak ada query tambahan
            print(f"{p.nama}: {len(p.orders)} order")
        
        return pengguna_list

Aggregasi dan Group By #

from sqlalchemy import func, Integer

def statistik_pengguna() -> dict:
    with get_session() as session:
        stmt = select(
            func.count(Pengguna.id).label("total"),
            func.avg(Pengguna.usia).label("rata_usia"),
            func.min(Pengguna.usia).label("usia_min"),
            func.max(Pengguna.usia).label("usia_max"),
        ).where(Pengguna.aktif == True)
        
        hasil = session.execute(stmt).one()
        return {
            "total":    hasil.total,
            "rata_usia": round(float(hasil.rata_usia or 0), 1),
            "usia_min": hasil.usia_min,
            "usia_max": hasil.usia_max,
        }

def total_order_per_pengguna() -> list[dict]:
    with get_session() as session:
        stmt = (
            select(
                Pengguna.nama,
                func.count(Order.id).label("jumlah_order"),
                func.sum(Order.total).label("total_belanja")
            )
            .join(Order, Pengguna.id == Order.pengguna_id)
            .group_by(Pengguna.id, Pengguna.nama)
            .having(func.count(Order.id) > 0)
            .order_by(desc("total_belanja"))
        )
        hasil = session.execute(stmt).all()
        return [
            {"nama": r.nama, "jumlah_order": r.jumlah_order, "total": float(r.total_belanja)}
            for r in hasil
        ]

Subquery #

from sqlalchemy import subquery

def pengguna_dengan_order_mahal(batas_harga: float) -> list[Pengguna]:
    with get_session() as session:
        # Subquery: id pengguna yang punya order di atas batas
        sub = (
            select(Order.pengguna_id)
            .where(Order.total > batas_harga)
            .distinct()
            .scalar_subquery()
        )
        
        stmt = select(Pengguna).where(Pengguna.id.in_(sub))
        return list(session.scalars(stmt))

Menjalankan SQL Mentah #

Kadang query ORM terlalu kompleks dan lebih mudah ditulis langsung sebagai SQL. SQLAlchemy tetap mendukung ini dengan aman melalui text().

from sqlalchemy import text

def jalankan_sql_mentah(query_str: str, params: dict = None) -> list[dict]:
    with engine.connect() as conn:
        hasil = conn.execute(text(query_str), params or {})
        kolom = hasil.keys()
        return [dict(zip(kolom, baris)) for baris in hasil.fetchall()]

# Penggunaan
hasil = jalankan_sql_mentah(
    "SELECT id, nama, email FROM pengguna WHERE usia > :usia_min ORDER BY nama",
    {"usia_min": 25}
)
for p in hasil:
    print(p["nama"])

# Stored procedure via text()
def panggil_stored_procedure(nama_proc: str, params: dict) -> list[dict]:
    with engine.connect() as conn:
        # Sintaks tergantung database (ini contoh PostgreSQL)
        hasil = conn.execute(text(f"SELECT * FROM {nama_proc}(:param1)"), params)
        return [dict(zip(hasil.keys(), baris)) for baris in hasil.fetchall()]

Migrasi dengan Alembic #

Alembic adalah tool migrasi database resmi untuk SQLAlchemy — melacak perubahan skema dan menerapkannya secara incremental.

pip install alembic

# Inisialisasi di direktori project
alembic init alembic

# Buat file migrasi otomatis dari perubahan model
alembic revision --autogenerate -m "tambah_tabel_pengguna"

# Terapkan migrasi ke database
alembic upgrade head

# Rollback satu versi
alembic downgrade -1

# alembic/env.py -- konfigurasi target metadata
from myapp.models import Base  # import Base yang berisi semua model

target_metadata = Base.metadata

Ringkasan #

Core vs ORM — gunakan Core untuk query kompleks dan bulk operation yang butuh performa; gunakan ORM untuk aplikasi dengan domain yang kaya logika bisnis.
create_engine() sekali saja — engine mengelola connection pool; buat satu instance saat aplikasi start, bukan per-request.
pool_pre_ping=True — aktifkan agar engine mengecek koneksi sebelum dipakai dan menghindari stale connection error.
Context manager untuk session — selalu gunakan with get_session() as session agar commit/rollback dan penutupan session terjamin.
session.get() untuk PK — lebih efisien dari select().where() karena mengecek identity map sebelum ke database.
selectinload() / joinedload() — gunakan eager loading untuk relasi agar tidak terjadi N+1 query saat mengakses relationship.
session.add_all() — gunakan untuk insert banyak objek sekaligus dalam satu transaksi.
Bulk update/delete via Core — untuk memperbarui atau menghapus banyak baris sekaligus, gunakan sa_update() dan sa_delete() yang jauh lebih efisien daripada loop ORM.
text() untuk SQL mentah — ketika query terlalu kompleks untuk ORM, gunakan text() dengan parameter bernama untuk tetap aman dari SQL injection.
Alembic untuk migrasi — jangan ubah skema database secara manual; gunakan Alembic agar perubahan terlacak dan bisa di-rollback.

← Sebelumnya: PostgreSQL Berikutnya: Django ORM →