PostgreSQL #

PostgreSQL adalah database relasional open-source paling canggih yang tersedia saat ini — mendukung JSON, array, full-text search, tipe data kustom, dan ekstensi seperti PostGIS untuk data geospasial. Di kalangan developer Python modern, PostgreSQL adalah pilihan utama untuk aplikasi web dan API, terutama karena integrasinya yang mulus dengan Django ORM, SQLAlchemy, dan FastAPI. Library psycopg2 adalah driver PostgreSQL paling populer untuk Python, mengikuti standar DB-API 2.0 dengan ekstensi-ekstensi yang memanfaatkan fitur khas PostgreSQL.

Instalasi #

pip install psycopg2-binary

psycopg2-binary menyertakan binary pre-compiled sehingga tidak perlu menginstal PostgreSQL development headers. Untuk lingkungan produksi, gunakan psycopg2 (tanpa -binary) yang dikompilasi dari source — lebih stabil dan disarankan oleh maintainer library. Jika ingin menggunakan versi terbaru (async-native), pertimbangkan psycopg (psycopg3).

Membuat Koneksi #

psycopg2 mendukung dua format koneksi: keyword arguments atau connection string (DSN). Placeholder parameter di psycopg2 menggunakan %s seperti MySQL, tapi bukan format string Python — ini adalah placeholder yang diproses oleh driver secara aman.

import psycopg2
import os

# ANTI-PATTERN: hardcode kredensial di kode
conn = psycopg2.connect(
    host="localhost",
    database="myapp",
    user="postgres",
    password="rahasia123"  # ✗ -- jangan lakukan ini
)

# BENAR: baca dari environment variable
def get_connection() -> psycopg2.extensions.connection:
    return psycopg2.connect(
        host=os.getenv("PG_HOST", "localhost"),
        port=int(os.getenv("PG_PORT", "5432")),
        dbname=os.getenv("PG_DB", "myapp"),
        user=os.getenv("PG_USER", "postgres"),
        password=os.getenv("PG_PASSWORD", ""),
        connect_timeout=10,
        options="-c timezone=Asia/Jakarta"  # set timezone per koneksi
    )

# Atau menggunakan DSN string (berguna untuk cloud/managed database)
def get_connection_dsn() -> psycopg2.extensions.connection:
    dsn = os.getenv("DATABASE_URL")
    # Format: postgresql://user:password@host:port/dbname
    return psycopg2.connect(dsn)

Context Manager untuk Koneksi #

import psycopg2
from contextlib import contextmanager
import os

@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("PostgreSQL:", versi[0][:40])

Membuat Tabel #

PostgreSQL memiliki tipe data yang lebih kaya dari database lain — SERIAL/BIGSERIAL untuk auto-increment, TEXT untuk string tak terbatas, JSONB untuk JSON, dan ARRAY untuk array native.

def create_tables(conn):
    cursor = conn.cursor()
    
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS pengguna (
            id          BIGSERIAL PRIMARY KEY,
            nama        TEXT            NOT NULL,
            email       TEXT            NOT NULL UNIQUE,
            usia        SMALLINT,
            aktif       BOOLEAN         DEFAULT TRUE,
            metadata    JSONB           DEFAULT '{}',
            tag         TEXT[],
            dibuat_pada TIMESTAMPTZ     DEFAULT NOW()
        )
    """)
    
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS produk (
            id        BIGSERIAL PRIMARY KEY,
            nama      TEXT            NOT NULL,
            harga     NUMERIC(15, 2)  NOT NULL,
            stok      INTEGER         DEFAULT 0,
            kategori  TEXT,
            atribut   JSONB           DEFAULT '{}'
        )
    """)
    
    # Index untuk performa query umum
    cursor.execute("""
        CREATE INDEX IF NOT EXISTS idx_pengguna_email ON pengguna (email)
    """)
    cursor.execute("""
        CREATE INDEX IF NOT EXISTS idx_pengguna_aktif ON pengguna (aktif)
        WHERE aktif = TRUE
    """)
    
    conn.commit()
    cursor.close()
    print("Tabel berhasil dibuat.")

with db_connection() as conn:
    create_tables(conn)

Tipe data khas PostgreSQL yang tidak ada di MySQL/MSSQL:
TEXT — string tanpa batas panjang (tidak perlu VARCHAR(n) kecuali ingin membatasi)
BIGSERIAL / SERIAL — auto-increment (setara BIGINT AUTO_INCREMENT)
BOOLEAN — tipe boolean native (bukan BIT atau NUMBER(1))
JSONB — JSON tersimpan dalam format binary, bisa di-index dan di-query
TEXT[] — array tipe apapun native di PostgreSQL
TIMESTAMPTZ — timestamp dengan timezone (rekomendasi untuk aplikasi multi-timezone)
UUID — tipe UUID native

Operasi CRUD #

Insert Data #

import psycopg2.extras

def tambah_pengguna(conn, nama: str, email: str, usia: int, tag: list[str] = None) -> int:
    cursor = conn.cursor()
    
    # RETURNING id -- PostgreSQL idiom untuk mendapat ID setelah insert
    cursor.execute(
        """
        INSERT INTO pengguna (nama, email, usia, tag)
        VALUES (%s, %s, %s, %s)
        RETURNING id
        """,
        (nama, email, usia, tag or [])
    )
    
    id_baru = cursor.fetchone()[0]
    cursor.close()
    return id_baru

def tambah_banyak_pengguna(conn, daftar: list[tuple]) -> int:
    cursor = conn.cursor()
    
    # execute_values -- jauh lebih efisien dari executemany untuk batch insert
    psycopg2.extras.execute_values(
        cursor,
        "INSERT INTO pengguna (nama, email, usia) VALUES %s",
        daftar
    )
    
    jumlah = cursor.rowcount
    cursor.close()
    return jumlah

with db_connection() as conn:
    id1 = tambah_pengguna(conn, "Budi Santoso", "[email protected]", 28, ["python", "backend"])
    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.")

Read Data #

def ambil_semua_pengguna(conn) -> list[dict]:
    # RealDictCursor -- setiap baris langsung jadi dict
    cursor = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
    
    cursor.execute("""
        SELECT id, nama, email, usia, aktif, tag, dibuat_pada
        FROM pengguna
        WHERE aktif = TRUE
        ORDER BY dibuat_pada DESC
    """)
    
    hasil = cursor.fetchall()
    cursor.close()
    return [dict(baris) for baris in hasil]

def ambil_pengguna_by_id(conn, pengguna_id: int) -> dict | None:
    cursor = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
    cursor.execute(
        "SELECT * FROM pengguna WHERE id = %s",
        (pengguna_id,)
    )
    baris = cursor.fetchone()
    cursor.close()
    return dict(baris) if baris else None

# Pagination dengan LIMIT OFFSET
def ambil_pengguna_halaman(conn, halaman: int = 1, per_halaman: int = 10) -> list[dict]:
    cursor = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
    offset = (halaman - 1) * per_halaman
    
    cursor.execute("""
        SELECT id, nama, email, usia
        FROM pengguna
        WHERE aktif = TRUE
        ORDER BY id
        LIMIT %s OFFSET %s
    """, (per_halaman, offset))
    
    hasil = cursor.fetchall()
    cursor.close()
    return [dict(baris) for baris in hasil]

# Query JSONB -- fitur khas PostgreSQL
def ambil_pengguna_by_metadata(conn, kunci: str, nilai: str) -> list[dict]:
    cursor = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
    
    # Operator -> untuk akses field JSONB
    cursor.execute(
        "SELECT * FROM pengguna WHERE metadata->>%s = %s",
        (kunci, nilai)
    )
    
    hasil = cursor.fetchall()
    cursor.close()
    return [dict(baris) for baris in hasil]

with db_connection() as conn:
    semua = ambil_semua_pengguna(conn)
    for p in semua:
        print(f"[{p['id']}] {p['nama']} — tag: {p['tag']}")

Update dan Delete #

def update_pengguna(conn, pengguna_id: int, nama: str, usia: int) -> bool:
    cursor = conn.cursor()
    cursor.execute(
        "UPDATE pengguna SET nama = %s, usia = %s WHERE id = %s",
        (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

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

UPSERT dengan ON CONFLICT #

INSERT ... ON CONFLICT adalah fitur PostgreSQL untuk melakukan insert atau update sekaligus — sangat berguna untuk sinkronisasi data dan operasi idempoten.

def upsert_pengguna(conn, email: str, nama: str, usia: int) -> int:
    cursor = conn.cursor()
    
    # Jika email sudah ada, update nama dan usia
    # Jika belum ada, insert baru
    cursor.execute(
        """
        INSERT INTO pengguna (nama, email, usia)
        VALUES (%s, %s, %s)
        ON CONFLICT (email) DO UPDATE SET
            nama  = EXCLUDED.nama,
            usia  = EXCLUDED.usia
        RETURNING id
        """,
        (nama, email, usia)
    )
    
    id_pengguna = cursor.fetchone()[0]
    cursor.close()
    return id_pengguna

# ON CONFLICT DO NOTHING -- abaikan jika sudah ada
def tambah_pengguna_jika_belum_ada(conn, email: str, nama: str) -> int | None:
    cursor = conn.cursor()
    cursor.execute(
        """
        INSERT INTO pengguna (nama, email)
        VALUES (%s, %s)
        ON CONFLICT (email) DO NOTHING
        RETURNING id
        """,
        (nama, email)
    )
    baris = cursor.fetchone()
    cursor.close()
    return baris[0] if baris else None

with db_connection() as conn:
    id1 = upsert_pengguna(conn, "[email protected]", "Budi Santoso", 28)
    id2 = upsert_pengguna(conn, "[email protected]", "Budi Santoso Wijaya", 29)
    print(f"ID sama: {id1 == id2}")  # True -- update, bukan insert baru

Connection Pool #

Untuk aplikasi web, gunakan psycopg2.pool agar koneksi tidak dibuat ulang setiap request:

import psycopg2.pool
import os

# Buat pool saat aplikasi start -- thread-safe pool
connection_pool = psycopg2.pool.ThreadedConnectionPool(
    minconn=2,
    maxconn=10,
    host=os.getenv("PG_HOST", "localhost"),
    dbname=os.getenv("PG_DB", "myapp"),
    user=os.getenv("PG_USER", "postgres"),
    password=os.getenv("PG_PASSWORD", "")
)

from contextlib import contextmanager

@contextmanager
def pooled_connection():
    conn = connection_pool.getconn()
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        connection_pool.putconn(conn)  # kembalikan ke pool

# Penggunaan identik dengan koneksi biasa
def ambil_pengguna_pool(pengguna_id: int) -> dict | None:
    with pooled_connection() as conn:
        cursor = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
        cursor.execute("SELECT * FROM pengguna WHERE id = %s", (pengguna_id,))
        baris = cursor.fetchone()
        return dict(baris) if baris else None

Transaksi #

def transfer_saldo(conn, dari_id: int, ke_id: int, jumlah: float) -> None:
    cursor = conn.cursor()
    
    try:
        # Kunci kedua baris sekaligus untuk mencegah deadlock
        cursor.execute("""
            SELECT id, saldo FROM akun
            WHERE id = ANY(%s)
            ORDER BY id          -- urutan konsisten cegah deadlock
            FOR UPDATE
        """, ([dari_id, ke_id],))
        
        akun = {baris[0]: baris[1] for baris in cursor.fetchall()}
        
        if akun.get(dari_id, 0) < jumlah:
            raise ValueError(f"Saldo tidak mencukupi. Tersedia: {akun.get(dari_id, 0)}")
        
        cursor.execute(
            "UPDATE akun SET saldo = saldo - %s WHERE id = %s",
            (jumlah, dari_id)
        )
        cursor.execute(
            "UPDATE akun SET saldo = saldo + %s WHERE id = %s",
            (jumlah, ke_id)
        )
        
        conn.commit()
        print(f"Transfer {jumlah:,.0f} berhasil.")
        
    except Exception as e:
        conn.rollback()
        print(f"Transaksi dibatalkan: {e}")
        raise
    finally:
        cursor.close()

Penanganan Error #

import psycopg2
from psycopg2 import errorcodes, errors

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) RETURNING id",
            (nama, email)
        )
        conn.commit()
        return cursor.fetchone()[0]
    
    except errors.UniqueViolation:
        conn.rollback()
        print(f"Email '{email}' sudah terdaftar.")
        return None
    
    except errors.NotNullViolation as e:
        conn.rollback()
        print(f"Field wajib tidak boleh kosong: {e.diag.column_name}")
        return None
    
    except psycopg2.DatabaseError as e:
        conn.rollback()
        print(f"Database error [{e.pgcode}]: {e.pgerror}")
        return None
    
    finally:
        cursor.close()

Ringkasan #

Placeholder %s — psycopg2 menggunakan %s sebagai placeholder, tapi ini bukan % format string Python; selalu gunakan tuple/list nilai terpisah sebagai argumen kedua execute().
RETURNING — gunakan klausa ini setelah INSERT/UPDATE/DELETE untuk mendapat nilai kolom yang terpengaruh tanpa perlu query tambahan.
RealDictCursor — gunakan cursor_factory=psycopg2.extras.RealDictCursor agar baris hasil query langsung jadi dict dengan nama kolom sebagai key.
execute_values() — gunakan untuk bulk insert yang jauh lebih efisien daripada loop executemany().
ON CONFLICT — manfaatkan INSERT ... ON CONFLICT DO UPDATE (UPSERT) untuk operasi insert-atau-update yang idempoten.
JSONB — simpan data semi-struktural langsung di PostgreSQL dengan tipe JSONB yang bisa di-index dan di-query menggunakan operator -> dan ->>.
TEXT[] — PostgreSQL mendukung array native; tidak perlu tabel relasi terpisah untuk data list sederhana.
ThreadedConnectionPool — gunakan connection pool untuk aplikasi web agar tidak membuat koneksi baru setiap request.
errors.UniqueViolation — tangani error spesifik PostgreSQL via modul psycopg2.errors daripada mencocokkan string pesan error.

← Sebelumnya: Oracle Berikutnya: SQL Alchemy →