Veri sözleşmesi ve şema doğrulama: bozuk kaydı boru hattının kapısında durdurmak

Dışarıdan gelen bir CSV ya da JSON dosyasını doğrudan veritabanına yüklemek, kapısı olmayan bir depoya mal kabul etmeye benzer. Veri sözleşmesi ve şema doğrulama tam bu kapıyı kurar: dosyayı işlemeden önce her kaydın hangi alanları, hangi tiplerde ve hangi aralıkta taşıması gerektiğini yazılı bir kurala bağlarsınız; kurala uymayan satır içeri girmez. Bu yazıda o kapıyı Pydantic ile nasıl kurduğumu, reddedilen satırları neden sessizce düşürmek yerine karantinaya aldığımı ve sözleşmeyi Python bilmeyen bir karşı tarafla JSON Schema üzerinden nasıl paylaştığımı anlatacağım.

Bu konuya ihtiyacım, boru hatlarında en pahalı hatanın hemen patlayan değil, aylar sonra fark edilen hata olduğunu görerek doğdu. Tutar sütununda virgüllü sayı gelen bir dosya, kimsenin bakmadığı bir try/except içinde "atlandı"; raporlar o ay eksik toplamla çıktı ve kimse neden diye sormadı. Sorun kodun çökmesi değildi; çökmemesiydi.

Aşağıdaki yaklaşım küçük ekipler için tasarlandı: tek bir model dosyası, tek bir yükleyici, tek bir karantina klasörü. Büyük çerçevelere ihtiyaç duymadan, SQLite ile kurduğum küçük boru hatlarının önüne de aynen takılabilir.

Sözleşme nedir, dosya biçiminden farkı ne?

"Dosya CSV, sütunlar belli" demek sözleşme değildir. Sözleşme, her alan için dört soruya cevap verir:

• Zorunluluk: Alan boş gelebilir mi? Boş gelirse bu bir hata mı, yoksa "bilinmiyor" anlamına mı gelir?
• Tip: Tarih ISO biçiminde mi, gün/ay/yıl mı? Tutarın ondalık ayracı nokta mı, virgül mü?
• Aralık ve küme: Adet sıfırdan büyük olmalı; para birimi yalnızca belirli kodlardan biri olabilir.
• Fazlalık: Beklenmeyen bir sütun gelirse görmezden mi gelinir, hata mı sayılır?

Bu cevaplar bir belgede değil, çalıştırılabilir kodda durmalı. Belge bayatlar; kod her dosyada yeniden sınanır. Sözleşmeyi kodda tutmanın ikinci faydası, ihlalin nerede ve neden olduğunu makinenin söyleyebilmesidir. "Dosya bozuk" ile "42. satır, tutar alanı, sıfırdan küçük" arasındaki fark, sorunu gönderen tarafa geri iletebilmek demektir.

Pydantic ile doğrulayıcı model

Pydantic, tip ipuçlarını doğrulama kuralına çeviren bir Python kütüphanesi. Aşağıdaki model bir sipariş dosyasının tek satırını tarif ediyor. Örnekler Pydantic 2 ve Python 3.10 üzeri içindir.

from datetime import date
from decimal import Decimal
from typing import Literal

from pydantic import BaseModel, ConfigDict, Field, field_validator


class SiparisKaydi(BaseModel):
    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)

    siparis_no: str = Field(min_length=1, max_length=32,
                            description="Kaynak sistemdeki sipariş numarası")
    tarih: date
    tutar: Decimal = Field(gt=0, max_digits=12, decimal_places=2)
    para_birimi: Literal["TRY", "USD", "EUR"]
    adet: int = Field(ge=1)
    kanal: str | None = None

    @field_validator("kanal", mode="before")
    @classmethod
    def bos_metni_none_yap(cls, deger):
        if isinstance(deger, str) and deger.strip() == "":
            return None
        return deger

Modelde dikkat ettiğim noktalar:

• extra="forbid": başlıkta tanımadığım bir sütun varsa satır reddedilir. Karşı tarafın sütun adını değiştirdiğini ilk bu kural haber verir; aksi hâlde yeni ad sessizce yok sayılır.
• Decimal ile tutar: float kuruş hesaplarında sapma üretir; max_digits ve decimal_places beklenmeyen hassasiyeti de yakalar.
• Literal ile para birimi: küme dışı bir kod, örneğin "TL", hata verir. Bu kasıtlı; dönüştürmeyi yükleyicide değil, kaynağın düzeltmesinde beklerim.
• mode="before" doğrulayıcısı: csv.DictReader boş hücreyi boş metin olarak verir; isteğe bağlı alan için bunu None yapmazsam Pydantic boş metni geçerli bir kanal adı sayar.

CSV'den okunan her değer metindir; Pydantic'in varsayılan esnek modu "2026-09-01" metnini tarihe, "3" metnini tam sayıya çevirir. JSON kaynağında bu dönüşüme daha az iş düşer ama kurallar aynı kalır.

Reddedilen satırı karantinaya alan yükleyici

Doğrulamanın asıl değeri, reddettiğinde ne yaptığında ortaya çıkar. Üç seçenek var: dosyanın tamamını reddetmek, bozuk satırı atlamak ya da bozuk satırı sebebiyle birlikte ayrı bir dosyaya yazmak. İlki tek bir hatalı satır yüzünden bütün günü kaybettirir; ikincisi girişte anlattığım sessiz kayıptır. Üçüncüsü, karantina, hem akışı sürdürür hem de kanıtı saklar.

import csv
import json
from pathlib import Path

from pydantic import ValidationError

from sozlesme import SiparisKaydi


def yukle(kaynak: Path, temiz: Path, karantina: Path) -> tuple[int, int]:
    kabul = red = 0
    alanlar = list(SiparisKaydi.model_fields)
    with (
        kaynak.open(encoding="utf-8", newline="") as f_in,
        temiz.open("w", encoding="utf-8", newline="") as f_ok,
        karantina.open("a", encoding="utf-8") as f_red,
    ):
        yazici = csv.DictWriter(f_ok, fieldnames=alanlar)
        yazici.writeheader()
        # Başlık 1. satır; ilk veri satırı 2'den başlar
        for satir_no, ham in enumerate(csv.DictReader(f_in), start=2):
            try:
                kayit = SiparisKaydi.model_validate(ham)
            except ValidationError as hata:
                red += 1
                f_red.write(json.dumps({
                    "dosya": kaynak.name,
                    "satir": satir_no,
                    "ham": ham,
                    "sebepler": [
                        {
                            "alan": ".".join(str(p) for p in h["loc"]),
                            "tur": h["type"],
                            "mesaj": h["msg"],
                        }
                        for h in hata.errors()
                    ],
                }, ensure_ascii=False) + "\n")
                continue
            kabul += 1
            yazici.writerow(kayit.model_dump(mode="json"))
    return kabul, red


if __name__ == "__main__":
    ok, bad = yukle(
        Path("gelen/siparis-2026-09-04.csv"),
        Path("temiz/siparis-2026-09-04.csv"),
        Path("karantina/siparis-2026-09-04.jsonl"),
    )
    print(f"kabul={ok} red={bad}")

Karantina dosyası satır başına bir JSON nesnesi tutar (JSON Lines). Her nesnede kaynak dosyanın adı, satır numarası, ham içerik ve Pydantic'in ürettiği hata listesi bulunur. Bu biçimi seçmemin sebepleri:

• Ham satır olduğu gibi saklandığı için düzeltme sonrası aynı satır yeniden yüklenebilir; hiçbir şey kaybolmaz.
• alan ve tur bilgisi, "hangi sütun en çok reddediliyor" sorusuna tek bir jq ya da pandas satırıyla cevap verir. Kaynağa geri bildirim buradan çıkar.
• Satır numarası, karşı tarafın kendi dosyasını açıp sorunu görmesini sağlar; onlara "dosyanız bozuk" değil, "84. satır" dersiniz.

Bu yükleyiciyi altı satırlık bir deneme dosyasıyla çalıştırdığımda iki satır geçti, dört satır karantinaya düştü. En öğretici olanı, tutarı 12,50 yazılmış satırdı: virgül ayraç sayıldığı için bütün sütunlar bir sağa kaydı ve hata tutarda değil, para biriminde ve adette göründü. Karantina kaydındaki ham içerik olmasa bu kaymayı hata mesajından anlamak mümkün olmazdı.

Bir kural daha eklerim: ret oranı belirlediğim eşiği aşarsa yükleme tamamen durur ve temiz dosya kullanılmaz. Birkaç satırın reddi normaldir; satırların yarısının reddi büyük ihtimalle başlık kaymış ya da ayraç değişmiş demektir ve o dosyanın yarısını içeri almak, hiç almamaktan kötüdür. Aynı mantığı Excel tabanlı kalite kontrol otomasyonunda da uygularım: kontrolü geçmeyen kayıt görünür bir yerde durur, raporun içinde kaybolmaz.

JSON Schema ile dile bağımsız sözleşme paylaşımı

Sözleşmenin Python dosyasında yaşaması, karşı tarafın da Python bildiğini varsayar. Çoğu zaman bilmez: dosyayı üreten sistem bir ERP, bir Node servisi ya da bir Excel makrosudur. Pydantic modelinden JSON Schema üretmek bu sorunu çözer; şema, hemen her dilde doğrulayıcı kütüphanesi bulunan ortak bir dildir.

import json
from pathlib import Path

from sozlesme import SiparisKaydi

sema = SiparisKaydi.model_json_schema()
hedef = Path("sozlesme/siparis-kaydi.v1.json")
hedef.write_text(json.dumps(sema, ensure_ascii=False, indent=2), encoding="utf-8")

Üretilen dosyayı sürüm numarasıyla depoya koyar ve kaynağı sağlayan tarafa veririm. Onlar dosyayı göndermeden önce kendi taraflarında aynı şemayla doğrulayabilir; hata benim kapımda değil, onların masasında yakalanır. Field(description=...) ile yazdığım açıklamalar şemaya aynen geçtiği için ayrı bir belge tutmam gerekmez. extra="forbid" kuralı da şemada additionalProperties: false olarak görünür; yani fazlalık kuralı karşı tarafa aynen taşınır.

• Şema dosyasının adı sürüm taşır: siparis-kaydi.v1.json. Sürüm, sözleşmenin kimliğidir.
• Pydantic'in ürettiği şema JSON Schema 2020-12 taslağını hedefler; karşı tarafın doğrulayıcısı bu taslağı desteklemeli.
• Şemayı üreten komut sürekli entegrasyonda çalışır ve depodaki dosyayla karşılaştırılır; model değişip şema güncellenmemişse derleme kırılır.

Şema evrimi: yeni alan gelince eski dosya ne olacak?

Sözleşmeler değişir. Soru, değişikliğin geçmişteki dosyaları ve hâlâ eski biçimde gönderen kaynakları bozup bozmayacağıdır. Kuralım basit: eklemek serbest, kaldırmak ve yeniden adlandırmak sürüm atlatır.

class SiparisKaydiV2(SiparisKaydi):
    # İsteğe bağlı ve varsayılanlı: V1 dosyaları doğrulamadan geçmeye devam eder
    musteri_segmenti: Literal["bireysel", "kurumsal"] | None = None

• Yeni alan isteğe bağlı ve varsayılan değerli eklenirse eski dosyalar doğrulamadan geçmeye devam eder; V2 modeli V1 dosyasını okuyabilir.
• Yeni alan zorunluysa eski dosyaların tamamı reddedilir. Bu bir hata değil, sinyaldir: sözleşmenin yeni sürümü vardır ve kaynaklar geçiş yapana kadar eski model devrede kalmalıdır.
• Bir alanın tipi ya da anlamı değişiyorsa (tutar KDV dâhil oldu, tarih saat dilimi kazandı) yeni bir alan adı kullanırım; eski adı bırakıp anlamını değiştirmek sözleşmenin en sinsi ihlalidir.
• extra="forbid" evrimde ters teper gibi görünür: kaynak yeni sütunu benden önce eklerse satırlar reddedilir. Bunu bilerek kabul ederim; karantina bana "bilmediğim bir sütun geldi" der, ben de modeli güncellerim. Sessizce yok saymak, sütunun varlığını hiç öğrenmemek demektir.
• Hangi modelin devreye gireceğini dosya başlığındaki sütun kümesinden ya da dosya adındaki sürüm ekinden seçerim; iki model bir süre yan yana yaşayabilir.

Özetle: sözleşme kodda yaşar, ihlal karantinaya gider, şema dile bağımsız paylaşılır ve her değişiklik sürüm alır. Bu dört adım, boru hattının çalışıyor görünürken yanlış sayı üretmesini engeller; hata olduğunda ise nerede olduğunu gösterir.

Kendi veri akışınızda dışarıdan gelen dosyaların kapısı yoksa ya da var ama reddettiği kayıtların nereye gittiğini bilmiyorsanız, danışmanlık çalışmalarımda tam bu tür yükleyicileri kuruyorum. Mevcut akışınızı birlikte gözden geçirmek için iletişim bölümünden yazabilirsiniz.