Penanganan Error sqlite3

Galat pada kode database datang dari dua arah: sisi Python — parameter salah, koneksi sudah ditutup — dan sisi engine SQLite — constraint dilanggar, tabel tidak ada, berkas terkunci. Modul sqlite3 menerjemahkan semuanya menjadi exception Python dengan hierarki yang rapi, dan tugas kita adalah menangkap yang tepat: terlalu sempit membuat program crash untuk hal yang bisa dipulihkan, terlalu lebar menyembunyikan bug di balik except. Chapter ini memetakan hierarki tersebut, membaca isi pesan galatnya, lalu menangani kasus paling ditakuti: database is locked.

Hierarki Exception sqlite3 #

Semua exception modul sqlite3 turunan sqlite3.Error, yang terbagi menjadi dua cabang: InterfaceError untuk masalah di lapisan pembungkus Python, dan DatabaseError untuk galat dari engine SQLite. Cabang DatabaseError sendiri punya enam anak dengan penyebab khas:

Dua yang paling sering kita temui di praktik adalah IntegrityError dan OperationalError; ProgrammingError biasanya muncul di fase pengembangan (query belum benar), sedangkan DataError, InternalError, dan NotSupportedError langka.

Galat dari Engine #

Ketika engine SQLite menolak sebuah statement, modul sqlite3 membungkus pesannya menjadi exception yang sesuai. Contoh berikut memicu tiga galat berbeda dalam satu program:

import sqlite3

con = sqlite3.connect(":memory:")
cur = con.cursor()
cur.execute("CREATE TABLE pelanggan (id INTEGER PRIMARY KEY, nama TEXT NOT NULL UNIQUE)")
cur.execute("INSERT INTO pelanggan (nama) VALUES (?)", ("Budi",))

try:
    cur.execute("INSERT INTO pelanggan (nama) VALUES (?)", ("Budi",))
except sqlite3.IntegrityError as e:
    print("tipe     :", type(e).__name__)
    print("pesan    :", e)
    print("args     :", e.args)
    print("errorname:", e.sqlite_errorname)

try:
    cur.execute("SELECT * FROM pelanggan_lama")
except sqlite3.OperationalError as e:
    print("pesan    :", e)

try:
    cur.execute("SELECT * DARI pelanggan")
except sqlite3.OperationalError as e:
    print("pesan    :", e)

con.close()

Output yang Dihasilkan:

tipe     : IntegrityError
pesan    : UNIQUE constraint failed: pelanggan.nama
args     : ('UNIQUE constraint failed: pelanggan.nama',)
errorname: SQLITE_CONSTRAINT_UNIQUE
pesan    : no such table: pelanggan_lama
pesan    : near "DARI": syntax error

Empat hal layak dicatat dari output tersebut:

1. IntegrityError untuk pelanggaran UNIQUE: pesan UNIQUE constraint failed: pelanggan.nama menyebut tabel dan kolom yang bermasalah — format pesan SQLite stabil antar rilis, sehingga aman untuk diperiksa isinya.
2. e.args: exception membawa payload berupa tuple; e.args[0] berisi string pesan yang sama dengan str(e).
3. e.sqlite_errorname: kode galat SQLite dalam bentuk nama — di sini SQLITE_CONSTRAINT_UNIQUE (kode numeriknya tersedia di e.sqlite_errorcode, yaitu 2067).
4. OperationalError untuk dua kasus berbeda: tabel yang tidak ada dan kesalahan sintaks SQL sama-sama berakhir di OperationalError — keduanya berasal dari kode galat engine SQLITE_ERROR, bukan salah pemrograman Python.

Menangkap Galat dengan Tepat #

Aturan mainnya: tangkap exception yang paling spesifik, dan hanya pulihkan kasus yang memang dipahami. Pola berikut menangani pelanggaran nama unik sebagai situasi normal (“pelanggan sudah terdaftar”) sambil melempar ulang galat lain yang tidak dikenalnya:

import sqlite3

con = sqlite3.connect(":memory:")
cur = con.cursor()
cur.execute("CREATE TABLE pelanggan (id INTEGER PRIMARY KEY, nama TEXT NOT NULL UNIQUE)")
cur.execute("INSERT INTO pelanggan (nama) VALUES (?)", ("Budi",))


def daftar_pelanggan_baru(nama):
    try:
        cur.execute("INSERT INTO pelanggan (nama) VALUES (?)", (nama,))
        return True
    except sqlite3.IntegrityError as e:
        if "UNIQUE constraint failed" in str(e):
            return False
        raise


print(daftar_pelanggan_baru("Siti"))
print(daftar_pelanggan_baru("Budi"))

con.close()

Output yang Dihasilkan:

True
False

Fungsi daftar_pelanggan_baru() mengembalikan True saat baris tersimpan dan False saat nama sudah dipakai orang lain. Dua detail pola ini yang membuatnya sehat:

1. except sqlite3.IntegrityError: hanya exception constraint yang ditangkap — OperationalError akibat salah ketik query tetap meledak dan terlihat saat pengembangan.
2. if "UNIQUE constraint failed" in str(e): ... raise: IntegrityError juga muncul untuk pelanggaran NOT NULL atau FOREIGN KEY; pemeriksaan isi pesan memastikan hanya kasus UNIQUE yang dipulihkan, sisanya dilempar ulang dengan raise telanjang untuk tetap membawa traceback asli.

database is locked #

SQLite mengunci seluruh berkas database saat satu koneksi memegang transaction tulis. Koneksi kedua yang ikut mencoba menulis akan ditolak dengan OperationalError: database is locked — kecuali koneksi tersebut disuruh menunggu. Parameter timeout pada connect() mengatur lama menunggu (bawaannya 5 detik); pragma busy_timeout mengaturnya lewat SQL dalam satuan milidetik:

import os
import sqlite3
import threading
import time

if os.path.exists("toko.db"):
    os.remove("toko.db")

penulis = sqlite3.connect("toko.db", check_same_thread=False)
penulis.execute("CREATE TABLE pesanan (id INTEGER PRIMARY KEY, isi TEXT)")
penulis.commit()
penulis.execute("INSERT INTO pesanan (isi) VALUES (?)", ("kopi gayo",))
# kunci tulis masih dipegang penulis karena belum commit

pemaksa = sqlite3.connect("toko.db", timeout=0)
try:
    pemaksa.execute("INSERT INTO pesanan (isi) VALUES (?)", ("gula aren",))
except sqlite3.OperationalError as e:
    print("OperationalError:", e)

pemaksa.execute("PRAGMA busy_timeout = 300")


def lepas_kunci():
    time.sleep(0.1)
    penulis.rollback()


threading.Thread(target=lepas_kunci).start()

mulai = time.monotonic()
pemaksa.execute("INSERT INTO pesanan (isi) VALUES (?)", ("gula aren",))
pemaksa.commit()
print("berhasil setelah menunggu", round(time.monotonic() - mulai, 2), "detik")
print("jumlah baris:", pemaksa.execute("SELECT COUNT(*) FROM pesanan").fetchone()[0])

penulis.close()
pemaksa.close()

Output yang Dihasilkan:

OperationalError: database is locked
berhasil setelah menunggu 0.1 detik
jumlah baris: 1

Alur program tersebut:

1. Koneksi penulis membuka transaction tulis dengan INSERT yang belum di-commit, sehingga memegang kunci database. (Argumen check_same_thread=False hanya diperlukan karena threading memakai koneksi ini dari thread lain.)
2. Koneksi pemaksa dengan timeout=0 langsung ditolak tanpa jeda — itulah pesan database is locked.
3. PRAGMA busy_timeout = 300 menyuruh koneksi pemaksa menunggu hingga 300 milidetik ketika berkas terkunci.
4. Thread lepas_kunci melepas kunci 0,1 detik kemudian lewat rollback() — dan INSERT pemaksa berhasil setelah menunggu kira-kira selama itu.

Dengan koneksi aman dan galat terkendali, kita siap menyusun kode menjadi lapisan yang rapi: satu modul khusus yang memusatkan seluruh akses database.