Oracle #

Oracle Database is a top-tier enterprise relational database dominant in banking, telecommunications, and government. Python interacts with Oracle through the python-oracledb library — the official successor to cx_Oracle, which is now deprecated. One of Oracle’s unique traits compared to other databases is its use of named placeholders (:name instead of ? or %s), the SEQUENCE concept for auto-increment, the RETURNING INTO clause, and the thin vs thick connection modes you need to understand before starting. Mastering these differences will save you from confusion when switching from MySQL or PostgreSQL to Oracle.

Installation #

pip install python-oracledb

python-oracledb comes in two modes:

Thin Mode (default)
  ├── Pure Python, no Oracle Client needed
  ├── Just pip install, ready to use
  └── Supports most features for common applications

Thick Mode
  ├── Requires Oracle Instant Client installed
  ├── Needs explicit initialization: oracledb.init_oracle_client()
  └── Supports advanced features: Advanced Queuing, DRCP, etc.

For most new applications, thin mode is sufficient and much easier to set up. Use thick mode only if you need advanced Oracle features unavailable in thin mode, or when connecting to older Oracle Database versions (before 12.1).

Creating a Connection #

Oracle uses a DSN (Data Source Name) or connection string different from other databases. There are two common formats: the Easy Connect String or a TNS alias.

import oracledb
import os

# ANTI-PATTERN: hardcoding credentials in code
conn = oracledb.connect(
    user="admin",
    password="secret123",  # ✗ -- don't do this
    dsn="localhost/ORCLPDB1"
)

# CORRECT: read from environment variables
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("Connection successful, Oracle version:", conn.version)
conn.close()

# Connection using a TNS alias (from tnsnames.ora)
conn = oracledb.connect(
    user=os.getenv("ORACLE_USER"),
    password=os.getenv("ORACLE_PASSWORD"),
    dsn="MYDB_PROD"  # alias defined in tnsnames.ora
)

# Thick mode (if you need advanced features)
oracledb.init_oracle_client(lib_dir="/opt/oracle/instantclient_21_9")
conn = oracledb.connect(user="admin", password="...", dsn="localhost/ORCLPDB1")

Connection Context Manager #

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()

# Usage
with db_connection() as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT SYSDATE FROM DUAL")
    date_val = cursor.fetchone()
    print("Oracle server date:", date_val[0])

Creating Tables and Sequences #

Oracle doesn’t have AUTO_INCREMENT — instead it uses a SEQUENCE and TRIGGER, or since Oracle 12c a GENERATED ALWAYS AS IDENTITY column.

def create_tables(conn):
    cursor = conn.cursor()
    
    # Drop existing tables (for 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  # ignore if it doesn't exist
    
    # Create a sequence for auto-increment (Oracle 11g and earlier)
    cursor.execute("""
        CREATE SEQUENCE seq_pengguna
            START WITH 1
            INCREMENT BY 1
            NOCACHE
            NOCYCLE
    """)
    
    # Create the table -- use IDENTITY for 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("Table and sequence created successfully.")

with db_connection() as conn:
    create_tables(conn)

Important Oracle differences from MySQL/MSSQL:
NUMBER replaces INT, BIGINT, DECIMAL
NVARCHAR2 for Unicode strings (the modern Oracle recommendation)
TIMESTAMP replaces DATETIME
SYSDATE or CURRENT_TIMESTAMP for server time
DUAL — Oracle’s dummy table for queries without a real table: SELECT 1 FROM DUAL
No native BOOLEAN — use NUMBER(1) (0/1) or CHAR(1) (Y/N)

CRUD Operations #

Named Placeholders — Oracle’s Unique Feature #

# ANTI-PATTERN: using ? (MSSQL) or %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]")) # ✗

# CORRECT: Oracle uses named placeholders with a colon prefix
cursor.execute(
    "INSERT INTO pengguna (nama, email) VALUES (:nama, :email)",
    {"nama": "Budi", "email": "[email protected]"}   # ✓ -- dict with keys matching the placeholders
)

# Or with positional tuples (order must match)
cursor.execute(
    "INSERT INTO pengguna (nama, email) VALUES (:1, :2)",
    ("Budi", "[email protected]")   # ✓ -- :1, :2, :3, ...
)

Inserting Data #

def tambah_pengguna(conn, nama: str, email: str, usia: int) -> int:
    cursor = conn.cursor()
    
    # RETURNING INTO to get the newly created ID
    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"New user 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} users added.")

Reading Data #

def ambil_semua_pengguna(conn) -> list[dict]:
    cursor = conn.cursor()
    
    # rowfactory -- automatically converts rows to dicts
    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 with 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']}")

Updating and Deleting #

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 Procedures #

Oracle supports stored procedures and functions via PL/SQL. Calling them from Python uses cursor.callproc() or cursor.callfunc().

-- Create the stored procedure in Oracle first
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()
    
    # Prepare the output variables
    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"Found: {pengguna['nama']}, age {pengguna['usia']}")

Transactions #

def transfer_saldo(conn, dari_id: int, ke_id: int, jumlah: float) -> None:
    cursor = conn.cursor()
    
    try:
        # Lock the row for read and update
        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"Insufficient balance. Available: {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 of {jumlah:,.0f} successful.")
        
    except Exception as e:
        conn.rollback()
        print(f"Transaction rolled back: {e}")
        raise
    finally:
        cursor.close()

Error Handling #

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}' is already registered.")
        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()

Summary #

:name placeholders — Oracle uses named placeholders (:name) or positional ones (:1, :2); not ? (MSSQL/pyodbc) or %s (MySQL).
RETURNING INTO — use this clause to get the value of a newly inserted column (including the ID), combined with cursor.var().
cursor.rowfactory — set this lambda so every query result row is returned as a dict with lowercase column names.
GENERATED ALWAYS AS IDENTITY — the modern Oracle 12c+ way to do auto-increment, replacing the SEQUENCE + TRIGGER pattern.
NVARCHAR2 not VARCHAR2 — use it for string columns storing non-ASCII/Unicode characters.
FETCH FIRST N ROWS ONLY — the Oracle 12c+ equivalent of LIMIT; for Oracle 11g and below use ROWNUM.
cursor.callproc() / cursor.callfunc() — the way to call PL/SQL stored procedures and functions from Python.
Thin vs thick mode — thin mode (default) needs no Oracle Client and suffices for most needs; enable thick mode only for advanced Oracle features.
ORA-00001 — the error code for a unique constraint violation; handle it specifically in except oracledb.IntegrityError.

← Previous: MSSQL Next: PostgreSQL →