PostgreSQL #

PostgreSQL is the most advanced open-source relational database available today — supporting JSON, arrays, full-text search, custom data types, and extensions like PostGIS for geospatial data. Among modern Python developers, PostgreSQL is the top choice for web applications and APIs, especially thanks to its seamless integration with Django ORM, SQLAlchemy, and FastAPI. The psycopg2 library is the most popular PostgreSQL driver for Python, following the DB-API 2.0 standard with extensions that leverage PostgreSQL’s signature features.

Installation #

pip install psycopg2-binary

psycopg2-binary bundles pre-compiled binaries so you don’t need to install PostgreSQL development headers. For production environments, use psycopg2 (without -binary), compiled from source — it’s more stable and recommended by the library maintainers. If you want the newest version (async-native), consider psycopg (psycopg3).

Creating a Connection #

psycopg2 supports two connection formats: keyword arguments or a connection string (DSN). psycopg2’s parameter placeholders use %s like MySQL, but it’s not Python string formatting — it’s a placeholder processed safely by the driver.

import psycopg2
import os

# ANTI-PATTERN: hardcoding credentials in code
conn = psycopg2.connect(
    host="localhost",
    database="myapp",
    user="postgres",
    password="secret123"  # ✗ -- don't do this
)

# CORRECT: read from environment variables
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 the timezone per connection
    )

# Or using a DSN string (useful for cloud/managed databases)
def get_connection_dsn() -> psycopg2.extensions.connection:
    dsn = os.getenv("DATABASE_URL")
    # Format: postgresql://user:***@host:port/dbname
    return psycopg2.connect(dsn)

Connection Context Manager #

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

# Usage
with db_connection() as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT version()")
    version = cursor.fetchone()
    print("PostgreSQL:", version[0][:40])

Creating Tables #

PostgreSQL has richer data types than other databases — SERIAL/BIGSERIAL for auto-increment, TEXT for unlimited strings, JSONB for JSON, and ARRAY for native arrays.

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 '{}'
        )
    """)
    
    # Indexes for common query performance
    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("Tables created successfully.")

with db_connection() as conn:
    create_tables(conn)

PostgreSQL data types not found in MySQL/MSSQL:
TEXT — unlimited-length strings (no need for VARCHAR(n) unless you want a limit)
BIGSERIAL / SERIAL — auto-increment (equivalent to BIGINT AUTO_INCREMENT)
BOOLEAN — native boolean type (not BIT or NUMBER(1))
JSONB — JSON stored in binary format, indexable and queryable
TEXT[] — native arrays of any type in PostgreSQL
TIMESTAMPTZ — timestamp with timezone (recommended for multi-timezone apps)
UUID — native UUID type

CRUD Operations #

Inserting Data #

import psycopg2.extras

def tambah_pengguna(conn, nama: str, email: str, usia: int, tag: list[str] = None) -> int:
    cursor = conn.cursor()
    
    # RETURNING id -- the PostgreSQL idiom for getting the ID after 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 -- far more efficient than executemany for batch inserts
    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"New user 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} users added.")

Reading Data #

def ambil_semua_pengguna(conn) -> list[dict]:
    # RealDictCursor -- every row is directly a 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 with 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]

# JSONB query -- a signature PostgreSQL feature
def ambil_pengguna_by_metadata(conn, kunci: str, nilai: str) -> list[dict]:
    cursor = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
    
    # The -> operator to access JSONB fields
    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']}")

Updating and Deleting #

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 with ON CONFLICT #

INSERT ... ON CONFLICT is a PostgreSQL feature for inserting or updating at once — very useful for data synchronization and idempotent operations.

def upsert_pengguna(conn, email: str, nama: str, usia: int) -> int:
    cursor = conn.cursor()
    
    # If the email exists, update name and age
    # If it doesn't exist, insert a new row
    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 -- ignore if it already exists
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"Same ID: {id1 == id2}")  # True -- updated, not a new insert

Connection Pooling #

For web applications, use psycopg2.pool so connections aren’t recreated per request:

import psycopg2.pool
import os

# Create the pool when the app starts -- 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)  # return to the pool

# Usage is identical to a regular connection
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

Transactions #

def transfer_saldo(conn, dari_id: int, ke_id: int, jumlah: float) -> None:
    cursor = conn.cursor()
    
    try:
        # Lock both rows at once to prevent deadlocks
        cursor.execute("""
            SELECT id, saldo FROM akun
            WHERE id = ANY(%s)
            ORDER BY id          -- consistent order prevents deadlocks
            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"Insufficient balance. Available: {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 of {jumlah:,.0f} successful.")
        
    except Exception as e:
        conn.rollback()
        print(f"Transaction rolled back: {e}")
        raise
    finally:
        cursor.close()

Error Handling #

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}' is already registered.")
        return None
    
    except errors.NotNullViolation as e:
        conn.rollback()
        print(f"A required field can't be empty: {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()

Summary #

%s placeholders — psycopg2 uses %s as placeholders, but this is not Python % string formatting; always pass a separate tuple/list of values as the second argument to execute().
RETURNING — use this clause after INSERT/UPDATE/DELETE to get affected column values without an extra query.
RealDictCursor — use cursor_factory=psycopg2.extras.RealDictCursor so query result rows are directly dicts with column names as keys.
execute_values() — use it for bulk inserts, far more efficient than an executemany() loop.
ON CONFLICT — leverage INSERT ... ON CONFLICT DO UPDATE (UPSERT) for idempotent insert-or-update operations.
JSONB — store semi-structured data directly in PostgreSQL with the JSONB type, which can be indexed and queried using the -> and ->> operators.
TEXT[] — PostgreSQL supports native arrays; no separate relation table needed for simple list data.
ThreadedConnectionPool — use a connection pool for web apps so you don’t create a new connection per request.
errors.UniqueViolation — handle specific PostgreSQL errors via the psycopg2.errors module rather than matching error message strings.

← Previous: Oracle Next: SQLAlchemy →