FastAPI ile API geliştirme

Neden önemli?

Flask dersinde bir web çatısının nasıl çalıştığını gördün: route kaydı, istek, yanıt, durum kodu. Ama JSON API yazarken Flask'ta işin büyük kısmı tekrar eden "bürokrasidir": gelen gövdeyi oku, her alanı elle doğrula, hatalı alanları toplayıp 400 döndür, dokümantasyonu ayrıca yaz ve kodla güncel tut. FastAPI bu bürokrasiyi tip ipuçlarından (type hints) otomatik üretir: fonksiyonunun parametrelerine yazdığın tiplere bakarak isteği ayrıştırır, pydantic ile doğrular, hatalıysa standart bir 422 yanıtı döner ve tüm API'nin etkileşimli dokümantasyonunu (/docs) kendisi oluşturur.

Bugün bankacılık, e-ticaret ve makine öğrenmesi ekiplerinde yeni Python API'lerinin büyük bölümü FastAPI ile yazılıyor. API ve web dersinde istemci tarafını ve kısa bir FastAPI örneğini, Metaprogramlama dersinde inspect.signature'ı, async dersinde olay döngüsünü gördün. Bu ders hepsini birleştiriyor.

Kurulum ve ilk API

# main.py
from fastapi import FastAPI

app = FastAPI(title="Anadolu Market API", version="1.0.0")

URUNLER = {1: {"ad": "Çay", "fiyat": 85.5, "kategori": "İçecek"},
           2: {"ad": "Kahve", "fiyat": 240.0, "kategori": "İçecek"}}

@app.get("/saglik")
def saglik():
    return {"durum": "ok"}

@app.get("/urunler/{urun_id}")
def urun_getir(urun_id: int, detay: bool = False):
    urun = URUNLER[urun_id]
    return urun if detay else {"ad": urun["ad"]}

Tip ipuçlarının rolü: parametre nereden gelir?

FastAPI parametrenin kaynağını adından ve tipinden çıkarır. Bu kuralları bilmek, "neden 422 alıyorum?" sorusunun yarısını çözer.

from typing import Annotated
from fastapi import FastAPI, Query, Path

app = FastAPI()

@app.get("/urunler")
def urun_ara(
    q: Annotated[str | None, Query(min_length=2, max_length=50)] = None,
    sayfa: Annotated[int, Query(ge=1)] = 1,
    boyut: Annotated[int, Query(ge=1, le=100)] = 20,
    etiket: Annotated[list[str], Query()] = [],       # ?etiket=yeni&etiket=indirim
):
    return {"q": q, "sayfa": sayfa, "boyut": boyut, "etiket": etiket}

@app.get("/subeler/{kod}")
def sube(kod: Annotated[str, Path(pattern=r"^[A-Z]{3}-\d{2}$")]):
    return {"kod": kod}

Annotated[tip, Query(...)] hem tipi hem de ek kuralları (en az/en çok, desen) bildirir. Bu kurallar hem doğrulamada hem de /docs sayfasında kullanılır: dokümantasyon kodla asla çelişmez, çünkü ikisi aynı kaynaktan üretilir.

pydantic modelleri: istek ve yanıt şeması

JSON gövdeyi bir pydantic modeliyle tanımlarsın. FastAPI gövdeyi okur, modele doğrulatır ve fonksiyonuna hazır bir nesne verir. Doğrulama başarısızsa fonksiyonun hiç çağrılmaz; istemci tüm hataların listesini 422 ile alır. Aynı mekanizma burada çalışıyor:

from pydantic import BaseModel, Field, ValidationError, field_validator

class SiparisKalemi(BaseModel):
    urun_id: int = Field(gt=0)
    adet: int = Field(ge=1, le=50)

class SiparisOlustur(BaseModel):
    musteri_id: int = Field(gt=0)
    sehir: str = Field(min_length=2)
    kalemler: list[SiparisKalemi] = Field(min_length=1)
    kupon: str | None = None

    @field_validator("sehir")
    @classmethod
    def sehri_duzenle(cls, v: str) -> str:
        return v.strip().title()

govde = {"musteri_id": "42", "sehir": "  izmir ", "kalemler": [{"urun_id": 1, "adet": 2}]}
s = SiparisOlustur.model_validate(govde)     # FastAPI bunu senin yerine yapar
print(s.musteri_id, type(s.musteri_id).__name__, s.sehir, s.kalemler[0].adet, s.kupon)

hatali = {"musteri_id": 0, "kalemler": [{"urun_id": 1, "adet": 99}]}
try:
    SiparisOlustur.model_validate(hatali)
except ValidationError as e:
    # FastAPI'nin 422 gövdesi: {"detail": [...]} ve konum başına "body" eklenir
    for h in e.errors():
        print(["body", *h["loc"]], h["type"])

Dikkat: "42" metni int'e çevrildi. pydantic varsayılan olarak "makul" dönüşümlere izin verir (lax mode); JSON'da sayıların bazen metin gelmesi yaygındır. Kesin tip istiyorsan Field(strict=True) ya da model_config = ConfigDict(strict=True) kullanırsın. Pydantic ile doğrulama dersinde gördüğün her şey burada geçerli.

Yanıt modeli: neyi döndürdüğünü kontrol et

Veritabanındaki kullanıcı kaydında şifre özeti (hash) vardır; bunu API yanıtında asla göstermemelisin. response_model yanıtı belirtilen şemaya göre filtreler ve doğrular:

from fastapi import FastAPI, status
from pydantic import BaseModel, EmailStr

class KullaniciOlustur(BaseModel):
    eposta: EmailStr
    sifre: str

class KullaniciYanit(BaseModel):
    id: int
    eposta: EmailStr

@app.post("/kullanicilar", response_model=KullaniciYanit, status_code=status.HTTP_201_CREATED)
def kayit(veri: KullaniciOlustur):
    kayit = {"id": 7, "eposta": veri.eposta, "sifre_hash": "pbkdf2$..."}
    return kayit                     # sifre_hash yanıtta GÖRÜNMEZ

from pydantic import BaseModel

class KullaniciYanit(BaseModel):
    id: int
    eposta: str

db_kaydi = {"id": 7, "eposta": "ayse@ilgazbank.com.tr", "sifre_hash": "pbkdf2$9f...", "rol": "admin"}
print(KullaniciYanit.model_validate(db_kaydi).model_dump())
print(KullaniciYanit.model_json_schema()["required"])

Hata yönetimi: HTTPException ve durum kodları

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse

class StokYetersiz(Exception):
    def __init__(self, urun):
        self.urun = urun

@app.get("/urunler/{urun_id}")
def urun_getir(urun_id: int):
    if urun_id not in URUNLER:
        raise HTTPException(status_code=404, detail="Ürün bulunamadı")
    return URUNLER[urun_id]

@app.exception_handler(StokYetersiz)
async def stok_hatasi(request: Request, hata: StokYetersiz):
    return JSONResponse(status_code=409, content={"detail": f"{hata.urun} için stok yetersiz"})

Flask'ta hatalı istek için genelde 400 dönülür; FastAPI şema doğrulamasında 422 kullanır. "İstek sözdizimsel olarak doğru (geçerli JSON) ama anlamsal olarak işlenemiyor" demektir. İstemci geliştiricileri detail listesindeki loc alanından hangi alanın hatalı olduğunu görür ve formda ilgili kutuyu işaretler.

Bağımlılık enjeksiyonu: Depends

Birçok uç nokta aynı hazırlığa ihtiyaç duyar: veritabanı oturumu açmak, giriş yapan kullanıcıyı bulmak, sayfalama parametrelerini okumak. FastAPI'de bu ortak parçalar bağımlılık (dependency) olarak yazılır ve Depends ile parametreye bağlanır. FastAPI her istekte bağımlılığı çağırır (onun da bağımlılıkları varsa önce onları), sonucu parametreye koyar. yield kullanan bağımlılıklarda yield'dan sonraki kod yanıt gönderildikten sonra çalışır: oturumu kapatmak için ideal.

from typing import Annotated
from fastapi import Depends, FastAPI, Header, HTTPException
from sqlalchemy.orm import Session

def get_db():
    db = SessionLocal()           # SQLAlchemy dersindeki sessionmaker
    try:
        yield db
    finally:
        db.close()                # yanıt gittikten sonra çalışır

def aktif_kullanici(authorization: Annotated[str | None, Header()] = None,
                    db: Session = Depends(get_db)):
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(401, "Kimlik doğrulanamadı", headers={"WWW-Authenticate": "Bearer"})
    kullanici = token_ile_bul(db, authorization.removeprefix("Bearer "))
    if kullanici is None:
        raise HTTPException(401, "Geçersiz token")
    return kullanici

class Sayfalama:
    def __init__(self, sayfa: int = 1, boyut: int = 20):
        self.sayfa, self.boyut = sayfa, min(boyut, 100)

@app.get("/siparislerim")
def siparislerim(kullanici=Depends(aktif_kullanici),
                 db: Session = Depends(get_db),           # aynı istekte get_db bir kez çalışır
                 s: Sayfalama = Depends()):
    return listele(db, kullanici.id, s.sayfa, s.boyut)

Bu tasarımın en büyük faydası test sırasında görülür: app.dependency_overrides[get_db] = sahte_db ile gerçek veritabanını tek satırda sahte bir sürümle değiştirirsin. FastAPI bağımlılıkları nasıl çözüyor? Metaprogramlama dersindeki inspect.signature ile. Özünü burada çalıştırabilirsin:

import inspect

class Depends:
    def __init__(self, bagimlilik):
        self.bagimlilik = bagimlilik

def coz(fn, istek):
    kwargs = {}
    for p in inspect.signature(fn).parameters.values():
        if isinstance(p.default, Depends):
            kwargs[p.name] = coz(p.default.bagimlilik, istek)   # özyinelemeli çöz
        elif p.name in istek:
            kwargs[p.name] = istek[p.name]
    return fn(**kwargs)

def sayfalama(sayfa=1, boyut=20):
    return {"sayfa": int(sayfa), "boyut": min(int(boyut), 100)}

def aktif_kullanici(token=None):
    return {"tok-ayse": "ayse"}.get(token, "anonim")

def siparislerim(kullanici=Depends(aktif_kullanici), s=Depends(sayfalama)):
    return f"{kullanici} için sayfa {s['sayfa']} (boyut {s['boyut']})"

print(coz(siparislerim, {"token": "tok-ayse", "sayfa": "3", "boyut": "500"}))
print(coz(siparislerim, {}))

async uç noktalar

FastAPI hem def hem async def uç noktaları destekler. async def ile tanımlanan uç nokta olay döngüsünde çalışır; içinde await ile asenkron HTTP istemcisi (httpx.AsyncClient), asenkron veritabanı sürücüsü kullanırsan tek işlem binlerce bekleyen isteği idare eder. Normal def ile tanımlananları FastAPI bir iş parçacığı havuzunda çalıştırır.

import asyncio

async def stok_sor(depo, urun):
    await asyncio.sleep(0.05)              # dış servis gecikmesi
    return {"İzmir": 12, "Ankara": 0, "Bursa": 5}[depo]

async def urun_stok(urun_id):              # @app.get("/urunler/{urun_id}/stok")
    depolar = ["İzmir", "Ankara", "Bursa"]
    sonuclar = await asyncio.gather(*(stok_sor(d, urun_id) for d in depolar))
    return {"urun_id": urun_id, "stok": dict(zip(depolar, sonuclar)), "toplam": sum(sonuclar)}

loop = asyncio.new_event_loop()
print(loop.run_until_complete(urun_stok(7)))
loop.close()

OpenAPI, CORS ve kimlik doğrulama

FastAPI tüm uç noktaları, parametreleri ve pydantic şemalarını OpenAPI standardında bir JSON belgesine çevirir. Bu belgeden Swagger arayüzü çıkar; ayrıca istemci kodu üreticileri (TypeScript, Kotlin) aynı belgeden otomatik istemci kütüphanesi üretebilir. Modelin JSON şemasını kendin de görebilirsin: SiparisOlustur.model_json_schema().

Tarayıcıda çalışan bir ön yüz (ör. https://magaza.anadolumarket.com) başka bir alan adındaki API'ye (https://api.anadolumarket.com) istek atacaksa tarayıcı CORS (Cross-Origin Resource Sharing) iznini kontrol eder:

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://magaza.anadolumarket.com"],   # "*" üretimde kullanma
    allow_methods=["GET", "POST"],
    allow_headers=["Authorization", "Content-Type"],
    allow_credentials=True,
)

Kimlik doğrulamada yaygın kalıp OAuth2 parola akışı + JWT'dir: istemci /token uç noktasına kullanıcı adı ve şifre gönderir, sunucu imzalı bir JWT (JSON Web Token) döner; istemci sonraki isteklerde Authorization: Bearer <token> başlığını gönderir. JWT üç parçadır: başlık, yük (kullanıcı kimliği, son kullanma zamanı exp) ve imza. Sunucu imzayı gizli anahtarla doğrular; yük şifreli değildir, yalnızca değiştirilemez. Bu yüzden içine şifre ya da kişisel veri koyma.

# pip install pyjwt "passlib[bcrypt]"
from datetime import datetime, timedelta, timezone
import jwt
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

GIZLI = os.environ["JWT_GIZLI_ANAHTAR"]
oauth2 = OAuth2PasswordBearer(tokenUrl="token")

@app.post("/token")
def token_al(form: OAuth2PasswordRequestForm = Depends()):
    kullanici = dogrula(form.username, form.password)       # bcrypt ile karşılaştır
    if not kullanici:
        raise HTTPException(401, "Hatalı kullanıcı adı ya da şifre")
    yuk = {"sub": str(kullanici.id), "exp": datetime.now(timezone.utc) + timedelta(minutes=30)}
    return {"access_token": jwt.encode(yuk, GIZLI, algorithm="HS256"), "token_type": "bearer"}

def aktif_kullanici(token: str = Depends(oauth2)):
    try:
        yuk = jwt.decode(token, GIZLI, algorithms=["HS256"])
    except jwt.PyJWTError:
        raise HTTPException(401, "Geçersiz ya da süresi dolmuş token")
    return kullanici_getir(int(yuk["sub"]))

TestClient ve çalıştırma

# tests/test_api.py
from fastapi.testclient import TestClient
from main import app, get_db

def sahte_db():
    yield {"urunler": {1: {"ad": "Çay"}}}

app.dependency_overrides[get_db] = sahte_db
client = TestClient(app)

def test_urun_bulunamadi():
    yanit = client.get("/urunler/999")
    assert yanit.status_code == 404
    assert yanit.json() == {"detail": "Ürün bulunamadı"}

def test_dogrulama():
    yanit = client.post("/siparisler", json={"musteri_id": 0})
    assert yanit.status_code == 422
    alanlar = {tuple(h["loc"]) for h in yanit.json()["detail"]}
    assert ("body", "musteri_id") in alanlar

# Geliştirme
fastapi dev main.py
# Üretim: birden çok işlem
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4
# ya da gunicorn yöneticisiyle
gunicorn main:app -k uvicorn.workers.UvicornWorker -w 4 -b 0.0.0.0:8000

Flask mı, FastAPI mı, Django mı?

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import StreamingResponse

app = FastAPI()

class Yonetici:
    def __init__(self):
        self.baglantilar: set[WebSocket] = set()
    async def baglan(self, ws: WebSocket):
        await ws.accept()
        self.baglantilar.add(ws)
    def ayril(self, ws: WebSocket):
        self.baglantilar.discard(ws)
    async def yayinla(self, mesaj: dict):
        for ws in list(self.baglantilar):
            try:
                await ws.send_json(mesaj)
            except Exception:
                self.ayril(ws)          # kopmuş istemciyi temizle

yonetici = Yonetici()

@app.websocket("/ws/siparisler")
async def siparis_akisi(ws: WebSocket):
    await yonetici.baglan(ws)
    try:
        while True:
            await ws.receive_text()     # istemciden gelen (ör. ping) mesajları dinle
    except WebSocketDisconnect:
        yonetici.ayril(ws)

@app.get("/olaylar")
async def olaylar():
    async def uret():
        for i in range(3):
            yield f"data: {{\"sira\": {i}}}\n\n"
    return StreamingResponse(uret(), media_type="text/event-stream")

import base64, json
import veri

con = veri.baglan("anadolu_market", tablolar=["orders"])

def imlec_yap(tarih, sid):
    return base64.urlsafe_b64encode(json.dumps([tarih, sid]).encode()).decode()

def sayfa(imlec=None, limit=5):
    sql, p = "SELECT order_id, order_date FROM orders", []
    if imlec:
        tarih, sid = json.loads(base64.urlsafe_b64decode(imlec))
        sql += " WHERE (order_date, order_id) < (?, ?)"
        p = [tarih, sid]
    satirlar = con.execute(sql + " ORDER BY order_date DESC, order_id DESC LIMIT ?", p + [limit]).fetchall()
    sonraki = imlec_yap(*satirlar[-1][::-1]) if len(satirlar) == limit else None
    return satirlar, sonraki

s1, im = sayfa()
print(s1)
print("imleç:", im)
s2, _ = sayfa(im)
print(s2)

class Kova:
    def __init__(self, kapasite, hiz):
        self.kapasite, self.hiz = kapasite, hiz
        self.jeton, self.son = float(kapasite), 0.0
    def izin(self, simdi):
        self.jeton = min(self.kapasite, self.jeton + (simdi - self.son) * self.hiz)
        self.son = simdi
        if self.jeton >= 1:
            self.jeton -= 1
            return True
        return False

k = Kova(kapasite=3, hiz=1)          # 3'lük patlama, saniyede 1 istek
print([k.izin(t) for t in (0, 0.1, 0.2, 0.3, 1.3, 1.4)])

import time
from fastapi import Request
from fastapi.responses import JSONResponse

kovalar: dict[str, Kova] = {}

@app.middleware("http")
async def hiz_siniri(request: Request, call_next):
    anahtar = request.headers.get("X-API-Key") or request.client.host
    kova = kovalar.setdefault(anahtar, Kova(kapasite=20, hiz=5))
    if not kova.izin(time.monotonic()):
        return JSONResponse({"detail": "Çok fazla istek"}, status_code=429, headers={"Retry-After": "1"})
    return await call_next(request)

# pip install "sqlalchemy[asyncio]" asyncpg
from typing import Annotated
from fastapi import Depends
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

motor = create_async_engine("postgresql+asyncpg://kullanici:sifre@db/market", pool_size=10)
OturumYap = async_sessionmaker(motor, expire_on_commit=False)

async def oturum_al():
    async with OturumYap() as oturum:
        yield oturum                    # istek bitince otomatik kapanır

Oturum = Annotated[AsyncSession, Depends(oturum_al)]

@app.get("/siparisler/{sid}")
async def siparis(sid: int, oturum: Oturum):
    sonuc = await oturum.execute(select(Siparis).where(Siparis.id == sid))
    return sonuc.scalar_one_or_none()