API tasarımına giriş: HTTP, doğrulama ve ilk servis

Neden önemli?

Kurumsal Python kodunun büyük bölümü bir ağ sınırında durur: Anadolu Market'in sipariş servisi kargo firmasının API'sine istek atar, Ilgaz Bank'ın kredi modülü skor servisinden yanıt bekler. Bu sınırda her şey ters gidebilir: zaman aşımı, 503, bozuk JSON, beklenmeyen alan. Bu derste HTTP'yi, istemci kütüphanelerini ve bir API yazmanın iskeletini öğreneceksin. Tarayıcıda ağ erişimi olmadığı için örnekler sahte istemci (fake client) ile çalışır; gerçek requests/httpx/FastAPI kodu gösterim olarak verilir. Sahte istemci yaklaşımı aynı zamanda ekiplerin birim testte kullandığı yöntemdir, boşa gitmez.

HTTP temelleri

Bir HTTP isteği: yöntem (GET, POST, PUT, PATCH, DELETE), yol (/siparisler/42), başlıklar (Authorization, Content-Type) ve gövde (genelde JSON). Yanıt: durum kodu, başlıklar, gövde. GET ve DELETE idempotent (tekrarlanınca aynı sonuç), POST değildir; bu yüzden POST'u körlemesine yeniden denemek çift sipariş yaratabilir.

requests ve httpx kalıpları

İki kütüphane de benzer arayüz sunar; httpx async ve HTTP/2 destekler. Üç kural: her isteğe zaman aşımı ver (varsayılan sonsuzdur), oturum (Session/Client) kullan (bağlantı tekrar kullanımı, ortak başlık), raise_for_status() ile 4xx/5xx'i istisnaya çevir.

# Gösterim: gerçek projede böyle yazılır (burada çalışmaz)
GOSTERIM = '''
import httpx

with httpx.Client(base_url="https://api.anadolu.example",
                  headers={"Authorization": "Bearer TOKEN"},
                  timeout=httpx.Timeout(5.0, connect=2.0)) as client:
    r = client.get("/siparisler", params={"durum": "bekliyor", "limit": 50})
    r.raise_for_status()
    siparisler = r.json()
    r2 = client.post("/siparisler", json={"musteri": 10, "tutar": 99.9})
    print(r2.status_code, r2.json()["id"])
'''
print(GOSTERIM.strip().splitlines()[0])
print('Satır sayısı:', len(GOSTERIM.strip().splitlines()))

Şimdi aynı sözleşmeye uyan, elimizdeki yanıtları döndüren küçük bir sahte istemci yazalım. Gerçek kodda bu sınıfın yerine httpx.Client geçer; iş mantığı değişmez.

import json

class SahteYanit:
    def __init__(self, status_code, govde='', headers=None):
        self.status_code = status_code
        self.text = govde
        self.headers = headers or {}
    def json(self):
        return json.loads(self.text)
    def raise_for_status(self):
        if self.status_code >= 400:
            raise RuntimeError(f'HTTP {self.status_code}')

class SahteIstemci:
    """Yol -> yanıt tablosu; gerçek httpx.Client ile aynı yüzey."""
    def __init__(self, tablo):
        self.tablo = tablo
        self.istekler = []
    def get(self, yol, params=None, timeout=5.0):
        self.istekler.append(('GET', yol, params))
        return self.tablo.get(yol, SahteYanit(404, '{"detail":"yok"}'))
    def post(self, yol, json=None, timeout=5.0):
        self.istekler.append(('POST', yol, json))
        return self.tablo.get(yol, SahteYanit(404, '{"detail":"yok"}'))

client = SahteIstemci({
    '/siparisler/42': SahteYanit(200, '{"id": 42, "tutar": 149.9, "durum": "kargoda"}'),
    '/siparisler/99': SahteYanit(404, '{"detail": "sipariş yok"}'),
})

def siparis_getir(client, no):
    r = client.get(f'/siparisler/{no}', timeout=5.0)
    if r.status_code == 404:
        return None
    r.raise_for_status()
    return r.json()

print(siparis_getir(client, 42))
print(siparis_getir(client, 99))
print(client.istekler)

FastAPI ile uç nokta

FastAPI, fonksiyon imzasındaki tip ipuçlarından (type hint) hem doğrulamayı hem belgeyi üretir. Gövde modeli pydantic BaseModel'dir: alan tipleri uymazsa otomatik 422 döner.

GOSTERIM = '''
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

app = FastAPI()

class SiparisIstek(BaseModel):
    musteri_id: int
    tutar: float = Field(gt=0)
    sehir: str

DB: dict[int, dict] = {}

@app.post("/siparisler", status_code=201)
def siparis_olustur(s: SiparisIstek):
    no = len(DB) + 1
    DB[no] = s.model_dump() | {"id": no, "durum": "bekliyor"}
    return DB[no]

@app.get("/siparisler/{no}")
def siparis_getir(no: int):
    if no not in DB:
        raise HTTPException(status_code=404, detail="sipariş yok")
    return DB[no]
'''
print('FastAPI uç noktaları:', [s.strip() for s in GOSTERIM.splitlines() if s.startswith('@app')])

Pyodide'de FastAPI yok; ama mantığı fonksiyon tabanlı yönlendirici (router) ile birebir kurabiliriz: yol deseni + yöntem bir fonksiyona eşlenir, gövde bir doğrulayıcıdan geçer, hata durumunda kod ve detay döner.

import json, re
from dataclasses import dataclass, asdict

class HTTPHata(Exception):
    def __init__(self, kod, detay):
        self.kod, self.detay = kod, detay

@dataclass
class SiparisIstek:          # pydantic mantığı, elle
    musteri_id: int
    tutar: float
    sehir: str
    @classmethod
    def dogrula(cls, ham: dict):
        hatalar = []
        if not isinstance(ham.get('musteri_id'), int): hatalar.append('musteri_id int olmalı')
        t = ham.get('tutar')
        if not isinstance(t, (int, float)) or t <= 0: hatalar.append('tutar > 0 olmalı')
        if not isinstance(ham.get('sehir'), str) or not ham['sehir']: hatalar.append('sehir boş olamaz')
        if hatalar:
            raise HTTPHata(422, hatalar)
        return cls(ham['musteri_id'], float(t), ham['sehir'])

class Uygulama:
    def __init__(self):
        self.rotalar = []          # (yöntem, regex, fonksiyon)
    def rota(self, yontem, desen):
        rx = re.compile('^' + re.sub(r'\{(\w+)\}', r'(?P<\1>[^/]+)', desen) + '$')
        def kaydet(fn):
            self.rotalar.append((yontem, rx, fn)); return fn
        return kaydet
    def istek(self, yontem, yol, govde=None):
        for y, rx, fn in self.rotalar:
            m = rx.match(yol)
            if m and y == yontem:
                try:
                    return 200, fn(**m.groupdict(), govde=govde)
                except HTTPHata as e:
                    return e.kod, {'detail': e.detay}
        return 404, {'detail': 'rota yok'}

app = Uygulama()
DB = {}

@app.rota('POST', '/siparisler')
def olustur(govde):
    s = SiparisIstek.dogrula(govde)
    no = len(DB) + 1
    DB[no] = asdict(s) | {'id': no, 'durum': 'bekliyor'}
    return DB[no]

@app.rota('GET', '/siparisler/{no}')
def getir(no, govde=None):
    no = int(no)
    if no not in DB:
        raise HTTPHata(404, 'sipariş yok')
    return DB[no]

print(app.istek('POST', '/siparisler', {'musteri_id': 7, 'tutar': 99.9, 'sehir': 'Bursa'}))
print(app.istek('POST', '/siparisler', {'musteri_id': '7', 'tutar': -1}))
print(app.istek('GET', '/siparisler/1'))
print(app.istek('GET', '/siparisler/5'))

Yeniden deneme ve zaman aşımı

Geçici hatalar (429, 502, 503, bağlantı kopması) için üstel geri çekilme (exponential backoff) kullan: 1 s, 2 s, 4 s... ve küçük rastgele sapma (jitter) ekle ki yüzlerce istemci aynı anda tekrar vurmasın. 4xx hatalarında (429 hariç) yeniden deneme yapma; istek zaten yanlış. POST'ta yeniden deneme yapacaksan Idempotency-Key başlığı gönder.

import random

def geri_cekilme_sureleri(deneme, taban=1.0, tavan=30.0, jitter=0.2):
    """deneme 0,1,2... için bekleme süreleri (saniye)."""
    sure = min(tavan, taban * 2 ** deneme)
    return round(sure * (1 + random.uniform(-jitter, jitter)), 2)

def yeniden_dene(fn, deneme_sayisi=4, uyku=lambda s: None):
    son_hata = None
    for i in range(deneme_sayisi):
        try:
            r = fn()
            if r.status_code in (429, 502, 503):
                raise RuntimeError(f'geçici: {r.status_code}')
            return r
        except RuntimeError as e:
            son_hata = e
            if i < deneme_sayisi - 1:
                uyku(geri_cekilme_sureleri(i))
    raise RuntimeError(f'{deneme_sayisi} denemede başarısız: {son_hata}')

random.seed(1)
yanitlar = iter([SahteYanit(503), SahteYanit(503), SahteYanit(200, '{"ok": true}')])
bekleme = []
r = yeniden_dene(lambda: next(yanitlar), uyku=bekleme.append)
print(r.status_code, r.json(), 'beklemeler:', bekleme)

import io, json
from wsgiref.util import setup_testing_defaults

SIPARISLER = {1042: {"no": 1042, "sehir": "İzmir", "tutar": 1250}}

def json_yanit(start_response, durum, veri, ek=()):
    govde = json.dumps(veri, ensure_ascii=False).encode("utf-8")
    start_response(durum, [("Content-Type", "application/json; charset=utf-8"),
                           ("Content-Length", str(len(govde))), *ek])
    return [govde]

def uygulama(environ, start_response):
    yontem, yol = environ["REQUEST_METHOD"], environ["PATH_INFO"]
    if yontem == "GET" and yol.startswith("/siparisler/"):
        s = SIPARISLER.get(int(yol.rsplit("/", 1)[1]))
        if s is None:
            return json_yanit(start_response, "404 Not Found",
                              {"title": "Sipariş bulunamadı", "status": 404})
        return json_yanit(start_response, "200 OK", s)
    if yontem == "POST" and yol == "/siparisler":
        boy = int(environ.get("CONTENT_LENGTH") or 0)
        veri = json.loads(environ["wsgi.input"].read(boy))
        no = max(SIPARISLER) + 1
        SIPARISLER[no] = {"no": no, **veri}
        return json_yanit(start_response, "201 Created", SIPARISLER[no],
                          [("Location", f"/siparisler/{no}")])
    return json_yanit(start_response, "405 Method Not Allowed", {"status": 405})

def istek_at(yontem, yol, govde=None):          # test istemcisi: ağ yok
    environ = {}
    setup_testing_defaults(environ)
    ham = json.dumps(govde).encode() if govde else b""
    environ.update(REQUEST_METHOD=yontem, PATH_INFO=yol,
                   CONTENT_LENGTH=str(len(ham)), **{"wsgi.input": io.BytesIO(ham)})
    yakalanan = {}
    def start_response(durum, basliklar):
        yakalanan["durum"], yakalanan["basliklar"] = durum, dict(basliklar)
    govde = b"".join(uygulama(environ, start_response))
    return yakalanan["durum"], yakalanan["basliklar"].get("Location"), json.loads(govde)

print(istek_at("GET", "/siparisler/1042"))
print(istek_at("POST", "/siparisler", {"sehir": "Bursa", "tutar": 300}))
print(istek_at("GET", "/siparisler/9"))

import hashlib, hmac, json

GIZLI = b"whsec_anadolu_test"

def imzala(zaman: int, govde: bytes) -> str:
    mesaj = str(zaman).encode() + b"." + govde
    return hmac.new(GIZLI, mesaj, hashlib.sha256).hexdigest()

def dogrula(zaman: int, govde: bytes, imza: str, simdi: int, tolerans=300) -> bool:
    if abs(simdi - zaman) > tolerans:
        return False                                   # eski istek: tekrar saldırısı
    return hmac.compare_digest(imzala(zaman, govde), imza)

govde = json.dumps({"olay": "odeme.tamamlandi", "tutar": 1250}).encode()
imza = imzala(1_700_000_000, govde)
print(dogrula(1_700_000_000, govde, imza, simdi=1_700_000_060))
print(dogrula(1_700_000_000, govde.replace(b"1250", b"9999"), imza, simdi=1_700_000_060))
print(dogrula(1_700_000_000, govde, imza, simdi=1_700_009_999))