SQLAlchemy #

SQLAlchemy is the most complete and widely used Python database library — not just an ORM, but a database toolkit with two distinct layers that can be used separately or together. SQLAlchemy Core provides an expressive SQL abstraction without hiding your queries, while SQLAlchemy ORM maps database tables to Python classes so you work with objects instead of raw SQL. Understanding the difference between these two layers, how to define relationships, and how to manage sessions is the foundation for using SQLAlchemy effectively in real projects.

The core concept of the SQLAlchemy ORM (Object-Relational Mapping) is mapping object-oriented programming elements in Python to relational tables in a database. This mapping relationship can be seen in the diagram below:

flowchart LR
    subgraph Python ["Python Environment (OOP)"]
        Class["Python Class (Model)<br>class User(Base)"]
        Attr["Object Attributes<br>id, name, email"]
        Obj["Object Instance<br>user = User(name='Budi')"]
    end

    subgraph ORM ["SQLAlchemy ORM"]
        Mapping["Mapping"]
    end

    subgraph DB ["Relational Database (SQL)"]
        Table["Database Table<br>users"]
        Col["Table Columns<br>id, name, email"]
        Row["Data Row (Record)<br>1 | Budi | [email protected]"]
    end

    Class --> Mapping
    Attr --> Mapping
    Obj --> Mapping

    Mapping --> Table
    Mapping --> Col
    Mapping --> Row

Installation #

pip install sqlalchemy

# Add the database driver you use:
pip install psycopg2-binary          # PostgreSQL
pip install mysql-connector-python   # MySQL
pip install pyodbc                   # MSSQL
# SQLite is already included in standard Python

Core vs ORM — SQLAlchemy’s Two Layers #

Before writing code, it’s important to understand the two ways of using SQLAlchemy and when to choose each:

SQLAlchemy Core
  ├── Works with tables, columns, and SQL expressions explicitly
  ├── You're still "thinking SQL" but with a Python API
  ├── Better performance for complex queries and bulk operations
  └── Good for: data pipelines, reporting, complex SQL queries

SQLAlchemy ORM
  ├── Maps tables to Python classes (models)
  ├── CRUD operations via Python objects, not raw SQL
  ├── Features: lazy/eager loading, relationships, identity map
  └── Good for: web apps, APIs, domains rich in business logic

SQLAlchemy 2.0 introduced a new API that’s more consistent between Core and ORM. This article uses SQLAlchemy 2.x — if you’re still on 1.x, some syntax differs, especially how queries are executed and how sessions are created.

Engine Setup and Connection #

Engine is the entry point to the database — it manages the connection pool and executes queries. An engine is usually created once when the app starts.

from sqlalchemy import create_engine, text
import os

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

# ANTI-PATTERN: hardcoding the database URL in code
engine = create_engine("postgresql://postgres:***@localhost/myapp")  # ✗

# CORRECT: read from an environment variable
engine = create_engine(
    DATABASE_URL,
    echo=False,           # True to log all SQL -- useful when debugging
    pool_size=5,          # number of connections maintained
    max_overflow=10,      # extra connections when the pool is full
    pool_pre_ping=True    # check the connection before use (prevents stale connections)
)

# Example URLs for various databases
# SQLite (local file):
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"
)

# Test the connection
with engine.connect() as conn:
    result = conn.execute(text("SELECT 1"))
    print("Connection successful:", result.fetchone())

Defining Models (ORM) #

A model is a Python class representing a database table. Since SQLAlchemy 2.0, the modern way uses DeclarativeBase and 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())
    
    # One-to-many relationship to 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")

# Create all tables that don't exist yet
Base.metadata.create_all(engine)

Session Management #

Session is the ORM’s unit of work — it manages tracked objects, collects changes, and sends them to the database as one transaction.

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: forgetting to close the session
session = SessionLocal()
pengguna = session.get(Pengguna, 1)
# ... forgot session.close() -- connection leak!

# CORRECT: always use the context manager
with get_session() as session:
    pengguna = session.get(Pengguna, 1)
    print(pengguna)

CRUD Operations with the 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()      # send to the DB but don't commit yet -- id is already available
        session.refresh(pengguna)  # reload from the 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)

# Usage
p = tambah_pengguna("Budi Santoso", "[email protected]", 28)
print(f"New user: {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:
        # Way 1: session.get() -- for PK lookups, checks the identity map first
        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 only specific fields
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]

# Complex filtering
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 the given attributes
        for key, value in kwargs.items():
            if hasattr(pengguna, key):
                setattr(pengguna, key, value)
        
        return True  # automatic commit via the context manager

# Bulk update is more efficient with 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

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

Advanced Queries #

Joining Tables #

from sqlalchemy.orm import joinedload, selectinload

# Inner join with the 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 of relationships -- avoids N+1 queries
def ambil_pengguna_dengan_orders() -> list[Pengguna]:
    with get_session() as session:
        stmt = (
            select(Pengguna)
            .options(selectinload(Pengguna.orders))  # load orders at once
            .where(Pengguna.aktif == True)
        )
        pengguna_list = list(session.scalars(stmt))
        
        for p in pengguna_list:
            # orders are already loaded, no extra queries
            print(f"{p.nama}: {len(p.orders)} orders")
        
        return pengguna_list

Aggregation and 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
        ]

Subqueries #

from sqlalchemy import subquery

def pengguna_dengan_order_mahal(batas_harga: float) -> list[Pengguna]:
    with get_session() as session:
        # Subquery: ids of users who have an order above the threshold
        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))

Running Raw SQL #

Sometimes an ORM query is too complex and easier to write directly as SQL. SQLAlchemy still supports this safely via 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()]

# Usage
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:
        # Syntax depends on the database (this is a PostgreSQL example)
        hasil = conn.execute(text(f"SELECT * FROM {nama_proc}(:param1)"), params)
        return [dict(zip(hasil.keys(), baris)) for baris in hasil.fetchall()]

Migrations with Alembic #

Alembic is the official database migration tool for SQLAlchemy — it tracks schema changes and applies them incrementally.

pip install alembic

# Initialize in the project directory
alembic init alembic

# Create a migration file automatically from model changes
alembic revision --autogenerate -m "add_pengguna_table"

# Apply the migration to the database
alembic upgrade head

# Roll back one version
alembic downgrade -1

# alembic/env.py -- metadata target configuration
from myapp.models import Base  # import the Base containing all models

target_metadata = Base.metadata

Summary #

Core vs ORM — use Core for complex queries and bulk operations needing performance; use the ORM for applications with a domain rich in business logic.
create_engine() once — the engine manages the connection pool; create one instance at app startup, not per request.
pool_pre_ping=True — enable it so the engine checks the connection before use and avoids stale-connection errors.
A session context manager — always use with get_session() as session so commit/rollback and session closing are guaranteed.
session.get() for PKs — more efficient than select().where() because it checks the identity map before hitting the database.
selectinload() / joinedload() — use eager loading for relationships to avoid N+1 queries when accessing relationship attributes.
session.add_all() — use it to insert many objects at once in a single transaction.
Bulk update/delete via Core — to update or delete many rows at once, use sa_update() and sa_delete(), far more efficient than an ORM loop.
text() for raw SQL — when a query is too complex for the ORM, use text() with named parameters to stay safe from SQL injection.
Alembic for migrations — don’t change database schemas manually; use Alembic so changes are tracked and rollbackable.

← Previous: PostgreSQL Next: Django ORM →