Oracle #

Oracle Database adalah database relasional enterprise kelas atas yang dominan di industri perbankan, telekomunikasi, dan pemerintahan. Python berinteraksi dengan Oracle melalui library python-oracledb — penerus resmi dari cx_Oracle yang kini sudah deprecated. Salah satu keunikan Oracle dari database lain adalah penggunaan placeholder bernama (:nama bukan ? atau %s), konsep SEQUENCE untuk auto-increment, klausa RETURNING INTO, dan mode koneksi thin vs thick yang perlu dipahami sebelum memulai. Menguasai perbedaan ini akan menghindarkan kamu dari kebingungan saat beralih dari MySQL atau PostgreSQL ke Oracle.

Instalasi #

pip install python-oracledb

python-oracledb hadir dalam dua mode:

Mode Thin  (default)
  ├── Pure Python, tidak perlu Oracle Client
  ├── Cukup pip install, langsung bisa digunakan
  └── Mendukung sebagian besar fitur untuk aplikasi umum

Mode Thick
  ├── Membutuhkan Oracle Instant Client terinstal
  ├── Perlu diinisialisasi eksplisit: oracledb.init_oracle_client()
  └── Mendukung fitur lanjutan: Advanced Queuing, DRCP, dll.

Untuk sebagian besar aplikasi baru, mode thin sudah cukup dan jauh lebih mudah di-setup. Gunakan mode thick hanya jika kamu membutuhkan fitur Oracle lanjutan yang tidak tersedia di mode thin, atau terhubung ke Oracle Database versi lama (sebelum 12.1).

Membuat Koneksi #

Oracle menggunakan DSN (Data Source Name) atau connection string yang berbeda dari database lain. Ada dua format umum: menggunakan Easy Connect String atau TNS alias.

import oracledb
import os

# ANTI-PATTERN: hardcode kredensial di kode
conn = oracledb.connect(
    user="admin",
    password="rahasia123",  # ✗ -- jangan lakukan ini
    dsn="localhost/ORCLPDB1"
)

# BENAR: baca dari environment variable
def get_connection() -> oracledb.Connection:
    user     = os.getenv("ORACLE_USER", "admin")
    password = os.getenv("ORACLE_PASSWORD")
    host     = os.getenv("ORACLE_HOST", "localhost")
    port     = os.getenv("ORACLE_PORT", "1521")
    service  = os.getenv("ORACLE_SERVICE", "ORCLPDB1")
    
    # Easy Connect String format: host:port/service_name
    dsn = f"{host}:{port}/{service}"
    
    return oracledb.connect(user=user, password=password, dsn=dsn)

conn = get_connection()
print("Koneksi berhasil, versi Oracle:", conn.version)
conn.close()

# Koneksi menggunakan TNS alias (dari tnsnames.ora)
conn = oracledb.connect(
    user=os.getenv("ORACLE_USER"),
    password=os.getenv("ORACLE_PASSWORD"),
    dsn="MYDB_PROD"  # alias yang didefinisikan di tnsnames.ora
)

# Mode thick (jika perlu fitur lanjutan)
oracledb.init_oracle_client(lib_dir="/opt/oracle/instantclient_21_9")
conn = oracledb.connect(user="admin", password="...", dsn="localhost/ORCLPDB1")

Context Manager untuk Koneksi #

import oracledb
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 SYSDATE FROM DUAL")
    tanggal = cursor.fetchone()
    print("Tanggal server Oracle:", tanggal[0])

Membuat Tabel dan Sequence #

Oracle tidak memiliki AUTO_INCREMENT — sebagai gantinya menggunakan SEQUENCE dan TRIGGER, atau sejak Oracle 12c menggunakan kolom GENERATED ALWAYS AS IDENTITY.

def create_tables(conn):
    cursor = conn.cursor()
    
    # Hapus tabel jika sudah ada (untuk dev/testing)
    for obj in ["pengguna", "seq_pengguna"]:
        try:
            if obj.startswith("seq_"):
                cursor.execute(f"DROP SEQUENCE {obj}")
            else:
                cursor.execute(f"DROP TABLE {obj} PURGE")
        except oracledb.DatabaseError:
            pass  # abaikan jika belum ada
    
    # Buat sequence untuk auto-increment (Oracle 11g dan sebelumnya)
    cursor.execute("""
        CREATE SEQUENCE seq_pengguna
            START WITH 1
            INCREMENT BY 1
            NOCACHE
            NOCYCLE
    """)
    
    # Buat tabel -- gunakan IDENTITY untuk Oracle 12c+
    cursor.execute("""
        CREATE TABLE pengguna (
            id          NUMBER GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
            nama        NVARCHAR2(100)  NOT NULL,
            email       NVARCHAR2(150)  NOT NULL,
            usia        NUMBER(3),
            aktif       NUMBER(1)       DEFAULT 1,
            dibuat_pada TIMESTAMP       DEFAULT CURRENT_TIMESTAMP,
            CONSTRAINT uq_pengguna_email UNIQUE (email)
        )
    """)
    
    conn.commit()
    cursor.close()
    print("Tabel dan sequence berhasil dibuat.")

with db_connection() as conn:
    create_tables(conn)

Perbedaan penting Oracle dari MySQL/MSSQL:
NUMBER menggantikan INT, BIGINT, DECIMAL
NVARCHAR2 untuk string Unicode (rekomendasi Oracle modern)
TIMESTAMP menggantikan DATETIME
SYSDATE atau CURRENT_TIMESTAMP untuk waktu server
DUAL — tabel dummy Oracle untuk query tanpa tabel nyata: SELECT 1 FROM DUAL
Tidak ada BOOLEAN native — gunakan NUMBER(1) (0/1) atau CHAR(1) (Y/N)

Operasi CRUD #

Placeholder Bernama — Keunikan Oracle #

# ANTI-PATTERN: menggunakan ? (MSSQL) atau %s (MySQL)
cursor.execute("INSERT INTO pengguna (nama, email) VALUES (?, ?)", ("Budi", "[email protected]"))   # ✗
cursor.execute("INSERT INTO pengguna (nama, email) VALUES (%s, %s)", ("Budi", "[email protected]")) # ✗

# BENAR: Oracle menggunakan placeholder bernama dengan awalan titik dua
cursor.execute(
    "INSERT INTO pengguna (nama, email) VALUES (:nama, :email)",
    {"nama": "Budi", "email": "[email protected]"}   # ✓ -- dict dengan key sesuai placeholder
)

# Atau dengan tuple posisional (urutan harus sesuai)
cursor.execute(
    "INSERT INTO pengguna (nama, email) VALUES (:1, :2)",
    ("Budi", "[email protected]")   # ✓ -- :1, :2, :3, ...
)

Insert Data #

def tambah_pengguna(conn, nama: str, email: str, usia: int) -> int:
    cursor = conn.cursor()
    
    # RETURNING INTO untuk mendapatkan ID yang baru dibuat
    id_var = cursor.var(oracledb.NUMBER)
    
    cursor.execute(
        """
        INSERT INTO pengguna (nama, email, usia)
        VALUES (:nama, :email, :usia)
        RETURNING id INTO :id_baru
        """,
        {"nama": nama, "email": email, "usia": usia, "id_baru": id_var}
    )
    
    id_baru = int(id_var.getvalue()[0])
    cursor.close()
    return id_baru

def tambah_banyak_pengguna(conn, daftar: list[dict]) -> int:
    cursor = conn.cursor()
    cursor.executemany(
        "INSERT INTO pengguna (nama, email, usia) VALUES (:nama, :email, :usia)",
        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 = [
        {"nama": "Sari Dewi",     "email": "[email protected]",  "usia": 25},
        {"nama": "Andi Prasetyo", "email": "[email protected]",  "usia": 32},
        {"nama": "Rina Marlina",  "email": "[email protected]",  "usia": 29},
    ]
    jumlah = tambah_banyak_pengguna(conn, data_baru)
    print(f"{jumlah} pengguna ditambahkan.")

Read Data #

def ambil_semua_pengguna(conn) -> list[dict]:
    cursor = conn.cursor()
    
    # rowfactory -- konversi baris menjadi dict otomatis
    cursor.rowfactory = lambda *args: dict(zip(
        [col[0].lower() for col in cursor.description], args
    ))
    
    cursor.execute("""
        SELECT id, nama, email, usia, aktif
        FROM pengguna
        WHERE aktif = 1
        ORDER BY dibuat_pada DESC
        FETCH FIRST 100 ROWS ONLY
    """)
    
    hasil = cursor.fetchall()
    cursor.close()
    return hasil

def ambil_pengguna_by_id(conn, pengguna_id: int) -> dict | None:
    cursor = conn.cursor()
    cursor.rowfactory = lambda *args: dict(zip(
        [col[0].lower() for col in cursor.description], args
    ))
    
    cursor.execute(
        "SELECT id, nama, email, usia FROM pengguna WHERE id = :id",
        {"id": pengguna_id}
    )
    hasil = cursor.fetchone()
    cursor.close()
    return hasil

# Pagination dengan OFFSET-FETCH (Oracle 12c+)
def ambil_pengguna_halaman(conn, halaman: int = 1, per_halaman: int = 10) -> list[dict]:
    cursor = conn.cursor()
    cursor.rowfactory = lambda *args: dict(zip(
        [col[0].lower() for col in cursor.description], args
    ))
    offset = (halaman - 1) * per_halaman
    
    cursor.execute("""
        SELECT id, nama, email, usia
        FROM pengguna
        ORDER BY id
        OFFSET :offset ROWS
        FETCH NEXT :per_halaman ROWS ONLY
    """, {"offset": offset, "per_halaman": per_halaman})
    
    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']}")

Update dan Delete #

def update_pengguna(conn, pengguna_id: int, nama: str, usia: int) -> bool:
    cursor = conn.cursor()
    cursor.execute(
        "UPDATE pengguna SET nama = :nama, usia = :usia WHERE id = :id",
        {"nama": nama, "usia": usia, "id": 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 = :id", {"id": pengguna_id})
    berhasil = cursor.rowcount > 0
    cursor.close()
    return berhasil

Stored Procedure #

Oracle mendukung stored procedure dan function melalui PL/SQL. Pemanggilan dari Python menggunakan cursor.callproc() atau cursor.callfunc().

-- Buat stored procedure di Oracle terlebih dahulu
CREATE OR REPLACE PROCEDURE get_pengguna_by_email(
    p_email     IN  pengguna.email%TYPE,
    p_nama      OUT pengguna.nama%TYPE,
    p_usia      OUT pengguna.usia%TYPE,
    p_ditemukan OUT NUMBER
) AS
BEGIN
    SELECT nama, usia INTO p_nama, p_usia
    FROM pengguna
    WHERE email = p_email AND ROWNUM = 1;
    p_ditemukan := 1;
EXCEPTION
    WHEN NO_DATA_FOUND THEN
        p_ditemukan := 0;
END;

import oracledb

def cari_pengguna_by_email(conn, email: str) -> dict | None:
    cursor = conn.cursor()
    
    # Siapkan variabel output
    nama_var      = cursor.var(oracledb.STRING)
    usia_var      = cursor.var(oracledb.NUMBER)
    ditemukan_var = cursor.var(oracledb.NUMBER)
    
    cursor.callproc(
        "get_pengguna_by_email",
        [email, nama_var, usia_var, ditemukan_var]
    )
    
    cursor.close()
    
    if int(ditemukan_var.getvalue()) == 0:
        return None
    
    return {
        "nama": nama_var.getvalue(),
        "usia": int(usia_var.getvalue())
    }

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

Transaksi #

def transfer_saldo(conn, dari_id: int, ke_id: int, jumlah: float) -> None:
    cursor = conn.cursor()
    
    try:
        # Kunci baris untuk dibaca dan diupdate
        cursor.execute(
            "SELECT saldo FROM akun WHERE id = :id FOR UPDATE",
            {"id": dari_id}
        )
        baris = cursor.fetchone()
        
        if not baris or baris[0] < jumlah:
            raise ValueError(f"Saldo tidak mencukupi. Tersedia: {baris[0] if baris else 0}")
        
        cursor.execute(
            "UPDATE akun SET saldo = saldo - :jumlah WHERE id = :id",
            {"jumlah": jumlah, "id": dari_id}
        )
        cursor.execute(
            "UPDATE akun SET saldo = saldo + :jumlah WHERE id = :id",
            {"jumlah": jumlah, "id": 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 oracledb

def tambah_pengguna_aman(conn, nama: str, email: str) -> int | None:
    cursor = conn.cursor()
    id_var = cursor.var(oracledb.NUMBER)
    
    try:
        cursor.execute(
            """
            INSERT INTO pengguna (nama, email)
            VALUES (:nama, :email)
            RETURNING id INTO :id_baru
            """,
            {"nama": nama, "email": email, "id_baru": id_var}
        )
        conn.commit()
        return int(id_var.getvalue()[0])
    
    except oracledb.IntegrityError as e:
        conn.rollback()
        # ORA-00001: unique constraint violated
        if "ORA-00001" in str(e):
            print(f"Email '{email}' sudah terdaftar.")
        else:
            print(f"Integrity error: {e}")
        return None
    
    except oracledb.DatabaseError as e:
        conn.rollback()
        error_obj = e.args[0]
        print(f"Oracle error [{error_obj.code}]: {error_obj.message}")
        return None
    
    finally:
        cursor.close()

Ringkasan #

Placeholder :nama — Oracle menggunakan placeholder bernama (:nama) atau posisional (:1, :2); bukan ? (MSSQL/pyodbc) atau %s (MySQL).
RETURNING INTO — gunakan klausa ini untuk mendapatkan nilai kolom yang baru di-insert (termasuk ID), dikombinasikan dengan cursor.var().
cursor.rowfactory — set lambda ini agar setiap baris hasil query dikembalikan sebagai dict dengan nama kolom lowercase.
GENERATED ALWAYS AS IDENTITY — cara modern Oracle 12c+ untuk auto-increment, menggantikan pola SEQUENCE + TRIGGER.
NVARCHAR2 bukan VARCHAR2 — gunakan untuk kolom string yang menyimpan karakter non-ASCII/Unicode.
FETCH FIRST N ROWS ONLY — padanan LIMIT di Oracle 12c+; untuk Oracle 11g ke bawah gunakan ROWNUM.
cursor.callproc() / cursor.callfunc() — cara memanggil stored procedure dan function PL/SQL dari Python.
Mode thin vs thick — mode thin (default) tidak butuh Oracle Client dan cukup untuk sebagian besar kebutuhan; aktifkan mode thick hanya untuk fitur Oracle lanjutan.
ORA-00001 — kode error untuk unique constraint violation; tangani secara spesifik di except oracledb.IntegrityError.

← Sebelumnya: MSSQL Berikutnya: PostgreSQL →