SQLAlchemy ORM untuk Python: Panduan Lengkap

📋 Daftar Isi

Pengenalan SQLAlchemy
Instalasi dan Konfigurasi
Mendefinisikan Models
CRUD Operations
Querying dan Filtering
Relationships
Teknik Lanjutan
Migrations dengan Alembic
Best Practices
Quiz Pemahaman

1. Pengenalan SQLAlchemy

SQLAlchemy adalah toolkit dan ORM (Object-Relational Mapping) paling populer untuk Python. SQLAlchemy memungkinkan Anda berinteraksi dengan database menggunakan Python objects, bukan SQL mentah.

Apa itu ORM?

ORM (Object-Relational Mapping) adalah teknik yang memetakan tabel database menjadi class Python. Setiap baris dalam tabel menjadi sebuah object, dan operasi database dilakukan melalui method Python.

SQLAlchemy vs SQL Mentah

Aspek	SQL Mentah	SQLAlchemy ORM
Keamanan	Rentan SQL injection	Aman (parameterized)
Portabilitas	Tergantung database	Multi-database
Readability	Raw SQL string	Pythonic syntax
Maintenance	Sulit di-refactor	Mudah dikelola
Kecepatan Dev	Lambat	Cepat
Kecepatan Query	Bisa dioptimasi manual	Perlu perhatian

Diagram: Cara Kerja ORM

┌─────────────────────────────────────────────────────────┐
│                   CARA KERJA ORM                        │
│                                                         │
│  Python Code         ORM Engine         Database        │
│  ──────────         ───────────         ────────        │
│                                                         │
│  user = User(       ──▶ Translate ──▶   INSERT INTO     │
│    name="Budi"                          users (name)    │
│  )                                      VALUES ('Budi') │
│                                                         │
│  session.add(user)  ──▶ Execute    ──▶  COMMIT          │
│  session.commit()                                           │
│                                                         │
│  session.query(User) ──▶ Translate ──▶ SELECT * FROM    │
│  .filter_by(                            users WHERE     │
│    name="Budi"                          name='Budi'     │
│  )                                      LIMIT 1         │
└─────────────────────────────────────────────────────────┘

2. Instalasi dan Konfigurasi

Terminal

# Instal SQLAlchemy
pip install sqlalchemy

# Verifikasi
python -c "import sqlalchemy; print(sqlalchemy.__version__)"
# Output: 2.0.x

# Untuk migrations
pip install alembic

# Driver database (pilih sesuai database yang dipakai)
pip install psycopg2-binary   # PostgreSQL
pip install pymysql            # MySQL/MariaDB
# SQLite sudah built-in di Python, tidak perlu driver

Koneksi ke Database

Python — Database Setup

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase

# ========================================
# 1. Membuat Engine (Koneksi ke Database)
# ========================================
# SQLite (untuk development/testing)
engine = create_engine('sqlite:///app.db', echo=True)

# SQLite in-memory (untuk testing)
engine = create_engine('sqlite:///:memory:', echo=True)

# PostgreSQL
engine = create_engine('postgresql://user:password@localhost:5432/mydb')

# MySQL
engine = create_engine('mysql+pymysql://user:password@localhost:3306/mydb')

# Parameter umum
engine = create_engine(
    'sqlite:///app.db',
    echo=True,           # Print SQL yang dijalankan (untuk debug)
    pool_size=5,         # Jumlah koneksi di pool
    max_overflow=10,     # Koneksi ekstra saat pool penuh
    pool_pre_ping=True   # Cek koneksi sebelum dipakai
)

# ========================================
# 2. Base Class untuk Models
# ========================================
# SQLAlchemy 2.0 style
class Base(DeclarativeBase):
    pass

# ========================================
# 3. Membuat Session
# ========================================
SessionLocal = sessionmaker(bind=engine)

# Membuat tabel
Base.metadata.create_all(engine)

# Menggunakan session
session = SessionLocal()
# ... operasi database ...
session.close()

# Atau gunakan context manager
with SessionLocal() as session:
    # ... operasi database ...
    pass  # session otomatis ditutup

3. Mendefinisikan Models

Setiap tabel database direpresentasikan sebagai sebuah class yang mewarisi dari Base. Setiap kolom didefinisikan sebagai atribut class menggunakan Column.

Model Dasar

Python — Models

from sqlalchemy import (
    Column, Integer, String, Float, Boolean,
    DateTime, Text, ForeignKey, Enum as SQLEnum
)
from sqlalchemy.orm import DeclarativeBase, relationship
from datetime import datetime
import enum

class Base(DeclarativeBase):
    pass

# ========================================
# Model User
# ========================================
class User(Base):
    __tablename__ = 'users'

    # Primary Key
    id = Column(Integer, primary_key=True, autoincrement=True)

    # Kolom biasa
    username = Column(String(50), unique=True, nullable=False, index=True)
    email = Column(String(100), unique=True, nullable=False)
    full_name = Column(String(100), nullable=False)
    age = Column(Integer, default=0)
    bio = Column(Text, nullable=True)
    is_active = Column(Boolean, default=True)
    balance = Column(Float, default=0.0)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

    # Relationship (akan dibahas di section 6)
    posts = relationship('Post', back_populates='author', cascade='all, delete-orphan')

    def __repr__(self):
        return f"<User(id={self.id}, username='{self.username}')>"

    def to_dict(self):
        return {
            'id': self.id,
            'username': self.username,
            'email': self.email,
            'full_name': self.full_name,
            'is_active': self.is_active
        }

# ========================================
# Model dengan Enum
# ========================================
class PostStatus(enum.Enum):
    DRAFT = "draft"
    PUBLISHED = "published"
    ARCHIVED = "archived"

class Post(Base):
    __tablename__ = 'posts'

    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(200), nullable=False)
    content = Column(Text, nullable=False)
    status = Column(SQLEnum(PostStatus), default=PostStatus.DRAFT)
    views = Column(Integer, default=0)
    created_at = Column(DateTime, default=datetime.utcnow)

    # Foreign Key
    author_id = Column(Integer, ForeignKey('users.id'), nullable=False)

    # Relationship
    author = relationship('User', back_populates='posts')
    tags = relationship('Tag', secondary='post_tags', back_populates='posts')

    def __repr__(self):
        return f"<Post(id={self.id}, title='{self.title}')>"

# ========================================
# Tabel Association (Many-to-Many)
# ========================================
from sqlalchemy import Table

post_tags = Table(
    'post_tags', Base.metadata,
    Column('post_id', Integer, ForeignKey('posts.id'), primary_key=True),
    Column('tag_id', Integer, ForeignKey('tags.id'), primary_key=True)
)

class Tag(Base):
    __tablename__ = 'tags'

    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(50), unique=True, nullable=False)

    posts = relationship('Post', secondary=post_tags, back_populates='tags')

    def __repr__(self):
        return f"<Tag(id={self.id}, name='{self.name}')>"

Tipe Data Kolom

Tipe SQLAlchemy	Tipe Python	Contoh Database
`Integer`	int	INT, INTEGER
`String(n)`	str	VARCHAR(n)
`Text`	str	TEXT
`Float`	float	FLOAT, REAL
`Boolean`	bool	BOOLEAN
`DateTime`	datetime	DATETIME, TIMESTAMP
`Date`	date	DATE
`JSON`	dict/list	JSON, JSONB
`LargeBinary`	bytes	BLOB, BYTEA
`Enum`	enum	ENUM

4. CRUD Operations

CRUD (Create, Read, Update, Delete) adalah operasi dasar yang dilakukan pada database. SQLAlchemy membuat operasi ini sangat Pythonic.

Python — CRUD Operations

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine('sqlite:///app.db')
Session = sessionmaker(bind=engine)

# ========================================
# CREATE — Menambah Data
# ========================================
with Session() as session:
    # Tambah satu data
    user = User(
        username='budi',
        email='budi@mail.com',
        full_name='Budi Santoso',
        age=25
    )
    session.add(user)
    session.commit()
    print(f"User dibuat dengan ID: {user.id}")

    # Tambah beberapa data sekaligus
    users = [
        User(username='ani', email='ani@mail.com', full_name='Ani Wijaya', age=23),
        User(username='citra', email='citra@mail.com', full_name='Citra Dewi', age=28),
        User(username='dodi', email='dodi@mail.com', full_name='Dodi Pratama', age=30),
    ]
    session.add_all(users)
    session.commit()

    # Tambah post (dengan foreign key)
    post = Post(
        title='Belajar SQLAlchemy',
        content='SQLAlchemy adalah ORM terbaik untuk Python...',
        author_id=user.id,
        status=PostStatus.PUBLISHED
    )
    session.add(post)
    session.commit()

# ========================================
# READ — Membaca Data
# ========================================
with Session() as session:
    # Ambil semua users
    all_users = session.query(User).all()
    for u in all_users:
        print(f"{u.username} - {u.email}")

    # Ambil user pertama yang cocok
    user = session.query(User).filter_by(username='budi').first()
    print(user)

    # Ambil berdasarkan primary key
    user = session.get(User, 1)  # SQLAlchemy 2.0
    # Atau: session.query(User).get(1)  # SQLAlchemy 1.x

    # Filter dengan kondisi
    active_users = session.query(User).filter(User.is_active == True).all()
    young_users = session.query(User).filter(User.age < 25).all()

    # Hitung jumlah
    count = session.query(User).count()
    print(f"Total users: {count}")

    # Exists
    exists = session.query(User).filter_by(username='budi').first() is not None

# ========================================
# UPDATE — Memperbarui Data
# ========================================
with Session() as session:
    # Cara 1: Update langsung pada object
    user = session.query(User).filter_by(username='budi').first()
    if user:
        user.full_name = 'Budi Santoso, S.Kom'
        user.age = 26
        session.commit()

    # Cara 2: Bulk update (lebih efisien untuk banyak data)
    session.query(User).filter(User.age < 25).update(
        {User.is_active: False},
        synchronize_session='fetch'
    )
    session.commit()

    # Cara 3: Update dengan synchronize options
    session.query(User).filter(User.username == 'ani').update(
        {'age': User.age + 1},  # Bisa pakai expression
        synchronize_session='evaluate'
    )
    session.commit()

# ========================================
# DELETE — Menghapus Data
# ========================================
with Session() as session:
    # Hapus satu data
    user = session.query(User).filter_by(username='dodi').first()
    if user:
        session.delete(user)
        session.commit()

    # Bulk delete
    session.query(User).filter(User.is_active == False).delete()
    session.commit()

💡 Tips: Context Manager

Selalu gunakan with Session() as session: untuk memastikan session ditutup dengan benar. Jika terjadi error, session akan otomatis melakukan rollback. Jika berhasil, Anda perlu memanggil session.commit() secara eksplisit.

5. Querying dan Filtering

SQLAlchemy menyediakan API yang sangat ekspresif untuk membangun query database dengan cara yang Pythonic.

Python — Advanced Queries

from sqlalchemy import and_, or_, not_, func, desc, asc
from sqlalchemy.orm import Session

with Session(engine) as session:
    # ========================================
    # Filtering
    # ========================================

    # filter() — lebih fleksibel
    users = session.query(User).filter(User.age > 25).all()
    users = session.query(User).filter(User.username.like('%udi%')).all()
    users = session.query(User).filter(User.username.in_(['budi', 'ani'])).all()
    users = session.query(User).filter(User.bio.is_(None)).all()  # IS NULL
    users = session.query(User).filter(User.bio.isnot(None)).all()  # IS NOT NULL

    # filter_by() — untuk equality sederhana
    user = session.query(User).filter_by(username='budi', is_active=True).first()

    # Kombinasi filter
    users = session.query(User).filter(
        and_(
            User.age >= 20,
            User.age <= 30,
            User.is_active == True
        )
    ).all()

    # OR
    users = session.query(User).filter(
        or_(User.username == 'budi', User.username == 'ani')
    ).all()

    # NOT
    users = session.query(User).filter(not_(User.is_active)).all()

    # ========================================
    # Sorting (ORDER BY)
    # ========================================
    users = session.query(User).order_by(User.username).all()  # ASC
    users = session.query(User).order_by(desc(User.created_at)).all()  # DESC
    users = session.query(User).order_by(User.age.desc(), User.username.asc()).all()

    # ========================================
    # Limit dan Offset (LIMIT/OFFSET)
    # ========================================
    # Pagination sederhana
    page = 2
    per_page = 10
    users = session.query(User)\
        .order_by(User.id)\
        .offset((page - 1) * per_page)\
        .limit(per_page)\
        .all()

    # ========================================
    # Select Kolom Tertentu
    # ========================================
    # Hanya ambil kolom tertentu
    result = session.query(User.username, User.email).all()
    for username, email in result:
        print(f"{username}: {email}")

    # Distinct
    ages = session.query(User.age).distinct().all()

    # ========================================
    # Aggregation (GROUP BY)
    # ========================================
    # Hitung user per age
    stats = session.query(
        User.age,
        func.count(User.id).label('count'),
        func.avg(User.age).label('avg_age')
    ).group_by(User.age).all()

    for age, count, avg in stats:
        print(f"Umur {age}: {count} users")

    # Having
    stats = session.query(
        User.age,
        func.count(User.id).label('count')
    ).group_by(User.age).having(func.count(User.id) > 1).all()

    # Max, Min, Sum
    max_age = session.query(func.max(User.age)).scalar()
    min_age = session.query(func.min(User.age)).scalar()
    total_balance = session.query(func.sum(User.balance)).scalar()

    # ========================================
    # Join
    # ========================================
    # Implicit join (via relationship)
    results = session.query(User, Post).filter(User.id == Post.author_id).all()

    # Explicit join
    results = session.query(User, Post).join(Post, User.id == Post.author_id).all()

    # Join dengan filter
    results = session.query(User.username, Post.title)\
        .join(Post)\
        .filter(Post.status == PostStatus.PUBLISHED)\
        .all()

    # Left outer join
    results = session.query(User, Post)\
        .outerjoin(Post, User.id == Post.author_id)\
        .all()

    # ========================================
    # Subquery
    # ========================================
    # Users yang punya post
    from sqlalchemy import select

    subq = select(Post.author_id).distinct()
    users_with_posts = session.query(User).filter(User.id.in_(subq)).all()

    # ========================================
    # Raw SQL (ketika perlu)
    # ========================================
    result = session.execute(
        text("SELECT * FROM users WHERE age > :min_age"),
        {"min_age": 25}
    )
    rows = result.fetchall()

6. Relationships

Relationship mendefinisikan hubungan antar tabel — salah satu kekuatan utama ORM.

Jenis Relationship

Diagram: Jenis Relationship

┌─────────────────────────────────────────────────────────┐
│                JENIS RELATIONSHIP                       │
│                                                         │
│  One-to-Many:                                           │
│  ┌──────────┐         ┌──────────┐                     │
│  │  User    │ 1 ─── * │  Post    │                     │
│  │  (1)     │         │  (many)  │                     │
│  └──────────┘         └──────────┘                     │
│                                                         │
│  One-to-One:                                            │
│  ┌──────────┐         ┌──────────┐                     │
│  │  User    │ 1 ─── 1 │  Profile │                     │
│  └──────────┘         └──────────┘                     │
│                                                         │
│  Many-to-Many:                                          │
│  ┌──────────┐         ┌──────────┐                     │
│  │  Post    │ * ─── * │  Tag     │                     │
│  └──────────┘         └──────────┘                     │
│       └──── post_tags (association table) ────┘        │
└─────────────────────────────────────────────────────────┘

Python — Relationships

from sqlalchemy.orm import relationship

# ========================================
# One-to-Many: User → Posts
# ========================================
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True)

    # back_populates membuat kedua sisi sinkron
    posts = relationship('Post', back_populates='author', cascade='all, delete-orphan')

class Post(Base):
    __tablename__ = 'posts'
    id = Column(Integer, primary_key=True)
    title = Column(String(200))
    author_id = Column(Integer, ForeignKey('users.id'))

    author = relationship('User', back_populates='posts')

# Menggunakan relationship
with Session(engine) as session:
    # Akses posts dari user
    user = session.query(User).filter_by(username='budi').first()
    for post in user.posts:
        print(post.title)

    # Akses author dari post
    post = session.query(Post).first()
    print(post.author.username)

    # Membuat post baru melalui relationship
    user.posts.append(Post(title='Post Baru', content='Isi...'))
    session.commit()

# ========================================
# One-to-One: User → Profile
# ========================================
class Profile(Base):
    __tablename__ = 'profiles'
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('users.id'), unique=True)
    avatar_url = Column(String(500))
    location = Column(String(100))

    user = relationship('User', back_populates='profile')

# Di User:
# profile = relationship('Profile', back_populates='user', uselist=False)

# ========================================
# Many-to-Many: Posts ↔ Tags
# ========================================
# (Sudah didefinisikan di section Models)

with Session(engine) as session:
    post = session.query(Post).first()
    tag = Tag(name='Python')

    # Tambah tag ke post
    post.tags.append(tag)
    session.commit()

    # Akses tags dari post
    for t in post.tags:
        print(t.name)

    # Akses posts dari tag
    tag = session.query(Tag).filter_by(name='Python').first()
    for p in tag.posts:
        print(p.title)

# ========================================
# Cascade Options
# ========================================
# cascade='all, delete-orphan' artinya:
# - Saat user dihapus, semua posts ikut terhapus
# - Saat post di-remove dari user.posts, post ikut terhapus

# cascade='save-update' (default) — cascade hanya saat save
# cascade='delete' — cascade saat delete parent
# cascade='all' — semua cascade kecuali delete-orphan
# cascade='delete-orphan' — hapus orphaned children

# ========================================
# Lazy Loading vs Eager Loading
# ========================================
from sqlalchemy.orm import joinedload, selectinload

# Lazy loading (default) — query terpisah saat diakses
users = session.query(User).all()
for user in users:
    print(user.posts)  # Query baru di sini! (N+1 problem)

# Eager loading — join di awal
users = session.query(User).options(joinedload(User.posts)).all()
for user in users:
    print(user.posts)  # Sudah ter-load, tidak ada query baru

# selectinload — untuk collection (lebih efisien untuk many)
users = session.query(User).options(selectinload(User.posts)).all()

7. Teknik Lanjutan

Python — Advanced Techniques

from sqlalchemy import event, text
from sqlalchemy.orm import Session

# ========================================
# 1. Hybrid Properties (computed columns)
# ========================================
from sqlalchemy.ext.hybrid import hybrid_property

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    first_name = Column(String(50))
    last_name = Column(String(50))
    age = Column(Integer)

    @hybrid_property
    def full_name(self):
        return f"{self.first_name} {self.last_name}"

    @hybrid_property
    def is_adult(self):
        return self.age >= 18

    # Bisa dipakai di query juga!
    @is_adult.expression
    def is_adult(cls):
        return cls.age >= 18

# Pakai di query
adults = session.query(User).filter(User.is_adult).all()

# ========================================
# 2. Event Listeners
# ========================================
@event.listens_for(User, 'before_insert')
def set_default(mapper, connection, target):
    """Dijalankan sebelum insert"""
    if not target.created_at:
        target.created_at = datetime.utcnow()

@event.listens_for(User, 'after_update')
def log_update(mapper, connection, target):
    """Dijalankan setelah update"""
    print(f"User {target.id} diperbarui")

# ========================================
# 3. Raw SQL Execution
# ========================================
with Session(engine) as session:
    # Jalankan raw SQL
    result = session.execute(text("SELECT * FROM users WHERE age > 25"))
    for row in result:
        print(row)

    # Dengan parameter
    result = session.execute(
        text("SELECT * FROM users WHERE username = :name"),
        {"name": "budi"}
    )

    # DDL operations
    session.execute(text("CREATE INDEX idx_users_email ON users(email)"))
    session.commit()

# ========================================
# 4. Transactions
# ========================================
with Session(engine) as session:
    # Semua operasi dalam satu transaction
    try:
        user = User(username='new_user', email='new@mail.com')
        session.add(user)

        post = Post(title='First Post', author_id=user.id)
        session.add(post)

        session.commit()  # Kedua operasi berhasil
    except Exception as e:
        session.rollback()  # Kedua operasi dibatalkan
        print(f"Error: {e}")

# ========================================
# 5. Bulk Operations (cepat untuk banyak data)
# ========================================
with Session(engine) as session:
    # Bulk insert
    users_data = [
        {'username': f'user_{i}', 'email': f'user{i}@mail.com', 'full_name': f'User {i}'}
        for i in range(1000)
    ]
    session.bulk_insert_mappings(User, users_data)
    session.commit()

    # Bulk update
    session.bulk_update_mappings(User, [
        {'id': 1, 'age': 26},
        {'id': 2, 'age': 24},
    ])
    session.commit()

8. Migrations dengan Alembic

Alembic adalah tool migrasi database untuk SQLAlchemy. Alembic memungkinkan Anda mengubah skema database secara terkontrol dan dapat di-reverse.

Terminal — Alembic Setup

# Instal Alembic
pip install alembic

# Inisialisasi Alembic di project
alembic init alembic

# Struktur folder yang dihasilan:
# project/
# ├── alembic/
# │   ├── versions/        # File migrasi
# │   └── env.py           # Konfigurasi environment
# ├── alembic.ini          # Konfigurasi Alembic
# └── models.py            # SQLAlchemy models

Python — Alembic Configuration

# alembic/env.py — konfigurasi untuk auto-generate
from alembic import context
from sqlalchemy import engine_from_config, pool
from models import Base  # Import Base dari models Anda

# Set target metadata untuk auto-generate
target_metadata = Base.metadata

# Konfigurasi di alembic.ini:
# sqlalchemy.url = sqlite:///app.db
# Atau: postgresql://user:pass@localhost/dbname

Terminal — Alembic Commands

# ========================================
# Membuat migrasi baru (auto-generate)
# Alembic mendeteksi perubahan pada models
# ========================================
alembic revision --autogenerate -m "tambah tabel users dan posts"

# Hasil file: alembic/versions/abc123_tambah_tabel_users.py

# ========================================
# Menjalankan migrasi
# ========================================
# Terapkan semua migrasi yang belum dijalankan
alembic upgrade head

# Terapkan satu migrasi ke depan
alembic upgrade +1

# Rollback satu migrasi
alembic downgrade -1

# Rollback ke awal (hapus semua)
alembic downgrade base

# ========================================
# Informasi migrasi
# ========================================
# Lihat versi saat ini
alembic current

# Lihat history migrasi
alembic history

# Lihat history dengan detail
alembic history --verbose

Python — Generated Migration

# Contoh file migrasi yang di-generate
"""tambah tabel users dan posts

Revision ID: abc123def456
Revises:
Create Date: 2026-06-26 10:00:00.000000
"""
from alembic import op
import sqlalchemy as sa

# revision identifiers
revision = 'abc123def456'
down_revision = None
branch_labels = None
depends_on = None

def upgrade() -> None:
    # Operasi yang dilakukan saat upgrade
    op.create_table(
        'users',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('username', sa.String(50), nullable=False),
        sa.Column('email', sa.String(100), nullable=False),
        sa.Column('full_name', sa.String(100), nullable=False),
        sa.Column('age', sa.Integer(), server_default='0'),
        sa.Column('is_active', sa.Boolean(), server_default='1'),
        sa.Column('created_at', sa.DateTime(), nullable=True),
        sa.PrimaryKeyConstraint('id'),
        sa.UniqueConstraint('username'),
        sa.UniqueConstraint('email')
    )
    op.create_index('ix_users_username', 'users', ['username'])

    op.create_table(
        'posts',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('title', sa.String(200), nullable=False),
        sa.Column('content', sa.Text(), nullable=False),
        sa.Column('author_id', sa.Integer(), nullable=False),
        sa.Column('created_at', sa.DateTime(), nullable=True),
        sa.ForeignKeyConstraint(['author_id'], ['users.id']),
        sa.PrimaryKeyConstraint('id')
    )

def downgrade() -> None:
    # Operasi yang dilakukan saat rollback
    op.drop_table('posts')
    op.drop_index('ix_users_username', table_name='users')
    op.drop_table('users')

9. Best Practices

💡 Best Practices SQLAlchemy

1. Selalu gunakan Session context manager — with Session() as s: memastikan session ditutup. 2. Hindari N+1 queries — gunakan joinedload() atau selectinload(). 3. Gunakan connection pooling — konfigurasi pool_size dan pool_pre_ping. 4. Jangan lupa rollback — tangkap exception dan rollback. 5. Gunakan Alembic untuk migrations — jangan create_all() di production.

Python — Complete Project Structure

# Contoh struktur project yang baik

# database.py — Konfigurasi database
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase

DATABASE_URL = "sqlite:///app.db"

engine = create_engine(DATABASE_URL, pool_pre_ping=True)
SessionLocal = sessionmaker(bind=engine)

class Base(DeclarativeBase):
    pass

def get_db():
    """Dependency untuk mendapatkan database session"""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# models.py — Definisi models
from database import Base
from sqlalchemy import Column, Integer, String

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)

# main.py — Penggunaan
from database import SessionLocal, engine, Base
from models import User

# Buat tabel (hanya untuk development)
Base.metadata.create_all(engine)

def create_user(name: str):
    with SessionLocal() as session:
        user = User(name=name)
        session.add(user)
        session.commit()
        session.refresh(user)  # Refresh untuk dapat ID
        return user

def get_users():
    with SessionLocal() as session:
        return session.query(User).all()

10. Quiz: Uji Pemahamanmu!

Setelah membaca tutorial di atas, jawablah 5 pertanyaan berikut untuk menguji pemahamanmu tentang SQLAlchemy ORM:

Pertanyaan 1: Apa fungsi utama dari ORM (Object-Relational Mapping)?

a) Mengenkripsi database

b) Memetakan tabel database menjadi class Python sehingga bisa diakses sebagai object

c) Membuat UI untuk database

d) Menggantikan kebutuhan database sepenuhnya

Pertanyaan 2: Apa perbedaan antara `session.add()` dan `session.add_all()`?

a) Tidak ada perbedaan

b) add() untuk satu object, add_all() untuk list of objects

c) add() langsung commit, add_all() perlu commit manual

d) add_all() lebih cepat dari add()

Pertanyaan 3: Apa fungsi dari `back_populates` dalam relationship?

a) Menghapus data terkait saat parent dihapus

b) Membuat kedua sisi relationship tetap sinkron

c) Membuat index pada foreign key

d) Mempercepat query

Pertanyaan 4: Apa masalah N+1 queries dalam ORM?

a) Database crash karena terlalu banyak koneksi

b) Setiap akses relationship memicu query baru, sehingga 1 query parent + N query children

c) ORM tidak bisa membuat lebih dari N queries

d) Database mengembalikan N+1 baris data

Pertanyaan 5: Tool apa yang digunakan untuk database migrations di SQLAlchemy?

a) Flask-Migrate

b) Django Migrations

c) Alembic

d) SQL-Migrate