API berkembang: kolom baru ditambahkan, tipe data diubah, dan kontrak yang tadinya tersirat mulai bergeser. Bayangkan TokoKita mengubah field total dari angka menjadi teks "Rp1.150.000" tanpa memberi tahu; pipeline kita akan menyimpan data yang tampak baik-baik saja tetapi merusak setiap analisis di hilir. Validasi pada saat ingestion memindahkan penemuan masalah dari minggu depan ke menit ini. Pada bagian ini kita membangun validator schema ringan di src/validator.py.

Schema Pesanan #

Kontrak bentuk satu baris pesanan dinyatakan sebagai JSON di config/schema_pesanan.json:

json 
{
  "type": "object",
  "required": ["id", "pelanggan", "email", "total", "status", "created_at", "updated_at"],
  "properties": {
    "id": {
      "type": "integer"
    },
    "pelanggan": {
      "type": "string",
      "min_length": 1
    },
    "email": {
      "type": "string",
      "pattern": "^[^@\\s]+@[^@\\s]+\\.[^@\\s]+$"
    },
    "total": {
      "type": "number",
      "minimum": 0
    },
    "status": {
      "type": "string",
      "enum": ["baru", "dibayar", "dikirim", "selesai", "dibatalkan"]
    },
    "created_at": {
      "type": "string",
      "format": "iso-datetime"
    },
    "updated_at": {
      "type": "string",
      "format": "iso-datetime"
    }
  }
}

Schema memakai subset kata kunci yang lazim di keluarga JSON Schema: type, required, properties, enum, minimum, min_length, pattern, dan format khusus iso-datetime. Menyimpan schema sebagai berkas terpisah membuat perubahan kontrak cukup dilakukan di satu tempat, tanpa menyentuh kode Python.

Validator Ringan #

Fungsi validate memeriksa satu instance terhadap satu subschema secara rekursif:

python 
def validate(instance, schema, path="pesanan"):
    """Validasi ``instance`` terhadap ``schema``; kembalikan daftar pesan galat.

    Daftar kosong berarti valid. Pesan galat menyertakan lokasi field
    (mis. ``data[3].total``) agar mudah ditelusuri.
    """
    galat = []
    jenis = schema.get("type")
    if jenis and not _cocok_jenis(instance, jenis):
        return [f"{path}: type harus {jenis}, didapat {_nama_jenis(instance)}"]

    if "enum" in schema and instance not in schema["enum"]:
        galat.append(f"{path}: nilai {instance!r} tidak ada di enum {schema['enum']}")
    if "minimum" in schema and isinstance(instance, (int, float)) \
            and not isinstance(instance, bool) and instance < schema["minimum"]:
        galat.append(
            f"{path}: nilai {instance} di bawah minimum {schema['minimum']}"
        )

    if isinstance(instance, str):
        if "min_length" in schema and len(instance) < schema["min_length"]:
            galat.append(
                f"{path}: panjang {len(instance)} di bawah min_length "
                f"{schema['min_length']}"
            )
        pola = schema.get("pattern")
        if pola and re.search(pola, instance) is None:
            galat.append(
                f"{path}: nilai {instance!r} tidak cocok dengan pattern {pola}"
            )
        if schema.get("format") == FORMAT_ISO:
            try:
                datetime.fromisoformat(instance)
            except ValueError:
                galat.append(f"{path}: nilai {instance!r} bukan ISO 8601 yang valid")

    if isinstance(instance, dict):
        for kunci in schema.get("required", []):
            if kunci not in instance:
                galat.append(f"{path}: field wajib {kunci!r} tidak ada")
        for kunci, subschema in (schema.get("properties") or {}).items():
            if kunci in instance:
                galat += validate(instance[kunci], subschema, f"{path}.{kunci}")

    if isinstance(instance, list):
        subschema = schema.get("items")
        if subschema:
            for indeks, elemen in enumerate(instance):
                galat += validate(elemen, subschema, f"{path}[{indeks}]")
    return galat

Mari kita telaah baris-baris kode di atas:

  1. Baris 54-56 (type tidak cocok): Ketidakcocokan tipe langsung mengembalikan satu pesan galat tanpa memeriksa aturan lain, karena aturan enum atau minimum tidak bermakna untuk tipe yang salah.
  2. Baris 58-64 (enum dan minimum): Nilai di luar daftar tetap atau di bawah batas bawah dicatat dengan nilai aslinya agar mudah diaudit.
  3. Baris 72-76 (pattern): Ekspresi reguler diperiksa dengan re.search, sama seperti perilaku JSON Schema.
  4. Baris 77-81 (format iso-datetime): Stempel waktu dites dengan datetime.fromisoformat, sehingga "besok-pagi" tertolak meskipun bertipe string.
  5. Baris 83-89 (required dan properties): Field wajib yang hilang dicatat satu per satu, lalu setiap field yang ada divalidasi rekursif dengan path yang bertambah, misalnya data[0].email.
  6. Baris 91-95 (items): Schema berbentuk array memvalidasi setiap elemennya dengan indeks pada path, misalnya data[3]; cabang inilah yang dipakai validate_rows untuk memeriksa seluruh baris envelope.

Pembantu validate_rows menerapkan validate pada daftar baris dan hanya mengembalikan baris yang bermasalah beserta indeksnya, sehingga pemanggil langsung tahu berapa baris yang dikecualikan.

Deteksi Perubahan Schema #

Skrip berikut mengambil satu pesanan dari contoh respons di data/contoh_pesanan_halaman1.json, lalu merusak tiga field untuk mensimulasikan API yang berubah bentuk:

bash 
python3 - <<'PY'
import json

from src.validator import muat_schema, validate_rows

schema = muat_schema("config/schema_pesanan.json")
contoh = json.load(open("data/contoh_pesanan_halaman1.json", encoding="utf-8"))

baris = contoh["data"][0]
baris["total"] = "seratus lima puluh ribu"   # skema lama: angka; skema baru: teks
baris["email"] = "budi#example.com"
baris["status"] = "refunded"

for indeks, galat in validate_rows([baris], schema):
    for pesan in galat:
        print(pesan)
PY

Output yang Dihasilkan:

text 
data[0].email: nilai 'budi#example.com' tidak cocok dengan pattern ^[^@\s]+@[^@\s]+\.[^@\s]+$
data[0].total: type harus number, didapat str
data[0].status: nilai 'refunded' tidak ada di enum ['baru', 'dibayar', 'dikirim', 'selesai', 'dibatalkan']

Setiap pesan menyebut lokasi field dengan tepat, termasuk indeks baris, sehingga tim bisa melaporkan perubahan kontrak ke pemilik API dengan bukti yang presisi. Praktik ini sejalan dengan chapter Validasi Data saat Ingestion dan pembahasan Schema Evolution pada bagian format data.

Kebijakan raise dan skip #

Pipeline memakai dua kebijakan terhadap baris tidak valid. Mode raise menghentikan run dengan ValidationError sebelum satu baris pun tersimpan, cocok ketika data pesanan harus utuh. Mode skip melompati baris bermasalah, mencatat jumlahnya pada ringkasan run, lalu menyimpan sisanya. Keduanya diatur lewat on_invalid di config/pipeline.yaml atau flag --on-invalid, dan penerapannya tampak pada Bagian 7 serta Bagian 9.

Uji Validator #

Sembilan tes di tests/test_validator.py menutup setiap kata kunci schema memakai baris contoh nyata:

python 
def test_tipe_salah(schema_pesanan):
    baris = dict(BARIS[0])
    baris["total"] = "seratus ribu"
    galat = validate(baris, schema_pesanan)
    assert any("total" in pesan and "number" in pesan for pesan in galat)


def test_enum_status(schema_pesanan):
    baris = dict(BARIS[0])
    baris["status"] = "refunded"
    galat = validate(baris, schema_pesanan)
    assert any("enum" in pesan for pesan in galat)

Tes pertama mensimulasikan perubahan tipe total dari angka ke teks, dan tes kedua mensimulasikan nilai status di luar daftar tetap. Tes lain di berkas yang sama menutup jebakan khas Python: True adalah instance dari int, sehingga validator secara eksplisit mengecualikan bool dari tipe integer. Sintaks pola regex yang dipakai keyword pattern terdokumentasi di dokumentasi modul re Python .

bash 
python3 -m pytest tests/test_validator.py -q

Output yang Dihasilkan:

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

Data yang lolos pagar schema kini layak dipercaya dan layak disimpan. Bagian berikutnya membahas tempat penyimpanannya: zona raw berformat JSON Lines yang tidak pernah menimpa data lama.