Kursus Python / Mengumpulkan JSON API ke SQLite tanpa duplikasi ID

Python

Dibuat 2026-11-12 Terakhir Diubah 2026-11-12

Kolektor API dapat menghindari penambahan duplikat pada eksekusi ulang dengan memvalidasi seluruh respons lalu menyimpan record berdasarkan ID eksternal yang stabil. Proyek ini menghubungkan pengambilan JSON dengan transaksi SQLite; ID yang sudah tersimpan diperbarui nama dan jumlahnya. Kebijakan identitas harus jelas agar pembaruan tidak menggabungkan transaksi yang berbeda.

Menentukan identitas, pembaruan, dan batas transaksi

id harus tetap menunjuk transaksi yang sama pada layanan sumber. Nama dan jumlah yang kebetulan sama tidak cukup untuk menyatakan dua transaksi identik. Program memakai upsert: menambahkan ID baru atau memperbarui record dengan ID yang sama. Jika satu respons memuat ID berulang, kemunculan terakhir menentukan nilai akhirnya.

Seluruh record diperiksa sebelum koneksi penyimpanan dibuka. Satu nilai tidak valid menggagalkan batch, bukan dilewati diam-diam. Setelah validasi selesai, pembuatan tabel dan penulisan batch berlangsung dalam satu transaksi database. Jaringan dan database lokal tetap dua tahap terpisah; transaksi database tidak mengunci atau membatalkan perubahan pada server API.

Gunakan Python 3.12 atau lebih baru karena contoh memakai autocommit=False. Modul json, urllib.request, dan sqlite3 tersedia pada instalasi Python yang menyertakan SQLite. Siapkan folder latihan baru agar database proyek lain tidak ikut diubah. Jika memakai database nyata dengan skema lama, lakukan migrasi tersendiri; CREATE TABLE IF NOT EXISTS tidak mengganti batas atau kolom yang sudah ada.

Menyiapkan respons lokal dan dua terminal

Simpan isi lengkap berikut sebagai expenses.json dengan UTF-8 dalam folder latihan:

[{"id":"e1","name":"coffee","amount":3500},{"id":"e2","name":"lunch","amount":8000}]

Buka terminal pertama pada folder tersebut, aktifkan lingkungan proyek, lalu mulai server latihan lokal:

python -m http.server 8000 --bind 127.0.0.1

Biarkan terminal pertama berjalan. Buka terminal kedua pada folder latihan yang sama untuk menjalankan kolektor. Jika port 8000 dipakai proses lain, pilih port kosong dan ubah nomor pada perintah server serta URL kolektor secara konsisten. Jangan menghentikan proses lain hanya untuk memakai nomor contoh.

Pengikatan ke 127.0.0.1 membatasi server ke komputer lokal. Jalankan server hanya pada folder latihan tanpa file pribadi; server menyajikan file dalam folder itu, bukan hanya JSON yang disebut dalam URL. Server latihan ini tidak menggantikan layanan API produksi.

Menjalankan kolektor lengkap dan membaca hasilnya

Simpan kode berikut sebagai collect.py dengan UTF-8:

import json
import sqlite3
import sys
from urllib.request import urlopen

with urlopen(sys.argv[1], timeout=5) as response:
    if response.status != 200:
        raise ValueError("unexpected status")
    raw = response.read(1_000_001)
    if len(raw) > 1_000_000:
        raise ValueError("response too large")
    data = json.loads(raw.decode("utf-8"))
if not isinstance(data, list) or len(data) > 10000:
    raise ValueError("invalid batch")
MAX_AMOUNT = 1_000_000_000_000
records = []
for row in data:
    if not isinstance(row, dict):
        raise ValueError("invalid row")
    key, name, amount = row.get("id"), row.get("name"), row.get("amount")
    if not isinstance(key, str) or not key.strip() or not isinstance(name, str) or not name.strip():
        raise ValueError("invalid identity")
    if type(amount) is not int or not 0 <= amount <= MAX_AMOUNT:
        raise ValueError("invalid amount")
    records.append((key, name, amount))
con = sqlite3.connect("collected.db", autocommit=False)
try:
    with con:
        con.execute("CREATE TABLE IF NOT EXISTS expenses (id TEXT PRIMARY KEY, name TEXT NOT NULL, amount INTEGER NOT NULL CHECK(amount BETWEEN 0 AND 1000000000000))")
        con.executemany("INSERT INTO expenses VALUES (?, ?, ?) ON CONFLICT(id) DO UPDATE SET name=excluded.name, amount=excluded.amount", records)
    count = con.execute("SELECT COUNT(*) FROM expenses").fetchone()[0]
    total = sum(row[0] for row in con.execute("SELECT amount FROM expenses"))
    print("stored:", count, "total:", total)
finally:
    con.close()

Pada terminal kedua dengan interpreter proyek yang sama, jalankan:

python collect.py http://127.0.0.1:8000/expenses.json

Hasil pertama adalah stored: 2 total: 11500. File collected.db dibuat pada folder kerja kolektor. Jalankan perintah yang sama sekali lagi; jumlah record tetap dua, bukan empat. Data yang dihitung pada keluaran adalah seluruh isi tabel saat itu, termasuk ID lama dari pengambilan terdahulu.

Program membatasi pembacaan menjadi paling banyak 1.000.001 byte untuk mendeteksi respons di atas 1.000.000 byte. Setelah decode UTF-8 dan pembacaan JSON, bentuk teratas harus list dengan paling banyak 10000 record. Setiap record harus dictionary dengan ID dan nama berupa string tidak kosong, serta jumlah integer persis dalam rentang 0 sampai 1.000.000.000.000. Boolean tidak diterima sebagai jumlah.

Pemeriksaan strip() memastikan ID dan nama tidak hanya berisi spasi, tetapi nilai yang disimpan tetap string asli. ID "e1" dan " e1 " masih berbeda. Jika sistem memerlukan normalisasi, tetapkan aturan itu sebelum pencocokan identitas dan uji dampaknya; jangan menganggap contoh sudah menghapus semua spasi.

Placeholder ? mengikat data sebagai parameter SQL. Blok with con melakukan commit saat penulisan berhasil dan rollback jika terjadi exception; finally menutup koneksi. Penjelasan batas tersebut tersedia dalam transaksi dan parameter SQLite.

Total dihitung dengan sum Python atas integer yang diambil dari tabel. SQLite SUM integer dapat melampaui rentang 64 bit ketika akumulasi cukup besar; total() berbasis floating point bukan pengganti yang setara untuk nilai uang tepat. Nilai yang masuk tetap dibatasi per record. Lihat aturan SUM dan total SQLite.

Menguji eksekusi ulang, kegagalan, dan kelengkapan

Jalankan respons asli dua kali dan pastikan kedua hasil tetap dua record dengan total 11500.
Ubah hanya jumlah e2 pada JSON menjadi 9000, simpan, lalu jalankan kolektor. Hasil harus dua record dengan total 12500.
Ubah jumlah itu menjadi string "9000". Kolektor harus gagal sebelum penulisan dan isi database sebelumnya tetap dua record dengan total 12500.
Kembalikan respons ke bentuk awal, jalankan lagi untuk memastikan pemulihan, lalu hentikan server milik latihan pada terminal pertama dengan Ctrl+C.

Kegagalan dapat berasal dari koneksi, status HTTP, decode UTF-8, JSON, validasi, atau penulisan database. Program contoh membiarkan traceback dan status gagal terlihat agar proses lanjutan tidak salah menganggap pekerjaan selesai. Periksa status keluar sebelum melanjutkan. Opsi timeout=5 berlaku pada operasi jaringan yang memblokir, bukan jaminan seluruh pekerjaan selalu selesai tepat dalam lima detik. Batas dan jenis kegagalan dijelaskan pada dasar klien HTTP dan API serta urlopen.

Respons kosong tidak menghapus record lama. Begitu juga ID yang tidak muncul pada respons terbaru tetap ada di database; kolektor ini memperbarui atau menambahkan, bukan menyelaraskan penghapusan. Aplikasi yang memerlukan salinan persis sumber harus menetapkan kebijakan penghapusan dan kelengkapan pengambilan secara terpisah.

Layanan luar juga memerlukan autentikasi, pembagian halaman, batas permintaan, dan kebijakan perubahan data. Beberapa halaman berhasil belum membuktikan seluruh koleksi selesai. Hindari percobaan ulang tanpa batas atau mengubah semua kegagalan menjadi list kosong karena keduanya dapat menyembunyikan data hilang. Rincian koneksi dan transaksi tersedia dalam dokumentasi sqlite3.

Python