Setelah memahami fondasi operasi CRUD pada Bagian 1 dan relasi multi-tabel serta keamanan query pada Bagian 2, kini tiba saatnya merangkai seluruh potongan puzzle tersebut menjadi sebuah arsitektur perangkat lunak yang utuh, profesional, dan siap pakai.

Dalam proyek skala menengah hingga besar, menuliskan query SQL secara tersebar langsung di sembarang baris kode akan membuat aplikasi sulit dirawat (*spaghetti code*). Standar industri yang digunakan para pengembang profesional adalah memadukannya dengan Pemrograman Berorientasi Objek (OOP) menggunakan Pola Repositori (*Repository Pattern*).

Artikel ini adalah Bagian 3 (Penutup Serial) dari trilogi panduan Database di Python:

• Bagian 1: Fondasi SQLite dan Operasi CRUD Pertama (Koneksi, pembuatan tabel, penyimpanan permanen, dan operasi Create, Read, Update, Delete).
• Bagian 2: Relasi Multi-Tabel (JOIN), Agregasi & Keamanan Query (Foreign key, query multi-tabel, sqlite3.Row, dan bahaya SQL Injection vs Parameterized Query).
• Bagian 3 (Sedang Dibaca): Arsitektur Python OOP + Database: Mini Inventory App (Integrasi Class Repository, context manager with, error handling, dan antarmuka menu CLI).

1. Memisahkan Model Entitas dan Logika Database (Repository Pattern)

Alih-alih mengoper data sebagai Tuple atau Dictionary mentah, kita mendefinisikan dua kelas terpisah:

1. Class Produk (Model): Mewakili satu objek barang di dunia nyata dengan atribut (kode, nama, harga, stok) serta method pembantu seperti total_nilai().
2. Class ProdukRepository (Data Access): Bertanggung jawab khusus untuk semua urusan koneksi dan eksekusi SQL (tambah(), ambil_semua(), update(), hapus()).

# 01-model-dan-repository-oop.py
import sqlite3

# 1. Model Entitas
class Produk:
    def __init__(self, kode, nama, harga, stok, id=None):
        self.id = id
        self.kode = kode
        self.nama = nama
        self.harga = harga
        self.stok = stok

    def total_nilai(self):
        return self.harga * self.stok

    def __repr__(self):
        return f"<Produk [{self.kode}] {self.nama} - Rp {int(self.harga):,} (Stok: {self.stok})>"

# 2. Repository Database
class ProdukRepository:
    def __init__(self, db_path="aplikasi_inventaris.db"):
        self.db_path = db_path
        self._init_db()

    def _init_db(self):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
            CREATE TABLE IF NOT EXISTS produk (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                kode TEXT UNIQUE NOT NULL,
                nama TEXT NOT NULL,
                harga REAL NOT NULL,
                stok INTEGER NOT NULL DEFAULT 0
            );
            """)

    def tambah(self, p: Produk):
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute(
                "INSERT INTO produk (kode, nama, harga, stok) VALUES (?, ?, ?, ?)",
                (p.kode, p.nama, p.harga, p.stok)
            )
            p.id = cursor.lastrowid
            return p

    def ambil_semua(self):
        with sqlite3.connect(self.db_path) as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.cursor()
            cursor.execute("SELECT id, kode, nama, harga, stok FROM produk ORDER BY id ASC")
            rows = cursor.fetchall()
            return [Produk(r["kode"], r["nama"], r["harga"], r["stok"], id=r["id"]) for r in rows]

Dengan cara ini, jika suatu saat kamu ingin mengganti SQLite ke PostgreSQL atau MySQL, kamu hanya perlu mengubah isi method di dalam ProdukRepository tanpa merusak logika aplikasi utamamu.

2. Transaksi Atomik (Rollback) & Penanganan Error

Dalam sistem perbankan atau pergudangan, proses transfer barang melibatkan dua langkah: mengurangi stok dari gudang asal dan menambahkannya ke gudang tujuan. Jika langkah pertama berhasil tetapi komputer mati sebelum langkah kedua selesai, stok barang akan hilang ke antah berantah!

Konsep Transaksi Atomik (ACID) menjamin prinsip “All or Nothing”: semua langkah wajib sukses bersamaan. Jika ada satu langkah yang gagal, Python akan melakukan Rollback otomatis untuk mengembalikan database ke kondisi semula:

# 02-transaksi-dan-error-handling.py
import sqlite3

NAMA_DB = "aplikasi_inventaris.db"

def transfer_stok(kode_asal, kode_tujuan, jumlah):
    print(f"\n[TRANSAKSI] Memindahkan {jumlah} stok dari '{kode_asal}' ke '{kode_tujuan}'...")
    try:
        # Context manager 'with conn:' otomatis menangani commit atau rollback
        with sqlite3.connect(NAMA_DB) as conn:
            cursor = conn.cursor()
            
            # 1. Kurangi stok asal
            cursor.execute("UPDATE produk SET stok = stok - ? WHERE kode = ? AND stok >= ?", (jumlah, kode_asal, jumlah))
            if cursor.rowcount == 0:
                raise ValueError(f"Stok barang asal '{kode_asal}' tidak mencukupi!")

            # 2. Tambah stok tujuan
            cursor.execute("UPDATE produk SET stok = stok + ? WHERE kode = ?", (jumlah, kode_tujuan))
            if cursor.rowcount == 0:
                raise ValueError(f"Barang tujuan '{kode_tujuan}' tidak ditemukan!")
                
        print("[SUKSES] Seluruh transaksi berhasil disimpan (Committed)!")
        
    except (sqlite3.Error, ValueError) as err:
        print(f"[ROLLBACK] Transaksi dibatalkan secara aman! Alasan: {err}")

Selain itu, kita menggunakan blok try...except sqlite3.IntegrityError untuk menangkap kode produk duplikat secara elegan tanpa membuat program mendadak macet (*crash*).

3. Membangun Aplikasi Inventaris CLI Interaktif

Kini kita satukan semuanya ke dalam satu aplikasi berbasis menu teks (*Command Line Interface*) yang menyediakan fitur lengkap: penambahan barang, penampilan tabel rapi, pencarian kata kunci, update stok, hingga penghapusan barang:

# 03-aplikasi-inventaris-cli.py
import sqlite3

class DatabaseManager:
    def __init__(self, db_name="aplikasi_inventaris.db"):
        self.db_name = db_name
        self.init_tables()

    def init_tables(self):
        with sqlite3.connect(self.db_name) as conn:
            conn.execute("""
            CREATE TABLE IF NOT EXISTS produk (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                kode TEXT UNIQUE NOT NULL,
                nama TEXT NOT NULL,
                harga REAL NOT NULL,
                stok INTEGER NOT NULL DEFAULT 0
            );
            """)

    def tambah_produk(self, kode, nama, harga, stok):
        try:
            with sqlite3.connect(self.db_name) as conn:
                conn.execute(
                    "INSERT INTO produk (kode, nama, harga, stok) VALUES (?, ?, ?, ?)",
                    (kode.upper(), nama, harga, stok)
                )
            return True, f"Produk '{nama}' berhasil ditambahkan!"
        except sqlite3.IntegrityError:
            return False, f"Error: Kode '{kode}' sudah terdaftar!"

    def get_semua_produk(self):
        with sqlite3.connect(self.db_name) as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.cursor()
            cursor.execute("SELECT kode, nama, harga, stok FROM produk ORDER BY kode ASC")
            return cursor.fetchall()

    def cari_produk(self, kata_kunci):
        with sqlite3.connect(self.db_name) as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.cursor()
            query = "SELECT kode, nama, harga, stok FROM produk WHERE nama LIKE ? OR kode LIKE ?"
            cursor.execute(query, (f"%{kata_kunci}%", f"%{kata_kunci}%"))
            return cursor.fetchall()

    def update_stok(self, kode, stok_baru):
        with sqlite3.connect(self.db_name) as conn:
            cursor = conn.cursor()
            cursor.execute("UPDATE produk SET stok = ? WHERE kode = ?", (stok_baru, kode.upper()))
            if cursor.rowcount > 0:
                return True, f"Stok barang '{kode}' diperbarui menjadi {stok_baru}."
            return False, f"Barang dengan kode '{kode}' tidak ditemukan."

    def hapus_produk(self, kode):
        with sqlite3.connect(self.db_name) as conn:
            cursor = conn.cursor()
            cursor.execute("DELETE FROM produk WHERE kode = ?", (kode.upper(),))
            if cursor.rowcount > 0:
                return True, f"Produk '{kode}' berhasil dihapus."
            return False, f"Barang dengan kode '{kode}' tidak ditemukan."

class InventarisApp:
    def __init__(self):
        self.db = DatabaseManager()

    def cetak_tabel(self, rows):
        if not rows:
            print("  (Tidak ada data produk yang ditemukan)\n")
            return
        
        garis = "  +" + "-"*10 + "+" + "-"*30 + "+" + "-"*15 + "+" + "-"*8 + "+"
        header = f"  | {'Kode':<8} | {'Nama Barang':<28} | {'Harga (Rp)':<13} | {'Stok':<6} |"
        print(garis)
        print(header)
        print(garis)
        for r in rows:
            hrg = f"{int(r['harga']):,}".replace(",", ".")
            print(f"  | {r['kode']:<8} | {r['nama']:<28} | {hrg:<13} | {r['stok']:<6} |")
        print(garis)
        print(f"  Total Data: {len(rows)} item\n")

    def run(self):
        while True:
            print("\n" + "="*50)
            print("     SISTEM MANAJEMEN INVENTARIS TOKO (SQLITE)")
            print("="*50)
            print("  [1] Tambah Produk Baru")
            print("  [2] Lihat Daftar Semua Produk")
            print("  [3] Cari Produk (Nama / Kode)")
            print("  [4] Perbarui Stok Barang")
            print("  [5] Hapus Produk")
            print("  [0] Keluar Aplikasi")
            print("-"*50)
            
            pilih = input("  Pilih Menu [0-5]: ").strip()
            if pilih == "1":
                k = input("  Kode Barang: ").strip()
                n = input("  Nama Barang: ").strip()
                try:
                    h = float(input("  Harga Satuan (Rp): "))
                    s = int(input("  Stok Awal: "))
                    _, pesan = self.db.tambah_produk(k, n, h, s)
                    print(f"  -> {pesan}")
                except ValueError:
                    print("  -> Input harga dan stok harus angka!")
            elif pilih == "2":
                self.cetak_tabel(self.db.get_semua_produk())
            elif pilih == "3":
                kunci = input("  Kata Kunci: ").strip()
                self.cetak_tabel(self.db.cari_produk(kunci))
            elif pilih == "4":
                k = input("  Kode Barang: ").strip()
                s = int(input("  Stok Baru: "))
                _, pesan = self.db.update_stok(k, s)
                print(f"  -> {pesan}")
            elif pilih == "5":
                k = input("  Kode Barang yang dihapus: ").strip()
                _, pesan = self.db.hapus_produk(k)
                print(f"  -> {pesan}")
            elif pilih == "0":
                print("  Sampai jumpa! Semua perubahan tersimpan permanen.\n")
                break

if __name__ == "__main__":
    app = InventarisApp()
    app.run()

💡 Ingin Menguji Query SQL Secara Mandiri?
Kamu bisa mengeksplorasi dan mencoba berbagai skenario SQL secara langsung di browser tanpa install apapun melalui SQL Playground Interaktif (Klik di Sini)!

Kesimpulan Trilogi Serial Database di Python

Selamat! Kamu telah menyelesaikan seluruh rangkaian trilogi pembelajaran Database di Python:

1. Dari fondasi database SQLite dan operasi CRUD pertama…
2. Meningkat ke relasi multi-tabel (Foreign Key & INNER JOIN), agregasi bisnis, dan perlindungan dari celah SQL Injection…
3. Hingga menguasai arsitektur berorientasi objek (OOP), manajemen transaksi atomik dengan rollback, dan membangun aplikasi inventaris CLI yang utuh.

Kini kamu memiliki fondasi yang sangat kokoh untuk melangkah lebih jauh: menghubungkan database ini dengan antarmuka web modern (FastAPI / Flask) atau membangun dasbor analitik data yang canggih!