Mode full pada Bagian 7 mengunduh seluruh 137 pesanan setiap eksekusi. Untuk dataset yang tumbuh terus, kebiasaan tersebut menghabiskan kuota API dan waktu run hanya untuk mengambil ulang data lama yang tidak berubah. Kabar baiknya, setiap pesanan membawa stempel updated_at, sehingga kita bisa mencatat posisi terakhir ingestion dan hanya meminta yang lebih baru. Pada bagian ini kita membangun state kursor di src/incremental.py dan mode incremental di pipeline. Perbandingan lengkap dua pendekatan ini dibahas pada chapter Full Load vs Incremental Load.

State Kursor yang Atomik #

State disimpan sebagai berkas JSON kecil dan ditulis secara atomik:

python 
def save_state(path, state):
    """Tulis state secara atomik: tulis ke ``.tmp`` lalu ganti nama berkas.

    Penggantian nama lewat ``os.replace`` bersifat atomik di satu filesystem,
    sehingga pipeline tidak pernah meninggalkan state setengah tertulis ketika
    proses terhenti di tengah penulisan.
    """
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    sementara = path.with_suffix(path.suffix + ".tmp")
    with sementara.open("w", encoding="utf-8") as berkas:
        json.dump(state, berkas, indent=2, ensure_ascii=False)
        berkas.write("\n")
    os.replace(sementara, path)

Mari kita telaah baris-baris kode di atas:

  1. Baris 31 (with_suffix(path.suffix + ".tmp")): Penulisan diarahkan ke berkas sementara state.json.tmp, bukan langsung ke berkas state.
  2. Baris 32-34 (json.dump): Seluruh state ditulis lengkap ke berkas sementara dan ditutup rapi oleh context manager.
  3. Baris 35 (os.replace): Penggantian nama dalam satu filesystem adalah operasi atomik menurut dokumentasi Python; pembaca tidak pernah melihat berkas setengah tertulis, dan proses yang terhenti di tengah jalan meninggalkan state lama yang tetap valid.

Sifat atomik penting karena state adalah kesepakatan antara run: kursor yang setengah tertulis bisa membuat run berikutnya melompati data. Pembandingannya dengan mode write langsung dibahas pada chapter Incremental Ingestion.

Kursor dan Dedup #

Dua fungsi kecil menjaga agar kursor selalu maju dan data tidak dobel:

python 
def next_cursor(rows, sebelumnya=None):
    """Kursor berikutnya: ``updated_at`` terbesar, tidak pernah mundur.

    Semua stempel waktu memakai format ISO 8601 seragam dari API, sehingga
    perbandingan teks memberi hasil yang sama dengan perbandingan waktu.
    """
    kandidat = [baris["updated_at"] for baris in rows if baris.get("updated_at")]
    if sebelumnya:
        kandidat.append(sebelumnya)
    return max(kandidat) if kandidat else sebelumnya


def dedupe_by_id(rows):
    """Buang duplikat ``id``; simpan baris dengan ``updated_at`` terbaru.

    Sumber duplikat paling umum: pesanan yang diperbarui tepat saat pagination
    berjalan sehingga muncul di dua halaman. Hasil mempertahankan urutan
    kemunculan pertama tiap ``id``.
    """
    terpilih = {}
    for baris in rows:
        pid = baris.get("id")
        if pid not in terpilih or \
                baris.get("updated_at", "") >= terpilih[pid].get("updated_at", ""):
            terpilih[pid] = baris
    return list(terpilih.values())

next_cursor mengambil updated_at terbesar dari baris yang baru ditulis dan tidak pernah mundur, karena kursor lama ikut menjadi kandidat. dedupe_by_id menjaga idempotensi: bila satu pesanan muncul di dua halaman karena diperbarui di tengah penjelajahan, hanya versi dengan updated_at terbaru yang disimpan. Kombinasi kursor plus dedup adalah pola dasar chapter Idempotent Ingestion.

Alur Incremental di Pipeline #

Mode incremental hanya berbeda pada tiga titik: kursor lama dibaca dari state, parameter updated_since disertakan pada setiap halaman, dan state disimpan setelah penulisan berhasil:

python 
        rid = run_id()
        state = load_state(self.state_path)
        kursor_lama = state.get("updated_at") if mode == "incremental" else None
        logger.info("run %s mode=%s kursor=%s", rid, mode, kursor_lama or "-")
python 
    def _fetch_page(self, kursor):
        """Bungkus client menjadi fungsi per-halaman untuk ``paginate``."""
        def fetch(nomor):
            params = {"page": nomor, "per_page": self.page_size}
            if kursor:
                params["updated_since"] = kursor
            return self.client.get_json(f"/api/v1/{self.dataset}", params)
        return fetch

Bila state masih kosong, kursor_lama bernilai None, parameter updated_since tidak dikirim, dan run incremental otomatis menjadi full load pertama. Urutan langkah pada run sengaja menempatkan save_state setelah write_jsonl, sehingga kursor hanya maju bila data benar-benar sudah ada di zona raw.

Run Incremental Nyata #

Setelah run full pada Bagian 7, kursor tersimpan di config/state.json. Run incremental pertama tidak menemukan perubahan apa pun:

bash 
python3 -m src.pipeline --mode incremental --demo

Output yang Dihasilkan:

text 
INFO ingestion.pipeline: fake API demo berjalan di http://127.0.0.1:41131
INFO ingestion.pipeline: run 20261017T074432Z mode=incremental kursor=2024-01-05T16:16:00+00:00
INFO ingestion.pagination: halaman 1/1: 0 baris
INFO ingestion.pipeline: run 20261017T074432Z selesai: 0 diambil, 0 ditulis, kursor=2024-01-05T16:16:00+00:00
=== Ringkasan Run ===
mode           : incremental
run_id         : 20261017T074432Z
baris_diambil  : 0
baris_duplikat : 0
baris_valid    : 0
baris_dilewati : 0
output         : -
kursor         : 2024-01-05T16:16:00+00:00

Nol baris diambil, tidak ada berkas baru, dan kursor tidak berubah. Selanjutnya kita mensimulasikan lima pesanan baru yang masuk setelah kursor, lalu menjalankan incremental lagi:

bash 
python3 -m src.pipeline --mode incremental --demo --demo-extra 5

Output yang Dihasilkan (bagian ringkasan; log INFO di atasnya dipangkas):

text 
=== Ringkasan Run ===
mode           : incremental
run_id         : 20261017T074435Z
baris_diambil  : 5
baris_duplikat : 0
baris_valid    : 5
baris_dilewati : 0
output         : data/raw/pesanan-20261017T074435Z.jsonl
kursor         : 2024-02-01T14:22:00+00:00

Lima pesanan baru ditulis ke berkas run tersendiri dan kursor maju ke updated_at terbesar. Isi state sekarang terbaca langsung:

bash 
cat config/state.json

Output yang Dihasilkan:

text 
{
  "updated_at": "2024-02-01T14:22:00+00:00",
  "last_run_id": "20261017T074435Z",
  "last_mode": "incremental"
}

Uji Incremental #

Tes di tests/test_incremental.py memeriksa state dan kursor sebagai fungsi murni, termasuk penulisan atomik yang tidak menyisakan berkas sementara:

python 
def test_next_cursor_mengambil_updated_at_terbesar():
    baris = [{"updated_at": "2024-01-02T00:00:00+00:00"},
             {"updated_at": "2024-01-05T00:00:00+00:00"}]
    assert next_cursor(baris) == "2024-01-05T00:00:00+00:00"


def test_next_cursor_tidak_pernah_mundur():
    baris = [{"updated_at": "2024-01-02T00:00:00+00:00"}]
    assert next_cursor(baris, "2024-01-05T00:00:00+00:00") == \
        "2024-01-05T00:00:00+00:00"
bash 
python3 -m pytest tests/test_incremental.py -q

Output yang Dihasilkan:

text 
.......                                                                  [100%]
7 passed in 0.12s

Pipeline kita kini hemat: mengambil hanya yang berubah, menulis tanpa menimpa, dan maju dari kursor terakhir. Satu langkah terakhir memastikan seluruh perilaku ini tidak mundur tanpa sepengetahuan kita, yaitu rangkaian pengujian menyeluruh pada bagian penutup. Jaminan atomik os.replace terdokumentasi di dokumentasi modul os Python .