Bitirme projesi: Anadolu Market Sipariş API'si

Proje: Anadolu Market Sipariş API'si

Bu bitirme projesinde, ileri seviyede öğrendiklerini tek bir üretim kalitesinde serviste birleştireceksin: müşterinin sepetini siparişe çeviren, stoğu düşen, kuponu uygulayan, sipariş durumunu yöneten ve bunu güvenli bir HTTP API'si olarak sunan Anadolu Market Sipariş API'si. İşe alım mülakatlarında "bana bir sipariş servisi tasarla" sorusunun cevabı tam olarak bu projedir.

Tarayıcıda FastAPI çalışmaz; ama projenin asıl değeri web çatısında değil, katmanlarda. Şema, depo ve servis katmanlarını gerçek pydantic ve SQLAlchemy ile (bellek içi SQLite) kuracak, uç noktaları küçük bir yönlendiriciyle taklit edeceksin. Her adımın FastAPI karşılığını da göreceksin; kendi bilgisayarında birkaç satırla gerçek API'ye dönüşür.

Mimari: dört katman

Katmanlı mimari (layered architecture), her katmanın tek bir sorumluluğu olduğu ve yalnızca bir alttaki katmanı tanıdığı bir düzendir. İstek yukarıdan aşağıya akar, cevap aşağıdan yukarıya döner:

  HTTP isteği  ──►  [ Uç nokta (endpoint) ]   yol, kimlik, şema doğrulama, HTTP kodu
                          │
                          ▼
                    [ Servis (service) ]       iş kuralları: stok, indirim, durum geçişi
                          │
                          ▼
                    [ Depo (repository) ]      veritabanı erişimi: ekle, getir, sorgula
                          │
                          ▼
                    [ Model / ORM ]            tablolar: urunler, siparisler, kalemler

Kazanç: servis katmanını HTTP olmadan test edersin, veritabanını değiştirmek yalnızca depoyu etkiler, aynı servisi yarın bir CLI ya da kuyruk tüketicisi de kullanabilir.

anadolu-siparis/
├── market/
│   ├── modeller.py      # SQLAlchemy ORM
│   ├── semalar.py       # pydantic
│   ├── depolar.py       # repository
│   ├── servisler.py     # iş kuralları
│   ├── hatalar.py       # alan hataları
│   ├── guvenlik.py      # hmac token
│   └── api.py           # FastAPI uç noktaları
├── tests/
├── Dockerfile
├── pyproject.toml
└── README.md

Adım 1: Şemalar (pydantic)

Dış dünyadan gelen her veri güvenilmezdir. pydantic şeması türü, aralığı ve fazladan alanları sınırda denetler; servis katmanına yalnızca temiz veri ulaşır. Pydantic dersinde (Veri doğrulama ve pydantic v2) temelini gördün, burada API sözleşmesini kuruyoruz:

from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator

class KalemGirdi(BaseModel):
    model_config = ConfigDict(extra="forbid")
    urun_id: int = Field(gt=0)
    adet: int = Field(ge=1, le=100)

class SiparisOlustur(BaseModel):
    model_config = ConfigDict(extra="forbid")
    kalemler: list[KalemGirdi] = Field(min_length=1)
    kupon: str | None = None

    @field_validator("kupon")
    @classmethod
    def kuponu_duzenle(cls, v):
        v = (v or "").strip().upper()
        return v or None

g = SiparisOlustur(kalemler=[{"urun_id": 1, "adet": 2}], kupon=" hosgeldin10 ")
print(g.kupon, g.model_dump())
try:
    SiparisOlustur(kalemler=[{"urun_id": 1, "adet": 0}], indirim=99)
except ValidationError as e:
    for h in e.errors():
        print(h["loc"], h["type"])

Adım 2: ORM modelleri ve depo

Fiyatları kuruş cinsinden tamsayı tutuyoruz: kayan noktalı sayı (0.1 + 0.2) para için uygun değildir, SQLite ise Decimal'ı doğal desteklemez. Depo (repository) sınıfı SQLAlchemy oturumunu (Session) sarmalar; servis SQL görmez. SQLAlchemy dersinde ORM'i gördün; burada onu bir desenin parçası yapıyoruz.

from sqlalchemy import ForeignKey, String, create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, relationship

class Base(DeclarativeBase):
    pass

class Urun(Base):
    __tablename__ = "urunler"
    id: Mapped[int] = mapped_column(primary_key=True)
    ad: Mapped[str] = mapped_column(String(80), unique=True)
    fiyat_kurus: Mapped[int]
    stok: Mapped[int]

class Siparis(Base):
    __tablename__ = "siparisler"
    id: Mapped[int] = mapped_column(primary_key=True)
    musteri: Mapped[str]
    durum: Mapped[str] = mapped_column(default="yeni")
    toplam_kurus: Mapped[int] = mapped_column(default=0)
    kalemler: Mapped[list["Kalem"]] = relationship(back_populates="siparis")

class Kalem(Base):
    __tablename__ = "kalemler"
    id: Mapped[int] = mapped_column(primary_key=True)
    siparis_id: Mapped[int] = mapped_column(ForeignKey("siparisler.id"))
    urun_id: Mapped[int] = mapped_column(ForeignKey("urunler.id"))
    adet: Mapped[int]
    birim_kurus: Mapped[int]
    siparis: Mapped[Siparis] = relationship(back_populates="kalemler")

class UrunDepo:
    def __init__(self, oturum):
        self.oturum = oturum
    def ekle(self, ad, fiyat_kurus, stok):
        u = Urun(ad=ad, fiyat_kurus=fiyat_kurus, stok=stok)
        self.oturum.add(u)
        self.oturum.flush()          # id'yi almak için
        return u
    def getir(self, urun_id):
        return self.oturum.get(Urun, urun_id)
    def stokta_olanlar(self):
        return self.oturum.scalars(select(Urun).where(Urun.stok > 0).order_by(Urun.ad)).all()

motor = create_engine("sqlite://")      # bellek içi
Base.metadata.create_all(motor)
with Session(motor) as oturum:
    depo = UrunDepo(oturum)
    depo.ekle("Çay 1 kg", 18950, 40)
    depo.ekle("Fındık 500 g", 24900, 0)
    depo.ekle("Bal 850 g", 42500, 12)
    oturum.commit()
    print([u.ad for u in depo.stokta_olanlar()], depo.getir(1).fiyat_kurus)

Adım 3: Servis katmanı ve iş kuralları

Projenin kalbi burası. Kurallar:

• Stok: siparişteki bütün kalemlerin stoğu önce kontrol edilir, ancak hepsi yeterliyse düşülür (ya hep ya hiç).
• İndirim: HOSGELDIN10 ara toplamdan %10 (aşağı yuvarlanmış kuruş), SEPET50 5.000 kuruş (50 TL) indirir ama ara toplamı geçemez; bilinmeyen kupon hatadır.
• Kargo: indirim sonrası tutar 75.000 kuruş (750 TL) ve üzeriyse ücretsiz, değilse 3.990 kuruş.
• Durum geçişleri: sipariş bir durum makinesidir (state machine); yalnızca izinli geçişler yapılabilir.

KUPONLAR = {"HOSGELDIN10": ("yuzde", 10), "SEPET50": ("tutar", 5000)}
KARGO_ESIK, KARGO_UCRET = 75000, 3990
GECISLER = {"yeni": {"odendi", "iptal"}, "odendi": {"kargoda", "iptal"},
            "kargoda": {"teslim"}, "teslim": set(), "iptal": set()}

class GecersizKupon(Exception): pass
class GecersizGecis(Exception): pass

def fiyat_hesapla(kalemler, kupon=None):
    ara = sum(fiyat * adet for fiyat, adet in kalemler)
    indirim = 0
    if kupon:
        if kupon not in KUPONLAR:
            raise GecersizKupon(kupon)
        tur, deger = KUPONLAR[kupon]
        indirim = ara * deger // 100 if tur == "yuzde" else min(deger, ara)
    net = ara - indirim
    kargo = 0 if net >= KARGO_ESIK else KARGO_UCRET
    return {"ara_toplam": ara, "indirim": indirim, "kargo": kargo, "toplam": net + kargo}

def gecis_kontrol(eski, yeni):
    if yeni not in GECISLER.get(eski, set()):
        raise GecersizGecis(f"{eski} -> {yeni}")
    return yeni

print(fiyat_hesapla([(18950, 2), (42500, 1)], "HOSGELDIN10"))
print(fiyat_hesapla([(42500, 2)]))
print(gecis_kontrol("yeni", "odendi"))
try:
    gecis_kontrol("teslim", "iptal")
except GecersizGecis as e:
    print("Reddedildi:", e)

Adım 4: Uç noktalar ve yönlendirme

Bir web çatısının çekirdeği yönlendiricidir (router): metot + yol desenini işleyici fonksiyona eşler ve yoldaki parametreleri (/siparisler/42 → sid=42) çıkarır. Önce gerçek FastAPI kodu:

from fastapi import Depends, FastAPI, Header, HTTPException

app = FastAPI(title="Anadolu Market Sipariş API")

def aktif_kullanici(authorization: str = Header()) -> str:
    return token_dogrula(authorization.removeprefix("Bearer "), GIZLI, time.time())

@app.post("/siparisler", status_code=201)
def siparis_olustur(girdi: SiparisOlustur, kullanici: str = Depends(aktif_kullanici),
                    servis: SiparisServisi = Depends(servis_al)):
    return servis.olustur(kullanici, girdi)

@app.get("/siparisler/{sid}")
def siparis_getir(sid: int, kullanici: str = Depends(aktif_kullanici),
                  servis: SiparisServisi = Depends(servis_al)):
    return servis.getir(kullanici, sid)

Kendi bilgisayarında dene: pip install fastapi uvicorn sqlalchemy, sonra uvicorn market.api:app --reload ve tarayıcıda http://127.0.0.1:8000/docs. FastAPI şemadan otomatik belge üretir. Aynı mantığı tarayıcıda küçük bir yönlendiriciyle kuralım:

class Yonlendirici:
    def __init__(self):
        self.rotalar = []

    def rota(self, metot, desen):
        def kaydet(f):
            self.rotalar.append((metot, desen.strip("/").split("/"), f))
            return f
        return kaydet

    def coz(self, metot, yol):
        parcalar = yol.strip("/").split("/")
        for m, desen, f in self.rotalar:
            if m != metot or len(desen) != len(parcalar):
                continue
            params = {}
            for d, p in zip(desen, parcalar):
                if d.startswith("{"):
                    ad, _, tur = d[1:-1].partition(":")
                    if tur == "int" and not p.isdigit():
                        break
                    params[ad] = int(p) if tur == "int" else p
                elif d != p:
                    break
            else:
                return f, params
        return None, {}

api = Yonlendirici()

@api.rota("GET", "/siparisler/{sid:int}")
def siparis_getir(sid):
    return 200, {"id": sid, "durum": "kargoda"}

f, params = api.coz("GET", "/siparisler/42")
print(params, f(**params))
print(api.coz("GET", "/siparisler/abc")[0])

Adım 5: Hata → HTTP durum kodu

Her alan hatası tek bir yerde HTTP koduna eşlenir. FastAPI'de bu @app.exception_handler(IsKuraliHatasi) ile yapılır. Eşleme sınıf hiyerarşisini izler: StokYetersiz, IsKuraliHatasi'nin alt sınıfıysa kendi kaydı yoksa üst sınıfın kodunu alır.

Adım 6: Kimlik doğrulama: hmac imzalı token

Gerçek projelerde JWT ve OAuth2 kullanılır (Web güvenliği dersinde gördün); hepsinin çekirdeği imzadır. Sunucu kullanici.son_kullanma metnini gizli anahtarla HMAC-SHA256 ile imzalar; token değiştirilirse imza tutmaz. Karşılaştırmayı hmac.compare_digest ile yaparsın; == ilk farklı karakterde durduğu için zamanlama saldırısına (timing attack) açıktır.

import hashlib, hmac

GIZLI = b"gelistirme-anahtari"     # üretimde ortam değişkeninden

class YetkiHatasi(Exception): pass

def imza(metin, gizli):
    return hmac.new(gizli, metin.encode(), hashlib.sha256).hexdigest()

def token_uret(kullanici, son, gizli=GIZLI):
    govde = f"{kullanici}.{son}"
    return f"{govde}.{imza(govde, gizli)}"

def token_dogrula(token, gizli, simdi):
    try:
        kullanici, son, gelen = token.rsplit(".", 2)
        son = int(son)
    except ValueError:
        raise YetkiHatasi("bozuk token")
    if not hmac.compare_digest(gelen, imza(f"{kullanici}.{son}", gizli)):
        raise YetkiHatasi("imza geçersiz")
    if simdi >= son:
        raise YetkiHatasi("süresi dolmuş")
    return kullanici

t = token_uret("ayse", 1_800_000_000)
print(t[:30] + "...")
print(token_dogrula(t, GIZLI, 1_700_000_000))
for sahte in [t.replace("ayse", "ali_"), "rastgele"]:
    try:
        token_dogrula(sahte, GIZLI, 1_700_000_000)
    except YetkiHatasi as e:
        print("Reddedildi:", e)

Adım 7: Testler

Katmanlar sayesinde her düzeyi ayrı test edersin: fiyat ve geçiş kuralları için saf birim testleri, depo için bellek içi SQLite ile entegrasyon testleri, servis için depo yerine mock, uç noktalar için istek sözlüğüyle uçtan uca testler.

import unittest
from unittest.mock import Mock

class StokYetersiz(Exception): pass

class SiparisServisi:
    def __init__(self, urun_depo):
        self.urun_depo = urun_depo
    def stok_dus(self, kalemler):
        urunler = {uid: self.urun_depo.getir(uid) for uid, _ in kalemler}
        eksik = [uid for uid, adet in kalemler if urunler[uid]["stok"] < adet]
        if eksik:
            raise StokYetersiz(eksik)
        for uid, adet in kalemler:
            self.urun_depo.stok_guncelle(uid, urunler[uid]["stok"] - adet)

class ServisTest(unittest.TestCase):
    def setUp(self):
        self.depo = Mock()
        self.depo.getir.side_effect = lambda uid: {1: {"stok": 5}, 2: {"stok": 1}}[uid]

    def test_hepsi_yeterli(self):
        SiparisServisi(self.depo).stok_dus([(1, 2), (2, 1)])
        self.assertEqual(self.depo.stok_guncelle.call_count, 2)

    def test_biri_yetersizse_hicbiri_dusmez(self):
        with self.assertRaises(StokYetersiz):
            SiparisServisi(self.depo).stok_dus([(1, 2), (2, 3)])
        self.depo.stok_guncelle.assert_not_called()

sonuc = unittest.TextTestRunner(verbosity=0).run(unittest.TestLoader().loadTestsFromTestCase(ServisTest))
print("Başarılı:", sonuc.wasSuccessful(), "| test:", sonuc.testsRun)

Adım 8: Dockerfile ve README

Docker dersindeki kurallarla imajı hazırla ve projeyi bir README ile teslim et:

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY market/ market/
RUN useradd --create-home app
USER app
EXPOSE 8000
CMD ["uvicorn", "market.api:app", "--host", "0.0.0.0", "--port", "8000"]

# Anadolu Market Sipariş API

## Çalıştırma
    docker compose up --build        # API + PostgreSQL
    open http://localhost:8000/docs

## Ortam değişkenleri
| Ad            | Zorunlu | Açıklama                    |
| DATABASE_URL  | evet    | postgresql://...            |
| TOKEN_GIZLI   | evet    | HMAC imza anahtarı          |

## Uç noktalar
POST  /siparisler                 201  sipariş oluştur
GET   /siparisler/{id}            200  sipariş getir
PATCH /siparisler/{id}/durum      200  durum değiştir

## Test
    python -m pytest -q

# compose.yaml
services:
  db:
    image: postgres:16
    environment: {POSTGRES_USER: market, POSTGRES_PASSWORD: ${DB_SIFRE}, POSTGRES_DB: market}
    volumes: [pgdata:/var/lib/postgresql/data]
    healthcheck: {test: ["CMD-SHELL", "pg_isready -U market"], interval: 5s, retries: 10}
  redis:
    image: redis:7-alpine
    healthcheck: {test: ["CMD", "redis-cli", "ping"], interval: 5s}
  migrate:
    build: .
    command: alembic upgrade head
    environment: &ortam
      DATABASE_URL: postgresql+psycopg://market:${DB_SIFRE}@db/market
      REDIS_URL: redis://redis:6379/0
    depends_on: {db: {condition: service_healthy}}
  api:
    build: .
    command: uvicorn market.app:app --host 0.0.0.0 --port 8000
    environment: *ortam
    ports: ["8000:8000"]
    depends_on:
      migrate: {condition: service_completed_successfully}
      redis: {condition: service_healthy}
  worker:
    build: .
    command: python -m market.worker
    environment: *ortam
    depends_on:
      migrate: {condition: service_completed_successfully}
      redis: {condition: service_healthy}
volumes:
  pgdata:

async def urunleri_getir(kategori: str) -> list[dict]:
    anahtar = f"urunler:{kategori}:v1"
    if (hazir := await redis.get(anahtar)) is not None:
        return json.loads(hazir)
    urunler = await depo.kategoriye_gore(kategori)
    await redis.set(anahtar, json.dumps(urunler), ex=300)     # 5 dakika
    return urunler

async def urun_guncelle(urun_id, alanlar):
    urun = await depo.guncelle(urun_id, alanlar)
    await redis.delete(f"urunler:{urun.kategori}:v1")         # geçersiz kıl
    return urun

# market/worker.py  (RQ ile; Celery ya da arq da olur)
from redis import Redis
from rq import Worker, Queue
Worker([Queue("eposta", connection=Redis.from_url(REDIS_URL))]).work()

# uç noktada
kuyruk.enqueue("market.isler.onay_epostasi", siparis.id, retry=Retry(max=3, interval=[10, 60, 300]))

import time

class JetonKovasi:
    def __init__(self, hiz, kapasite, saat=time.monotonic):
        self.hiz, self.kapasite, self.saat = hiz, kapasite, saat
        self.durum = {}                            # anahtar -> (jeton, son_zaman)

    def izin(self, anahtar):
        simdi = self.saat()
        jeton, son = self.durum.get(anahtar, (self.kapasite, simdi))
        jeton = min(self.kapasite, jeton + (simdi - son) * self.hiz)
        if jeton >= 1:
            self.durum[anahtar] = (jeton - 1, simdi)
            return True, 0.0
        self.durum[anahtar] = (jeton, simdi)
        return False, round((1 - jeton) / self.hiz, 2)    # Retry-After (sn)

zaman = [0.0]
kova = JetonKovasi(hiz=1, kapasite=3, saat=lambda: zaman[0])
for t in [0, 0, 0, 0, 0.5, 1.2, 5]:
    zaman[0] = t
    print(t, kova.izin("ip:10.0.0.7"))