Data pembayaran toko online tidak berada di database kita — payment gateway menyimpannya sendiri dan hanya menyediakan akses lewat REST API di atas HTTP. Pola ini berlaku hampir di semua layanan pihak ketiga: kurir, marketplace, layanan pengiriman, semuanya membuka data lewat endpoint yang mengembalikan JSON. Ingestion dari API berarti menarik data lewat permintaan HTTP, mengurai respons, dan menangani kegagalan dengan rapi karena jaringan dan batasan layanan adalah kenyataan harian. Kita akan memakai library requests yang sudah terpasang, dan menguji semuanya lewat server tiruan lokal sehingga tidak membutuhkan koneksi internet.

REST dan JSON dalam Ringkas #

REST API menyediakan data di alamat bernama endpoint — misalnya /api/transaksi — dan client mengaksesnya dengan method HTTP. Method GET berarti membaca data, POST mengirim data; sebagian besar pekerjaan ingestion memakai GET. Respons REST hampir selalu berformat JSON: teks terstruktur yang bisa berupa objek bertingkat dan array, yang di Python langsung menjadi dictionary dan list setelah diurai. Satu endpoint biasanya juga membatasi jumlah baris per respons lewat pagination — pembahasannya kita khususkan di chapter Menangani Pagination API.

Library requests dan Autentikasi Header #

Library requests adalah standar de facto untuk HTTP di Python. Satu panggilan requests.get(url, headers=..., timeout=...) mengirim permintaan dan mengembalikan objek respons lengkap dengan kode status, header, dan isi. Akses API hampir selalu dilindungi token yang dikirim lewat header Authorization dengan skema Bearer — pola Bearer YOUR_API_TOKEN ini dibaca server lalu dicocokkan ke daftar token yang sah.

Penanganan Galat Dasar #

Permintaan HTTP gagal dalam dua rupa. Pertama, server menjawab dengan kode status galat — 401 untuk token salah, 404 untuk endpoint tidak ada, 429 untuk terlalu banyak permintaan. Metode raise_for_status() pada respons memeriksa kode tersebut dan memicu requests.exceptions.HTTPError sehingga galat terdeteksi sebelum data diurai. Kedua, permintaan tidak pernah sampai — server mati atau jaringan putus — yang memicu requests.exceptions.ConnectionError. Keduanya wajib ditangani karena pipeline produksi harus tetap hidup ketika layanan pihak ketiga sedang bermasalah.

Implementasi: Server Tiruan Lokal #

Kita bangun server tiruan dari modul http.server stdlib yang berjalan di thread, menyajikan JSON transaksi di 127.0.0.1, lalu menariknya dengan requests — persis pola menarik payment gateway sungguhan, hanya saja semua terjadi di dalam satu mesin:

python 
import json
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer

import requests

DATA_TRANSAKSI = {
    "transaksi": [
        {"id": "TRX-001", "pelanggan": "Budi Santoso", "total": 150000},
        {"id": "TRX-002", "pelanggan": "Siti Rahma", "total": 275000},
        {"id": "TRX-003", "pelanggan": "Dewi Lestari", "total": 98000},
    ]
}


class HandlerTiruan(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == "/api/transaksi":
            body = json.dumps(DATA_TRANSAKSI).encode()
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)
        else:
            self.send_response(404)
            self.send_header("Content-Length", "0")
            self.end_headers()

    def log_message(self, format, *args):
        pass  # matikan log bawaan agar keluaran bersih


server = HTTPServer(("127.0.0.1", 0), HandlerTiruan)  # port 0: port bebas
threading.Thread(target=server.serve_forever, daemon=True).start()
url = f"http://127.0.0.1:{server.server_address[1]}"

try:
    respons = requests.get(
        f"{url}/api/transaksi",
        headers={"Authorization": "Bearer YOUR_API_TOKEN"},
        timeout=5,
    )
    respons.raise_for_status()
    data = respons.json()
    print(f"Status: {respons.status_code}, jumlah transaksi: {len(data['transaksi'])}")
    for t in data["transaksi"]:
        print(f"  {t['id']} | {t['pelanggan']} | Rp{t['total']}")

    print("\nMenguji endpoint yang tidak ada:")
    respons_salah = requests.get(f"{url}/api/tidak-ada", timeout=5)
    try:
        respons_salah.raise_for_status()
    except requests.exceptions.HTTPError as galat:
        print(f"Permintaan gagal dengan status {galat.response.status_code}")
finally:
    server.shutdown()
    server.server_close()

Output yang Dihasilkan:

text 
Status: 200, jumlah transaksi: 3
  TRX-001 | Budi Santoso | Rp150000
  TRX-002 | Siti Rahma | Rp275000
  TRX-003 | Dewi Lestari | Rp98000

Menguji endpoint yang tidak ada:
Permintaan gagal dengan status 404

Mari kita telaah baris-baris kode di atas:

  1. Baris 16-28 (HandlerTiruan): handler menjawab 200 beserta isi JSON untuk path /api/transaksi, dan 404 untuk path lain — meniru perilaku API sungguhan termasuk jalur gagalnya.
  2. Baris 19-24 (tulis respons): isi di-serialize menjadi bytes, panjangnya diberitahu lewat header Content-Length, lalu ditulis ke socket — protokol HTTP menuntut ketiga langkah ini berurutan.
  3. Baris 34-35 (server di thread): HTTPServer dengan port 0 meminta sistem operasi memilih port bebas; serve_forever dijalankan sebagai thread daemon agar blok utama tetap bisa bekerja.
  4. Baris 39-43 (requests.get): permintaan membawa header autentikasi dan timeout=5 — tanpa timeout, permintaan bermasalah bisa menggantung tanpa batas.
  5. Baris 44-45 (raise_for_status dan json()): kode status diperiksa sebelum penguraian; respons.json() mengubah JSON menjadi struktur Python.
  6. Baris 56-58 (finally): server dimatikan dengan shutdown() dan server_close() apa pun hasilnya, sehingga proses berakhir bersih tanpa thread yang tertinggal.

Pola server tiruan seperti ini juga menjadi fondasi pengujian: skrip bisa dijalankan berulang tanpa koneksi internet dan hasilnya selalu sama. Dokumentasi resminya ada di requests.readthedocs.io dan http.server .

Menarik data lewat GET menempatkan kita sebagai peminta — pola pull. Beberapa layanan membalik arah tersebut dan justru mengirim event begitu terjadi, dan menerima pengiriman itu menuntut endpoint penerima milik kita sendiri, dibahas di chapter Ingestion dari Webhook.