Proyek: Pipeline Ingestion API

Pada proyek sebelumnya, Proyek: Pipeline Data CSV ke PostgreSQL, sumber data kita berupa berkas CSV yang sudah tersedia lengkap di disk. Kali ini sumber datanya hidup: sebuah REST API publik yang memotong respons menjadi halaman-halaman, membatasi jumlah permintaan, dan sesekali mengalami galat. Kita akan membangun pipeline ingestion yang mengambil data pesanan dari API publik fiktif TokoKita, memvalidasi bentuk responsnya, lalu menyimpan data mentah ke zona raw dalam format JSON Lines dengan dukungan mode incremental. Lima masalah khas ingestion dari API — pagination, rate limit, galat sementara, perubahan schema, dan pengambilan ulang data yang tidak berubah — kita selesaikan satu per satu pada tiap bagian proyek.

Arsitektur Ringkas #

Alur pipeline mengikuti pola extract yang tangguh sebelum data disentuh oleh transformasi apa pun:

flowchart LR A[ApiTokoKita] --> B[ApiClient timeout + retry] B --> C[paginate penjelajah halaman] C --> D[validator schema pesanan] D --> E[(zona raw JSON Lines)] E --> F[state kursor updated_at] F -. updated_since .-> A

Data mentah disimpan apa adanya di zona raw, sementara state kursor mencatat posisi terakhir ingestion agar run berikutnya bersifat incremental. Konsep zone dan cursor dibahas pada chapter Konsep Data Ingestion dan Incremental Ingestion.

Sumber Data API TokoKita #

API pesanan TokoKita bersifat fiktif dan menyediakan dua endpoint dengan kontrak berikut:

Endpoint	Parameter	Keterangan
`GET /api/v1/pesanan`	`page`, `per_page`	Daftar pesanan terpaginasi dengan envelope `data` + `meta`
`GET /api/v1/pesanan`	`updated_since`	Hanya pesanan dengan `updated_at` lebih baru dari nilai parameter
`GET /api/v1/health`	-	Pemeriksaan kesehatan sederhana

Server juga mensimulasikan perilaku API publik sesungguhnya: respons 429 dengan header Retry-After saat kuota terlampaui, galat 500 sesaat, dan tolakan 400 untuk parameter tidak valid.

Perilaku yang Diharapkan #

Sebuah eksekusi lengkap dijalankan dari akar bundle proyek dan mencetak ringkasan run berikut:

bash

python3 -m src.pipeline --mode full --demo

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

text
=== Ringkasan Run ===
mode           : full
run_id         : 20261017T074428Z
baris_diambil  : 137
baris_duplikat : 0
baris_valid    : 137
baris_dilewati : 0
output         : data/raw/pesanan-20261017T074428Z.jsonl
kursor         : 2024-01-05T16:16:00+00:00

Run incremental berikutnya hanya mengambil pesanan baru sejak kursor terakhir:

bash

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

Output yang Dihasilkan (bagian ringkasan):

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

Seluruh perilaku pipeline dikunci oleh 52 tes pytest yang menjalankan fake server di localhost, sehingga setiap perubahan kode langsung terdeteksi.

Struktur Proyek #

text
pipeline-ingestion-api/
├── _index.md
├── part-1 ... part-9/
├── data/                  # zona raw JSON Lines + contoh respons
├── src/                   # fake_api, client, pagination, retry,
│                          # validator, storage, incremental, pipeline
├── tests/                 # conftest + 8 berkas tes pytest
└── config/                # pipeline.yaml, schema_pesanan.json, state kursor

Seluruh kode sumber proyek tersedia pada bagian Berkas Sumber di bawah halaman ini. Setiap bagian pengerjaan merujuk ke berkas yang sama, sehingga kondisi proyek tetap utuh di setiap tahap.

Tahapan Pengerjaan #

Proyek dikerjakan dalam sembilan bagian berurutan dan setiap bagian menghasilkan kode yang dapat dijalankan:

Bagian 1 memahami kontrak API sumber dan menjalankan fake server lokal.
Bagian 2 membangun klien HTTP dengan session, timeout, dan klasifikasi galat.
Bagian 3 menjelajahi seluruh halaman respons dengan pola iterator.
Bagian 4 menghormati rate limit lewat header Retry-After.
Bagian 5 menerapkan retry dengan exponential backoff dan jitter.
Bagian 6 memvalidasi schema respons sebelum data dipercaya.
Bagian 7 menyimpan data mentah ke zona raw dalam format JSON Lines.
Bagian 8 membuat ingestion incremental dengan state kursor updated_at.
Bagian 9 menguji seluruh pipeline secara menyeluruh dengan pytest.

Mulailah dari Bagian 1 karena setiap bagian dibangun di atas hasil bagian sebelumnya.

# Bagian 1: Memahami API Sumber

Pelajari kontrak API pesanan TokoKita dan jalankan fake server lokal sebagai sumber data yang dapat direproduksi.

Buka → # Bagian 2: Mengambil Data dari API

Bangun klien HTTP dengan requests: session, timeout, dan pemisahan galat permanen dari galat sementara.

Buka → # Bagian 3: Menangani Pagination

Jelajahi seluruh halaman respons API dengan pola iterator dan pagar pengaman max_pages.

Buka → # Bagian 4: Menangani Rate Limit

Hormati batas kuota API: tangani respons 429 dan header Retry-After tanpa membebani server.

Buka → # Bagian 5: Menerapkan Retry

Terapkan percobaan ulang dengan exponential backoff dan jitter untuk galat sementara.

Buka → # Bagian 6: Memvalidasi Respons

Validasi schema respons sebelum data disimpan agar perubahan API terdeteksi lebih awal.

Buka → # Bagian 7: Menyimpan Data Mentah

Simpan data mentah ke zona raw dalam format JSON Lines, satu berkas per run yang tidak pernah tertimpa.

Buka → # Bagian 8: Membuat Ingestion Incremental

Lanjutkan ingestion dari kursor updated_at sehingga setiap run hanya mengambil data baru.

Buka → # Bagian 9: Menguji Pipeline

Uji seluruh pipeline dengan pytest: unit test fungsi murni, integrasi fake server, dan end-to-end.

Buka →