Kenapa Validasi Hasil Query Itu Penting?

Saat bekerja dengan database, menulis query SQL yang “jalan” saja tidak cukup. Query bisa saja tereksekusi tanpa error, tapi hasilnya salah kolom bernilai NULL yang seharusnya terisi, duplikasi data, jumlah baris yang tidak sesuai ekspektasi, atau tipe data yang tidak konsisten. Kesalahan seperti ini sering lolos dari pengecekan manual, terutama saat data sudah berjumlah ribuan hingga jutaan baris.

Di sinilah Python berperan. Dengan pustaka seperti pandas, sqlalchemy, dan pytest, kita bisa membangun proses validasi otomatis yang menjamin hasil query benar-benar sesuai standar kualitas data yang kita inginkan — baik untuk kebutuhan ETL, reporting, maupun QA sebelum data digunakan oleh tim lain.

Artikel ini akan membahas konsep validasi data hasil query SQL, jenis-jenis pengecekan yang umum dilakukan, dan tutorial praktis menggunakan Python yang bisa langsung kamu coba.

Jenis-Jenis Validasi Hasil Query

Sebelum masuk ke kode, ada baiknya memahami beberapa kategori validasi yang umum dilakukan:

Setiap kategori ini bisa diimplementasikan dengan kombinasi pandas untuk manipulasi data dan assert atau pytest untuk pengujian otomatis.

kamu juga dapat mempelajari hal yang berkaitan dengan SQL ini sihlakan masuk ke artikel dibawah ini

Kenapa Harus Pakai Python untuk Validasi Query SQL?

Sebenarnya SQL sendiri bisa melakukan sebagian validasi (misalnya COUNT(*), WHERE kolom IS NULL, dll). Tapi Python dipakai karena beberapa alasan:

1. Query SQL cuma menghasilkan data, bukan logika pengujian. SQL bagus untuk mengambil data, tapi tidak dirancang untuk membuat aturan “kalau begini, gagal; kalau begitu, lolos” secara terstruktur dan bisa dibaca ulang. Python punya assert, kondisi if/else, dan struktur program yang jauh lebih fleksibel untuk logika seperti ini.
2. Bisa dijadikan otomatis & berulang. Dengan Python, validasi bisa disimpan sebagai fungsi/test yang dijalankan berkali-kali — bahkan otomatis tiap ada perubahan data (lewat pytest, cron job, atau CI/CD). Kalau cuma pakai query SQL manual, kamu harus cek satu-satu tiap kali.
3. Menggabungkan banyak sumber data. Python bisa membandingkan hasil dari query database dengan file CSV, API, atau database lain sekaligus — sesuatu yang susah dilakukan murni pakai SQL.
4. Laporan & alert yang lebih kaya. Python bisa mencetak pesan error yang jelas, menyimpan log, kirim notifikasi (email/Slack), bahkan bikin dashboard — bukan cuma angka mentah dari SELECT.
5. Ekosistem library data yang kuat. pandas mempermudah manipulasi tabel (cek duplikat, null, rentang nilai) hanya dengan satu baris kode, dibanding menulis banyak query SQL terpisah untuk tiap jenis pengecekan.

Jadi intinya: SQL untuk mengambil data, Python untuk menguji dan memvalidasi data itu secara otomatis dan terstruktur.

Tools yang Digunakan

• Python 3.8+
• pandas — untuk memuat dan memanipulasi hasil query
• SQLAlchemy — untuk koneksi ke database
• pytest (opsional) — untuk menjalankan validasi sebagai test otomatis

Install semua tools ini lewat terminal VScode

Install dependency yang dibutuhkan:

pip install pandas sqlalchemy pytest

ini saya sudah install pandasnya jadi noticenya seperti ini :

Untuk driver database, sesuaikan dengan yang digunakan, contoh untuk PostgreSQL:

pip install psycopg2-binary

atau untuk MySQL:

pip install pymysql

Tutorial Praktis: Validasi Hasil Query SQL dengan Python

Kita akan membuat contoh dengan database SQLite agar tutorial ini bisa langsung dipraktekkan tanpa perlu setup server database.

Langkah 1 — Siapkan Database Contoh

Buat file setup_db.py untuk membuat database dan data contoh:

import sqlite3

conn = sqlite3.connect("toko.db")
cursor = conn.cursor()

cursor.execute("""
CREATE TABLE IF NOT EXISTS pelanggan (
    id INTEGER PRIMARY KEY,
    nama TEXT,
    email TEXT,
    umur INTEGER
)
""")

data_pelanggan = [
    (1, "Andi Saputra", "[email protected]", 28),
    (2, "Budi Santoso", "[email protected]", 34),
    (3, "Citra Dewi", None, 25),           # email kosong (sengaja untuk contoh)
    (4, "Dewi Lestari", "[email protected]", -5),  # umur tidak valid (sengaja untuk contoh)
    (5, "Andi Saputra", "[email protected]", 28),  # duplikat (sengaja untuk contoh)
]

cursor.executemany(
    "INSERT OR REPLACE INTO pelanggan VALUES (?, ?, ?, ?)", data_pelanggan
)

conn.commit()
conn.close()
print("Database 'toko.db' berhasil dibuat.")

begini isi codenya jika kamu sudah memasukan semua codenya :

lalu di terminal vscodenya jalankan :

python setup_db.py

Ini untuk hasilnya :

Langkah 2 — Jalankan Query dan Muat ke Pandas DataFrame

Buat file validasi_query.py:

lalu masukan code ini dan gabungkan code ini sampai langkah ke 8 :

import pandas as pd
from sqlalchemy import create_engine

# Buat koneksi ke database
engine = create_engine("sqlite:///toko.db")

# Query yang akan divalidasi
query = "SELECT * FROM pelanggan"
df = pd.read_sql(query, engine)

print(df)

Langkah 3 — Validasi Struktur (Schema)

Pastikan kolom yang dihasilkan query sesuai ekspektasi:

def validasi_struktur(df, kolom_wajib):
    kolom_hilang = set(kolom_wajib) - set(df.columns)
    if kolom_hilang:
        raise AssertionError(f"Kolom berikut tidak ditemukan: {kolom_hilang}")
    print("✅ Validasi struktur: OK")

validasi_struktur(df, ["id", "nama", "email", "umur"])

Langkah 4 — Validasi Jumlah Baris

def validasi_jumlah_baris(df, minimal=1):
    jumlah = len(df)
    assert jumlah >= minimal, f"Jumlah baris ({jumlah}) kurang dari minimal ({minimal})"
    print(f"✅ Validasi jumlah baris: OK ({jumlah} baris)")

validasi_jumlah_baris(df, minimal=1)

Langkah 5 — Validasi Nilai Kosong (Null Check)

def validasi_null(df, kolom_wajib):
    hasil = {}
    for kolom in kolom_wajib:
        jumlah_null = df[kolom].isnull().sum()
        hasil[kolom] = jumlah_null
        if jumlah_null > 0:
            print(f"⚠️  Kolom '{kolom}' memiliki {jumlah_null} nilai kosong")
        else:
            print(f"✅ Kolom '{kolom}': tidak ada nilai kosong")
    return hasil

validasi_null(df, ["nama", "email"])

Langkah 6 — Validasi Keunikan Data

def validasi_unik(df, kolom):
    duplikat = df[df.duplicated(subset=kolom, keep=False)]
    if not duplikat.empty:
        print(f"⚠️  Ditemukan {len(duplikat)} baris duplikat pada kolom '{kolom}':")
        print(duplikat)
    else:
        print(f"✅ Kolom '{kolom}': semua nilai unik")
    return duplikat

validasi_unik(df, "email")

Berikut tampilan code dari 3-6 yang sudah di gabungkan di file validasi_query.py

Langkah 7 — Validasi Rentang Nilai

def validasi_rentang(df, kolom, minimal, maksimal):
    tidak_valid = df[(df[kolom] < minimal) | (df[kolom] > maksimal)]
    if not tidak_valid.empty:
        print(f"⚠️  Ditemukan {len(tidak_valid)} baris dengan '{kolom}' di luar rentang [{minimal}, {maksimal}]:")
        print(tidak_valid)
    else:
        print(f"✅ Kolom '{kolom}': semua nilai dalam rentang valid")
    return tidak_valid

validasi_rentang(df, "umur", minimal=0, maksimal=120)

Langkah 8 — Menjalankan Semua Validasi Sekaligus

Gabungkan semua fungsi menjadi satu alur validasi lengkap:

import pandas as pd
from sqlalchemy import create_engine

def jalankan_validasi():
    engine = create_engine("sqlite:///toko.db")
    df = pd.read_sql("SELECT * FROM pelanggan", engine)

    print("=== Hasil Query ===")
    print(df, "\n")

    print("=== Mulai Validasi ===")
    validasi_struktur(df, ["id", "nama", "email", "umur"])
    validasi_jumlah_baris(df, minimal=1)
    validasi_null(df, ["nama", "email"])
    validasi_unik(df, "email")
    validasi_rentang(df, "umur", minimal=0, maksimal=120)
    print("=== Validasi Selesai ===")

if __name__ == "__main__":
    jalankan_validasi()

Jalankan:

python validasi_query.py

Karena data contoh sengaja dibuat dengan beberapa masalah (email kosong, umur negatif, dan data duplikat), kamu akan melihat output peringatan (⚠️) untuk baris-baris yang bermasalah — persis seperti yang akan terjadi jika kamu mendapati data kotor di dunia nyata.

Menjadikan Validasi sebagai Automated Test dengan Pytest

Agar validasi bisa dijalankan otomatis (misalnya di pipeline CI/CD), kita bisa mengubahnya menjadi test formal menggunakan pytest.

Buat file test_validasi.py:

import pandas as pd
import pytest
from sqlalchemy import create_engine

@pytest.fixture(scope="module")
def df():
    engine = create_engine("sqlite:///toko.db")
    return pd.read_sql("SELECT * FROM pelanggan", engine)

def test_kolom_tersedia(df):
    kolom_wajib = {"id", "nama", "email", "umur"}
    assert kolom_wajib.issubset(set(df.columns))

def test_jumlah_baris_tidak_kosong(df):
    assert len(df) > 0

def test_email_tidak_boleh_kosong(df):
    assert df["email"].isnull().sum() == 0, "Ada email yang kosong"

def test_email_harus_unik(df):
    duplikat = df[df.duplicated(subset="email", keep=False)]
    assert duplikat.empty, f"Ditemukan email duplikat:\n{duplikat}"

def test_umur_dalam_rentang_valid(df):
    tidak_valid = df[(df["umur"] < 0) | (df["umur"] > 120)]
    assert tidak_valid.empty, f"Ditemukan umur tidak valid:\n{tidak_valid}"

Jalankan dengan:

pytest test_validasi.py -v

Karena data contoh memang mengandung masalah, beberapa test akan gagal (FAILED) — ini menunjukkan bahwa validasi bekerja dengan benar dalam mendeteksi data yang tidak sesuai standar. Coba perbaiki data di setup_db.py, jalankan ulang setup_db.py, lalu jalankan kembali pytest untuk melihat semua test menjadi PASSED.

Maka hasilnya akan seperti ini, ini nanti menunjukankan 5 data dan akan memberikan kita jawabanan dimana yang kosong dan ada isinya (passed and failed)

Tips Tambahan untuk Validasi Data di Proyek Nyata

1. Bandingkan dengan sumber lain (cross-validation) — misalnya jumlah baris hasil query harus cocok dengan SELECT COUNT(*) dari tabel sumber.
2. Gunakan pandas.testing.assert_frame_equal jika ingin membandingkan dua DataFrame secara detail (berguna untuk regresi test setelah perubahan query).
3. Logging, bukan hanya print — di lingkungan produksi, gunakan modul logging agar hasil validasi tercatat dan bisa dipantau.
4. Jalankan di pipeline CI/CD — integrasikan test dengan GitHub Actions, GitLab CI, atau Airflow agar validasi berjalan otomatis setiap kali ada perubahan data atau query.
5. Gunakan library khusus data quality untuk kebutuhan yang lebih kompleks, seperti Great Expectations atau pandera, yang menyediakan framework validasi data yang lebih terstruktur dan bisa menghasilkan laporan otomatis.

Kesimpulan

Validasi hasil query SQL dengan Python membantu memastikan data yang kita gunakan benar-benar akurat, konsisten, dan bebas dari anomali sebelum digunakan untuk analisis maupun pengambilan keputusan. Dengan kombinasi pandas, SQLAlchemy, dan pytest, proses validasi bisa dilakukan secara terstruktur dan bahkan diotomatisasi sepenuhnya sebagai bagian dari pipeline data.

Pendekatan ini terutama bermanfaat dalam skenario seperti:

• QA sebelum data masuk ke dashboard atau laporan
• Proses ETL yang melibatkan transformasi data kompleks
• Regresi test setelah perubahan query atau skema database

Semakin awal masalah data terdeteksi, semakin murah biaya untuk memperbaikinya dan validasi otomatis seperti ini adalah cara paling efisien untuk mendeteksinya sejak dini.

Penulis

Junior Alfredo Benerd Setiawan