Verdiğiniz eğitim içeriğini satır satır inceleyerek, belirttiğiniz kurallar doğrultusunda (dakika/saniye bilgileri, "Önizleme" ve "ders" satırları, anlamsız ifadeler temizlenerek) detaylı bir kaynak dokümana dönüştürdüm.

İlk olarak "Öğrenecekleriniz" bölümünün kapsamlı bir açıklamasını yapıp ardından müfredattaki tüm ana konuları derinlemesine, teorik anlatımlar, kod tasarımları ve pratik uygulama örnekleriyle açıkladım.

1. "Öğrenecekleriniz" Bölümünün Detaylı Açıklaması

Bu eğitim serisi, modern yazılım dünyasının en kritik bileşenlerinden biri olan API (Uygulama Programlama Arayüzü) kavramını sıfırdan ele almayı hedefler. Katılımcılara sadece kod yazmayı değil, modern mimarilerin temelini oluşturan REST ilkelerini ve internetin iletişim dili olan HTTP metotlarını (GET, POST, PUT, DELETE) öğretmeyi amaçlar.

Eğitimin ana odağı olan FastAPI, Python dünyasındaki en performanslı ve modern asenkron framework'tür. Kurs boyunca bir yazılımcının production (canlı sistem) ortamında ihtiyaç duyacağı tüm modern yetkinlikler adım adım aktarılmaktadır:

2. Konu Başlıklarının Detaylı Açıklamaları, Tasarımları ve Uygulamaları

BÖLÜM I: API Kavramı ve FastAPI Temelleri

1. API Nedir ve Nasıl Çalışır?

Açıklama: API (Application Programming Interface), iki farklı yazılımın veya cihazın birbiriyle güvenli ve kurallı bir şekilde iletişim kurmasını sağlayan bir köprüdür. Örneğin; telefonunuzdaki hava durumu uygulaması, Meteoroloji Genel Müdürlüğü'nün sunucularına bir API isteği gönderir ve oradan aldığı güncel verileri size görsel olarak sunar.

2. REST İlkeleri ve HTTP Metotları

Açıklama: REST (Representational State Transfer), web servisleri tasarlanırken uyulması gereken bir mimari felsefedir. REST mimarisinde her şey bir kaynaktır (Resource) (Örn: Kullanıcılar, Ürünler, Makaleler). Bu kaynaklar üzerinde işlem yapmak için standart HTTP metotları kullanılır:

3. FastAPI Nedir ve Diğer Frameworklerden Farkı Nedir?

Açıklama: FastAPI; Python tabanlı, asenkron (async/await) destekli, yüksek performanslı ve modern bir mikro-web framework'üdür.

🛠️ İlk FastAPI Sunucusu Uygulaması

İlk FastAPI projemizi ayağa kaldıralım.

Kurulum Komutları:

pip install fastapi uvicorn

Uygulama Kodu (main.py):

from fastapi import FastAPI

# FastAPI nesnesini başlatıyoruz
app = FastAPI(title="İlk API Sunucum")

# Kök (Root) dizine yapılan GET isteğini yakalıyoruz
@app.get("/")
def ana_sayfa():
    return {"mesaj": "Merhaba Dünya! FastAPI sunucunuz başarıyla çalışıyor."}

Sunucuyu Başlatma Komutu:

uvicorn main:app --reload

--reload parametresi, kodda bir değişiklik yaptığınızda sunucunun otomatik olarak yeniden başlamasını sağlar.

BÖLÜM II: API Uç Noktaları (Endpoints) Geliştirme

1. Yol Parametreleri (Path Parameters)

Açıklama: Belirli tek bir kaynağa erişmek için URL rotasının içerisine dinamik olarak yerleştirilen değişkenlerdir. Örneğin; ://magaza.com adresindeki 45 bir yol parametresidir ve 45 id'li ürünü temsil eder.

2. Sorgu Parametreleri (Query Parameters)

Açıklama: URL'in sonuna ? işaretinden sonra eklenen, verileri filtrelemek, sıralamak veya sayfalamak için kullanılan anahtar-değer çiftleridir. Örneğin; ://magaza.com adresinde kategori ve sirala birer sorgu parametresidir.

3. Pydantic Modelleri ve İstek Gövdeleri (Request Bodies)

Açıklama: POST veya PUT isteklerinde sunucuya gönderilecek karmaşık veriler URL üzerinden gönderilemez; HTTP İstek Gövdesinde (Request Body) JSON formatında gönderilir. Pydantic, bu gelen JSON verisinin şemasını çıkartmamızı, veri tiplerini (int, string, email vb.) otomatik doğrulamamızı sağlar.

🛠️ CRUD (Create, Read, Update, Delete) Endpoints Uygulaması

Bellek üzerinde (In-Memory) çalışan, yol parametresi, sorgu parametresi ve Pydantic modeli içeren tam bir ürün yönetim API'si tasarlayalım.

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, Field
from typing import Optional

app = FastAPI()

# Sahte Veritabanı (Bellek Üstünde)
URUNLER_DB = {
    1: {"isim": "Kablosuz Kulaklık", "fiyat": 1250.0, "stokta_mi": True},
    2: {"isim": "Akıllı Saat", "fiyat": 3400.0, "stokta_mi": True}
}

# Pydantic Şeması (Veri Doğrulama için)
class UrunSema(BaseModel):
    isim: str = Field(..., min_length=2, max_length=50)
    fiyat: float = Field(..., gt=0, description="Fiyat sıfırdan büyük olmalıdır.")
    stokta_mi: Optional[bool] = True

# 1. READ ALL (Sorgu Parametresi Filtreli GET)
@app.get("/urunler")
def urunleri_listele(min_fiyat: Optional[float] = None):
    if min_fiyat:
        filtrelenmiş = {k: v for k, v in URUNLER_DB.items() if v["fiyat"] >= min_fiyat}
        return filtrelenmiş
    return URUNLER_DB

# 2. READ ONE (Yol Parametreli GET)
@app.get("/urunler/{urun_id}")
def urun_getir(urun_id: int):
    if urun_id not in URUNLER_DB:
        raise HTTPException(status_code=404, detail="Ürün bulunamadı!")
    return URUNLER_DB[urun_id]

# 3. CREATE (Request Body içeren POST)
@app.post("/urunler", status_code=status.HTTP_201_CREATED)
def urun_olustur(urun: UrunSema):
    yeni_id = max(URUNLER_DB.keys()) + 1 if URUNLER_DB else 1
    URUNLER_DB[yeni_id] = urun.model_dump()
    return {"mesaj": "Ürün başarıyla eklendi", "id": yeni_id, "veri": URUNLER_DB[yeni_id]}

# 4. DELETE (DELETE)
@app.delete("/urunler/{urun_id}")
def urun_sil(urun_id: int):
    if urun_id not in URUNLER_DB:
        raise HTTPException(status_code=404, detail="Silinmek istenen ürün mevcut değil.")
    del URUNLER_DB[urun_id]
    return {"mesaj": f"ID'si {urun_id} olan ürün başarıyla silindi."}

BÖLÜM III: Gerçek Zamanlı Proje - AI Yapay Zeka Özgeçmiş (Resume) Analiz Aracı

1. Proje Mimarisi ve İşleyiş Akışı

Açıklama: Bu modülde istemci sisteme bir PDF/Metin formatında özgeçmiş yükler. FastAPI bu dosyayı alır, metne dönüştürür ve OpenAI veya Google Gemini API'lerine bir "Prompt" ile gönderir. Gelen yapılandırılmış analiz sonuçlarını (Adayın yetenekleri, tecrübe yılı, uygunluk skoru) kullanıcıya JSON formatında döner.

🛠️ Yapay Zeka Entegrasyon Uygulaması

Temel düzeyde bir OpenAI entegrasyonu sağlayan endpoint tasarımı yapalım.

from fastapi import FastAPI, UploadFile, File, HTTPException
import os

app = FastAPI(title="AI Resume Parser")

# Gerçek projede 'pip install openai' kurulmalı ve API Key tanımlanmalıdır.
# Örnek simülasyon amaçlı mock AI fonksiyonu yazıyoruz.
def yapay_zeka_analiz_et(metin: str):
    # Burada normal şartlarda OpenAI API çağrısı yapılır.
    if "Python" in metin or "FastAPI" in metin:
        return {"skor": 95, "uygunluk": "Yüksek", "Eksiklikler": "Yok"}
    return {"skor": 40, "uygunluk": "Düşük", "Eksiklikler": "Backend tecrübesi eksik"}

@app.post("/resume/analyze")
async def resume_analiz_et(file: UploadFile = File(...)):
    # Dosya uzantı kontrolü
    if not file.filename.endswith('.txt'):
        raise HTTPException(status_code=400, detail="Şimdilik sadece .txt formatındaki CV'ler desteklenmektedir.")
    
    # Dosya içeriğini okuma
    icerik = await file.read()
    cv_metni = icerik.decode("utf-8")
    
    # Yapay zekaya gönderim ve sonuç
    analiz_sonucu = yapay_zeka_analiz_et(cv_metni)
    
    return {
        "dosya_adi": file.filename,
        "analiz": analiz_sonucu
    }

BÖLÜM IV: Veritabanı Entegrasyonu (SQLAlchemy ORM)

1. ORM (Object-Relational Mapping) Mantığı ve SQLite

Açıklama: Klasik yöntemlerde veritabanı işlemleri için SELECT * FROM kullanıcılar gibi SQL cümleleri yazılır. ORM mimarisinde ise veritabanındaki her bir tabloya karşılık gelen bir Python sınıfı (Model) tanımlanır. Veri eklemek için sadece o sınıftan bir nesne üretip kaydetmek yeterlidir. Eğitimde hafif, ayar gerektirmeyen yerel bir veritabanı olan SQLite kullanılmaktadır.

🛠️ SQLAlchemy Yapılandırması ve CRUD İşlemleri Uygulaması

Kurulum:

pip install sqlalchemy

Veritabanı Altyapısı (database.py):

from sqlalchemy import create_engine, Column, Integer, String, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLITE_URL = "sqlite:///./test_veritabani.db"

engine = create_engine(SQLITE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# Bağımlılık Enjeksiyonu (Dependency Injection) için Veritabanı Session fonksiyonu
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Veritabanı Modeli ve API Entegrasyonu (app.py):

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
import database # Yukarıda yazdığımız dosya

# Tabloyu Tanımlayalım
class KullaniciModel(database.Base):
    __tablename__ = "kullanicilar"
    id = database.Column(database.Integer, primary_key=True, index=True)
    isim = database.Column(database.String)
    email = database.Column(database.String, unique=True)

# Tabloları fiziksel dosyada oluşturalım
database.Base.metadata.create_all(bind=database.engine)

app = FastAPI()

# Pydantic Girdi Şeması
from pydantic import BaseModel
class KullaniciOlusturSema(BaseModel):
    isim: str
    email: str

@app.post("/kullanici-ekle")
def veritabanina_yaz(kullanici: KullaniciOlusturSema, db: Session = Depends(database.get_db)):
    # Email kontrolü
    mevcut = db.query(KullaniciModel).filter(KullaniciModel.email == kullanici.email).first()
    if mevcut:
        raise HTTPException(status_code=400, detail="Bu email zaten kayıtlı.")
        
    # ORM objesi oluşturma ve kaydetme
    yeni_kullanici = KullaniciModel(isim=kullanici.isim, email=kullanici.email)
    db.add(yeni_kullanici)
    db.commit()
    db.refresh(yeni_kullanici)
    return {"mesaj": "Kullanıcı veritabanına kaydedildi", "id": yeni_kullanici.id}

BÖLÜM V: Kimlik Doğrulama (Authentication) ve Güvenlik

1. Şifrelerin Bcrypt ile Hashlenmesi

Açıklama: Kullanıcı şifreleri veritabanında asla "123456" gibi açık metin (plain text) olarak saklanamaz. Siber saldırılarda veritabanı ele geçirilse bile şifrelerin çözülememesi gerekir. Bcrypt, şifreyi geri döndürülemez matematiksel bir karmaşaya (hash) çevirir.

2. JWT (JSON Web Token) Nedir?

Açıklama: Kullanıcı sisteme başarılı bir şekilde giriş yaptığında (Login), sunucu ona şifrelenmiş, içinde kullanıcının ID'sini ve token'ın son kullanma tarihini barındıran bir metin anahtarı (JWT) verir. Kullanıcı sonraki adımlarda (Örn: Sepete ürün ekleme) şifresini göndermek yerine bu JWT token'ı isteğin başlığına (Header) ekleyerek kendini doğrular.

🛠️ JWT ve Bcrypt Güvenlik Altyapısı Tasarımı

Kurulum:

pip install passlib[bcrypt] pyjwt

Güvenlik Modülü (security.py):

from passlib.context import CryptContext
import jwt
from datetime import datetime, timedelta

SECRET_KEY = "SUPER_GIZLI_ANAHTAR_KIMSENIN_BILMEMESI_GEREKIR"
ALGORITMA = "HS256"

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# 1. Şifreyi Hash'leme
def sifre_hashle(plain_password: str) -> str:
    return pwd_context.hash(plain_password)

# 2. Girilen Şifre Doğru mu Kontrolü
def sifre_dogrula(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

# 3. JWT Token Üretme
def jwt_token_uret(veri: dict, gecerlilik_suresi_dk: int = 15):
    to_encode = veri.copy()
    expire = datetime.utcnow() + timedelta(minutes=gecerlilik_suresi_dk)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITMA)
    return encoded_jwt

BÖLÜM VI: Capstone Projesi (Güvenli Kullanıcı Kimlik Doğrulama API'si)

Açıklama: Bu bölüm, önceki tüm bölümlerin (FastAPI, Pydantic, SQLAlchemy, Bcrypt ve JWT) birleştiği, uçtan uca üye olma (Register), giriş yapma (Login) ve sadece giriş yapmış kişilerin görebileceği Korumalı Rotaları içeren bitirme projesidir.

🛠️ Kapsamlı Capstone Projesi Uygulaması

Tüm mimariyi birleştiren production-ready güvenli üyelik akışı kod yapısı:

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
import database
import security
import jwt

app = FastAPI(title="Secure Auth Capstone API")

# OAuth2 Şeması kurulumu (Swagger dokümantasyonunda kilit ikonu açar)
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

# SQLAlchemy Kullanıcı Tablo Tanımı
class User(database.Base):
    __tablename__ = "users"
    id = database.Column(database.Integer, primary_key=True, index=True)
    username = database.Column(database.String, unique=True, index=True)
    hashed_password = database.Column(database.String)

database.Base.metadata.create_all(bind=database.engine)

# Pydantic Şemaları
from pydantic import BaseModel
class UserRegister(BaseModel):
    username: str
    password: str

# 1. REGISTER ENDPOINT (Kayıt Olma)
@app.post("/register", status_code=status.HTTP_201_CREATED)
def register(user_data: UserRegister, db: Session = Depends(database.get_db)):
    db_user = db.query(User).filter(User.username == user_data.username).first()
    if db_user:
        raise HTTPException(status_code=400, detail="Bu kullanıcı adı zaten alınmış.")
    
    # Şifreyi güvenlik modülümüz ile hashleyip kaydediyoruz
    gizlenmis_sifre = security.sifre_hashle(user_data.password)
    yeni_user = User(username=user_data.username, hashed_password=gizlenmis_sifre)
    db.add(yeni_user)
    db.commit()
    return {"mesaj": "Kullanıcı başarıyla oluşturuldu."}

# 2. LOGIN ENDPOINT (Giriş Yap ve Token Al)
@app.post("/login")
def login(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(database.get_db)):
    db_user = db.query(User).filter(User.username == form_data.username).first()
    if not db_user or not security.sifre_dogrula(form_data.password, db_user.hashed_password):
        raise HTTPException(status_code=401, detail="Kullanıcı adı veya şifre hatalı.")
    
    # Giriş başarılı ise JWT token üretiliyor
    access_token = security.jwt_token_uret(veri={"sub": db_user.username})
    return {"access_token": access_token, "token_type": "bearer"}

# 3. PROTECTED ROUTE (Sadece Giriş Yapmış Kullanıcılara Özel Bölüm)
@app.get("/dashboard/profile")
def profile_sayfasi(token: str = Depends(oauth2_scheme)):
    try:
        # Gelen token'ı çözüp doğruluyoruz
        payload = jwt.decode(token, security.SECRET_KEY, algorithms=[security.ALGORITMA])
        username: str = payload.get("sub")
        if username is None:
            raise HTTPException(status_code=401, detail="Geçersiz Token")
    except jwt.PyJWTError:
        raise HTTPException(status_code=401, detail="Token süresi dolmuş veya geçersiz kimlik.")
        
    return {
        "mesaj": f"Hoş geldin {username}, bu bilgi sadece yetkili kullanıcılara gösterilir.",
        "gizli_veri": "Şirket finansal raporu v1.2"
    }

BÖLÜM VII: FastAPI Deployment (Canlıya Alma)

Açıklama: Yazılan API'lerin yerel bilgisayarımız dışında (localhost), tüm dünyanın erişebileceği bulut sunucularına aktarılması sürecidir. Production ortamında FastAPI uygulamasını doğrudan uvicorn ile değil, işçi süreçlerini (workers) yöneten Gunicorn arkasında Uvicorn işçileriyle çalıştırmak kararlılık sağlar.

Canlıya Alma Adımları Tasarımı:

  1. Gereksinim Dosyası (requirements.txt): Projenin bağımlılıkları listelenir. (pip freeze > requirements.txt)

  2. Uygulama Sunucusu Hazırlığı (Procfile veya Dockerfile): Sunucunun projeyi nasıl başlatacağını söyleyen komut satırıdır.

    web: uvicorn main:app --host 0.0.0.0 --port $PORT

  3. Platform Seçimi: AWS, Railway veya Render paneline GitHub reposu bağlanarak otomatik deploy (CI/CD) tetiklenir.



Gönderdiğiniz eğitim müfredatı belgesini satır satır inceleyerek, istediğiniz kurallar doğrultusunda (süreleri, "Önizleme" ve "ders" ibarelerini, anlamsız satırları temizleyerek) kapsamlı bir geliştirme ve öğrenim rehberine dönüştürdüm.

Bu eğitim, FastAPI ile Test Güdümlü Geliştirme (TDD) felsefesini kullanarak kurumsal, güvenli ve ölçeklenebilir backend projeleri üretmeyi hedefler. İçerik oldukça derin ve uzun olduğu için ilk aşamada Müfredat Tasarımı ve Planlaması ile birlikte "Öğrenecekleriniz" bölümünün detaylı teorik/pratik açılımını yapacağız. Ardından adım adım uygulamalı konu anlatımlarına geçeceğiz.

🎯 BÖLÜM 1: "ÖĞRENECEKLERİNİZ" PARAGRAFININ DETAYLI AÇIKLAMASI

Eğitimin bu giriş kısmı, modern bir Python backend geliştiricisinin production (canlı ortam) kalitesinde bir mimari kurabilmesi için ihtiyaç duyduğu araç setini tanımlar. Maddelerin teknik karşılıkları ve projeye katkıları şunlardır:

🏗️ BÖLÜM 2: KONU TASARIMI VE MİMARİ PLANLAMA

Projemizi inşa ederken uygulayacağımız klasör yapısı ve veri akış şeması şu şekilde tasarlanmıştır:

Proje Klasör Yapısı (Tasarım)

fastapi_tdd_project/
│
├── .env                  # Gizli ortam değişkenleri
├── Dockerfile            # API imaj tanımı
├── docker-compose.yml    # PostgreSQL, Adminer ve Test DB servisleri
├── requirements.txt      # Bağımlılıklar
├── alembic.ini           # Veritabanı göç konfigürasyonu
│
├── app/
│   ├── __init__.py
│   ├── main.py           # Uygulama ana giriş kapısı
│   ├── config.py         # Ayarların pydantic ile okunması
│   ├── database.py       # SQLAlchemy bağlantı motoru
│   ├── models/           # Veritabanı tabloları (SQLAlchemy)
│   ├── schemas/          # Veri doğrulama kalıpları (Pydantic)
│   └── routers/          # API Uç noktaları (Endpoints)
│
└── tests/
    ├── __init__.py
    ├── conftest.py       # Pytest fixture'ları (DB sıfırlama otomasyonu)
    ├── test_structure.py # Tablo ve şema doğrulama testleri
    └── test_endpoints.py # CRUD API entegrasyon testleri

💻 BÖLÜM 3: MÜFREDATIN SATIR SATIR İŞLENMESİ VE UYGULAMALAR

📦 KISIM 1: Geliştirme Ortamının ve Projenin Hazırlanması

Konu Başlıkları:

💡 Detaylı Açıklama ve Teori:

Sanal ortamlar, bilgisayarınızdaki küresel Python kurulumunu kirletmeden her projeye özel izole kütüphane odaları yaratmanızı sağlar. pip ile bağımlılık yönetimi yaparken kütüphane sürümlerini sabitlemek production ortamında çökme yaşanmasını engeller. Black kodu otomatik olarak standartlara göre biçimlendirirken, Flake8 kod hatalarını ve standart dışı yazımları tarar.

🛠️ Uygulama:

Proje klasörünü oluşturalım, sanal ortamı aktif edip temel paketleri yükleyelim.

# Klasör oluşturma ve giriş
mkdir fastapi_tdd_project && cd fastapi_tdd_project

# Sanal ortam oluşturma (Windows/Mac ortak mantığı)
python -m venv venv

# Sanal ortamı aktif etme
# Windows için: venv\Scripts\activate
# Mac/Linux için:
source venv/activate

# Gerekli bağımlılıkların requirements.txt dosyasına yazılması
cat <<EOT >> requirements.txt
fastapi==0.110.0
uvicorn==0.28.0
pydantic==2.6.4
black==24.2.0
flake8==7.0.0
EOT

# Paketlerin yüklenmesi
pip install -r requirements.txt

app/main.py dosyasının ilk hali:

from fastapi import FastAPI

app = FastAPI(title="FastAPI TDD E-Ticaret API")


@app.get("/")
def read_root():
    return {"status": "Uygulama Çalışıyor"}

🐳 KISIM 2: Veritabanı Ortam Konfigürasyonu ve Docker

Konu Başlıkları:

💡 Detaylı Açıklama ve Teori:

Docker, veritabanı gibi dış bileşenleri bilgisayarınıza doğrudan kurup konfigürasyon karmaşası yaşamak yerine, hafif bir sanal konteyner içinde saniyeler içinde ayağa kaldırmanızı sağlar. SQLAlchemy, Python sınıflarını (Class) SQL tablolarına dönüştüren bir ORM mimarisidir. Veritabanında bir kolon eklediğimizde bunu canlıya yansıtmak için Alembic aracıyla sürüm kontrolü (git benzeri) sağlarız.

🛠️ Uygulama:

Docker ortamını ve veritabanı bağlantı ayarlarını yapalım.

.env Dosyası:

DATABASE_URL=postgresql://user_tdd:pass_tdd@localhost:5432/db_tdd

docker-compose.yml Dosyası:

version: '3.8'

services:
  db:
    image: postgres:15-alpine
    container_name: tdd_postgres_db
    environment:
      POSTGRES_USER: user_tdd
      POSTGRES_PASSWORD: pass_tdd
      POSTGRES_DB: db_tdd
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data

  adminer:
    image: adminer
    container_name: tdd_adminer
    ports:
      - "8080:8080"

volumes:
  postgres_data:

(Bu dosyayı hazırladıktan sonra terminalde docker-compose up -d komutu çalıştırılarak PostgreSQL ve arayüz aracı Adminer ayağa kaldırılır).

app/database.py Dosyası:

import os
from dotenv import load_file
from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base, sessionmaker

# .env dosyasını yükle
load_file()

DATABASE_URL = os.getenv("DATABASE_URL")

engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()


# Veritabanı oturum bağımlılığı
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

🧪 KISIM 3: Test Ortamının Kurulması (Test Environment Setup)

Konu Başlıkları:

💡 Detaylı Açıklama ve Teori:

TDD (Test-Driven Development) yaparken en kritik kural testlerin izole olmasıdır. Testler gerçek veritabanındaki verileri bozmamalı, değiştirmemelidir. Bu yüzden testler başlarken geçici bir veritabanı şeması açılır, test verileriyle CRUD işlemleri yapılır ve test bittiğinde bu geçici alan tamamen yok edilir.

🛠️ Uygulama:

pytest paketini requirements'a ekleyip testlerin merkezi yönetim kalbi olan conftest.py yapısını kuralım.

tests/conftest.py Dosyası:

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.database import Base
from app.main import app
from fastapi.testclient import TestClient

# Test için izole geçici veritabanı (Bellekte veya test DB'sinde)
TEST_DATABASE_URL = "sqlite:///./test_temp.db"  # Kolaylık adına sqlite örneklenmiştir

engine = create_engine(TEST_DATABASE_URL, connect_args={"check_same_thread": False})
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)


@pytest.fixture(scope="session", autouse=True)
def setup_database():
    # Test başlamadan önce tabloları sıfırdan oluştur
    Base.metadata.create_all(bind=engine)
    yield
    # Testler bittikten sonra tabloları temizle / sil
    Base.metadata.drop_all(bind=engine)


@pytest.fixture
def db_session():
    connection = engine.connect()
    transaction = connection.begin()
    session = TestingSessionLocal(bind=connection)

    yield session

    session.close()
    transaction.rollback()
    connection.close()


@pytest.fixture
def client(db_session):
    # Uygulamanın veritabanı bağımlılığını test session'ı ile değiştiriyoruz
    def override_get_db():
        try:
            yield db_session
        finally:
            pass

    app.dependency_overrides[app.database.get_db] = override_get_db
    yield TestClient(app)
    del app.dependency_overrides[app.database.get_db]

🏗️ KISIM 4: Veritabanı Yapısal ve Migrasyon Testleri (Structural Testing)

Konu Başlıkları:

💡 Detaylı Açıklama ve Teori:

Geleneksel geliştirmede önce tablo oluşturulur, sonra kod yazılır. TDD yaklaşımında ise önce tablonun yapısal özelliklerini denetleyen test yazılır (Test Kalır - Red), ardından tablo modeli oluşturulur (Test Geçer - Green). Bu sayede veritabanı şemasında planlanan hiçbir kısıtlama (örneğin ismin boş geçilemez olması veya benzersiz olması) gözden kaçmaz.

🛠️ Uygulama (Önce Testi Yazalım):

Kategori tablomuzda id (Primary Key), name (String, Benzersiz, Boş bırakılamaz) ve is_active (Boolean, Varsayılan: True) alanları olmasını istiyoruz.

tests/test_structure.py Dosyası:

from sqlalchemy import inspect
from app.database import engine
from app.models.category import Category


def test_category_table_structure():
    inspector = inspect(engine)
    columns = inspector.get_columns("category")

    # 1. Test Case: Tablo var mı ve kolon isimleri doğru mu?
    col_names = [c["name"] for c in columns]
    assert "id" in col_names
    assert "name" in col_names
    assert "is_active" in col_names

    # 2. Test Case: Nullable (Boş bırakılabilirlik) kontrolü
    for c in columns:
        if c["name"] == "name":
            assert c["nullable"] is False  # name alanı boş bırakılamaz olmalı


def test_category_model_attributes():
    # 3. Test Case: Varsayılan değer kontrolü
    cat = Category(name="Elektronik")
    assert cat.is_active is True

(Bu testi şimdi çalıştırırsak Category modeli henüz var olmadığı için hata verecektir).

Şimdi Modeli Oluşturup Testi Geçirelim (app/models/category.py):

from sqlalchemy import Boolean, Column, Integer, String
from app.database import Base


class Category(Base):
    __tablename__ = "category"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, unique=True, nullable=False)
    is_active = Column(Boolean, default=True)

🌐 KISIM 5: API Uç Noktası (Endpoint) Geliştirme ve Unit Testleri

Konu Başlıkları:

💡 Detaylı Açıklama ve Teori:

Pydantic şemaları API'ye gelen JSON verisini doğrular. Eğer istemci eksik veya hatalı veri tipi gönderirse, FastAPI otomatik olarak 422 Unprocessable Entity döner. Unit test aşamasında mock yapılar veya doğrudan router fonksiyonları çağrılarak uç noktaların davranış mantığı (İstisnai durumlar dahil) denetlenir.

🛠️ Uygulama:

Önce Pydantic Şemasını Tasarlayalım (app/schemas/category.py):

from pydantic import BaseModel, Field


class CategoryCreate(BaseModel):
    name: str = Field(..., min_length=2, max_length=50)


class CategoryResponse(BaseModel):
    id: int
    name: str
    is_active: bool

    class Config:
        from_attributes = True

Endpoint Testlerinin Yazılması (tests/test_endpoints.py):

def test_create_category_endpoint(client):
    # Başarılı kayıt senaryosu
    response = client.post("/categories/", json={"name": "Ev & Yaşam"})
    assert response.status_code == 201
    assert response.json()["name"] == "Ev & Yaşam"
    assert "id" in response.json()


def test_create_category_bad_request(client):
    # Geçersiz veri (Kısa isim) gönderme senaryosu
    response = client.post("/categories/", json={"name": "A"})
    assert response.status_code == 422  # Pydantic validation hatası

Router ve Endpoint Kodunun Yazılması (app/routers/category.py):

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.database import get_db
from app.models.category import Category
from app.schemas.category import CategoryCreate, CategoryResponse

router = APIRouter(prefix="/categories", tags=["Categories"])


@router.post("/", response_model=CategoryResponse, status_code=status.HTTP_201_CREATED)
def create_category(category_in: CategoryCreate, db: Session = Depends(get_db)):
    # Benzersizlik kontrolü
    db_cat = db.query(Category).filter(Category.name == category_in.name).first()
    if db_cat:
        raise HTTPException(status_code=400, detail="Bu kategori zaten mevcut.")

    new_category = Category(name=category_in.name)
    db.add(new_category)
    db.commit()
    db.refresh(new_category)
    return new_category

🔗 KISIM 6: API Uç Noktası Entegrasyon Testleri (Integration Testing)

Konu Başlıkları:

💡 Detaylı Açıklama ve Teori:

Entegrasyon testleri, sistemin tüm parçalarının (API katmanı, Pydantic doğrulama katmanı, SQL Veritabanı ve kısıtlamalar) bir bütün olarak kusursuz entegre olup olmadığını doğrular. Kod kapsama analizi (pytest-cov), yazdığımız testlerin tüm proje satırlarının ne kadarına dokunduğunu söyler.

🛠️ Uygulama:

Kategorileri Listeleme (Get All), Güncelleme (Put) ve Silme (Delete) süreçlerinin entegrasyon testlerini ekleyelim.

tests/test_endpoints.py Devamı (Integration Tests):

def test_get_all_categories_integration(client, db_session):
    # Veritabanına test öncesi elle iki kayıt atalım
    cat1 = Category(name="Moda")
    cat2 = Category(name="Kozmetik")
    db_session.add_all([cat1, cat2])
    db_session.commit()

    # API'den tüm listeyi isteyelim
    response = client.get("/categories/")
    assert response.status_code == 200
    assert len(response.json()) >= 2


def test_delete_category_not_found(client):
    # Olmayan bir kategoriyi silmeye çalışma senaryosu
    response = client.delete("/categories/9999")
    assert response.status_code == 404

app/routers/category.py Tamamlayıcı CRUD Kodları:

from typing import List


@router.get("/", response_model=List[CategoryResponse])
def get_all_categories(db: Session = Depends(get_db)):
    return db.query(Category).all()


@router.delete("/{category_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_category(category_id: int, db: Session = Depends(get_db)):
    db_cat = db.query(Category).filter(Category.id == category_id).first()
    if not db_cat:
        raise HTTPException(
            status_code=404, detail="Silinmek istenen kategori bulunamadı."
        )

    db.delete(db_cat)
    db.commit()
    return None

📊 Testlerin Koşturulması ve Kod Kapsama Ölçümü:

Terminal üzerinden aşağıdaki komutlar çalıştırılarak test süreçleri izlenir:

# Tüm testleri çalıştırır
pytest

# Kod kapsama oranını (Coverage) ekrana basar
pytest --cov=app tests/



Eğitimin ikinci büyük aşaması olan Alembic Çoklu Veritabanı (Multi-Database) Yapılandırması ile İleri Düzey Tablo İlişkileri (ERD) ve Kısıtlama Testleri konularını sırasıyla, en ince yapısal detaylarına kadar ele alıyoruz.

🌐 BÖLÜM 4: ALEMBIC ÇOKLU VERİTABANI (MULTI-DATABASE) KONFİGÜRASYONU

Büyük ölçekli projelerde uygulama veritabanı (production) ile test veritabanının (test) veya operasyonel veritabanı (OLTP) ile analitik veritabanının (OLAP) şemalarını aynı anda göç ettirmek (migration) gerekebilir. Alembic buna "Multidb" veya "Multiple Databases" mimarisi der.

💡 Teorik Altyapı ve Mantık

Çoklu veritabanı senaryosunda tek bir alembic.ini ve tek bir env.py dosyası üzerinden birden fazla SQLAlchemy engine nesnesi yönetilir. Değişiklikler her veritabanı motoru için ayrı birer "bölüm" (section) olarak kaydedilir.

🛠️ Uygulama: Çoklu Veritabanı Altyapısının Kurulması

1. alembic.ini Dosyasının Yapılandırılması
Alembic projenizi alembic init --template multidb alembic_multidb komutuyla başlattığınızda veya mevcut dosyayı düzenlediğinizde veritabanı anahtarlarını tanımlamanız gerekir:

[alembic]
script_location = alembic_multidb
# Yönetilecek veritabanlarının isim listesi
databases = engine_prod, engine_test

[engine_prod]
sqlalchemy.url = postgresql://user_tdd:pass_tdd@localhost:5432/db_prod

[engine_test]
sqlalchemy.url = postgresql://user_tdd:pass_tdd@localhost:5432/db_test_clean

2. alembic_multidb/env.py Dosyasının Düzenlenmesi
Çoklu şema takibi için env.py içerisindeki run_migrations_online fonksiyonu her bir veritabanı motorunu tek tek dönerek tetiklemelidir:

import os
from logging.config import fileConfig
from alembic import context
from sqlalchemy import engine_from_config, pool
from app.database import Base  # Modellerinizin bağlı olduğu Base sınıfı

config = context.config
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# Hedef metadata nesnesi
target_metadata = {
    "engine_prod": Base.metadata,
    "engine_test": Base.metadata,
}


def run_migrations_online():
    """Çoklu veritabanı motoru için online migrasyon koşturma senaryosu."""
    # alembic.ini içindeki 'databases' değerini oku
    db_names = config.get_main_option("databases", "").split(",")

    engines = {}
    for name in db_names:
        name = name.strip()
        # Her veritabanı için dinamik url'leri çevresel değişkenlerden de ezebilirsiniz
        engines[name] = engine_from_config(
            config.get_section(name),
            prefix="sqlalchemy.",
            poolclass=pool.NullPool,
        )

    for name, engine in engines.items():
        print(
            f"--- {name} Veritabanı İçin Göç Senaryoları Çalıştırılıyor ---"
        )
        connection = engine.connect()
        context.configure(
            connection=connection,
            target_metadata=target_metadata.get(name),
        )
        try:
            with context.begin_transaction():
                context.run_migrations(engine_name=name)
        finally:
            connection.close()


if context.is_offline_mode():
    raise NotImplementedError(
        "Çoklu veritabanı mimarisinde offline mod desteklenmemektedir."
    )
else:
    run_migrations_online()

🏗️ BÖLÜM 5: İLERİ DÜZEY TABLO İLİŞKİLERİ VE ERD KISITLAMA TESTLERİ

E-ticaret sistemimizin çekirdeğini oluşturan gelişmiş ürün varyasyon mimarisine odaklanıyoruz. Tasarlayacağımız ERD (İlişki) şeması şu kısıtlamaları içermektedir:

  1. ProductLine (Ürün Serisi/Varyasyonu): Bir ürünün birden fazla varyasyonu (örn: Kırmızı-M, Mavi-L) olabilir.

  2. ProductImage (Ürün Görseli): Her ürün varyasyonunun sırasını (order) belirten birden fazla görseli olur.

  3. Seasonal Events (Mevsimsel Kampanyalar): Varyasyonların veya ürünlerin dahil olduğu dönemlik kısıtlamalar.

  4. Attributes & AttributeValue (Özellikler & Değerler): "Renk" (Attribute) -> "Kırmızı" (AttributeValue).

  5. ProductLine-AttributeValue: Çoka çok (Many-to-Many) ilişki tablosu.

  6. Product-ProductType: Ürün ile ürün tiplerini bağlayan çoka çok (Many-to-Many) ilişki tablosu.

🛠️ UYGULAMA 1: ProductLine Tablosu (Ürün Varyasyon Katmanı)

💡 Yapısal Kısıtlama Planı

🧪 Yapısal Test Kodu (tests/test_product_line.py)

TDD felsefesine sadık kalmak adına, henüz tabloyu yazmadan yapısal gereksinimlerimizi denetleyen testimizi kurguluyoruz:

import pytest
from sqlalchemy import inspect
from sqlalchemy.exc import IntegrityError
from app.database import engine
from app.models.product_line import ProductLine


def test_product_line_constraints():
    inspector = inspect(engine)
    columns = inspector.get_columns("product_line")
    col_dict = {c["name"]: c for c in columns}

    # 1. SKU Uzunluğu ve Nullable Kontrolü
    assert col_dict["sku"]["nullable"] is False
    # Not: SQLite ve Postgres tip uzunluklarını farklı saklayabilir, kontrol hassas olmalıdır
    assert col_dict["price"]["nullable"] is False

    # 2. Foreign Key Kontrolü
    fkeys = inspector.get_foreign_keys("product_line")
    fk_products = [fk for fk in fkeys if fk["referred_table"] == "product"]
    assert len(fk_products) == 1
    assert fk_products[0]["referred_columns"] == ["id"]

💻 Tablo Modelinin İnşası (app/models/product_line.py)

Testi başarıyla geçirecek kaynak kodumuzu yazıyoruz:

from sqlalchemy import Column, Decimal, ForeignKey, Integer, String
from sqlalchemy.orm import relationship
from app.database import Base


class ProductLine(Base):
    __tablename__ = "product_line"

    id = Column(Integer, primary_key=True, index=True)
    price = Column(Decimal(10, 2), nullable=False)
    sku = Column(String(20), unique=True, nullable=False, index=True)
    order = Column(Integer, default=1, nullable=False)

    # İlişkiler
    product_id = Column(
        Integer, ForeignKey("product.id", ondelete="CASCADE"), nullable=False
    )
    product = relationship("Product", back_populates="product_lines")
    images = relationship(
        "ProductImage", back_populates="product_line", cascade="all, delete-orphan"
    )

🛠️ UYGULAMA 2: ProductImage Tablosu (Görsel Katmanı)

💡 Yapısal Kısıtlama Planı

🧪 Yapısal Test Kodu (tests/test_product_image.py)

def test_product_image_structure():
    inspector = inspect(engine)
    columns = inspector.get_columns("product_image")
    col_dict = {c["name"]: c for c in columns}

    assert "url" in col_dict
    assert col_dict["url"]["nullable"] is False

    # İlişki kontrolü
    fkeys = inspector.get_foreign_keys("product_image")
    assert any(fk["referred_table"] == "product_line" for fk in fkeys)

💻 Tablo Modelinin İnşası (app/models/product_image.py)

from sqlalchemy import Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship
from app.database import Base


class ProductImage(Base):
    __tablename__ = "product_image"

    id = Column(Integer, primary_key=True, index=True)
    url = Column(String(255), nullable=False)
    order = Column(Integer, nullable=False, default=1)

    product_line_id = Column(
        Integer,
        ForeignKey("product_line.id", ondelete="CASCADE"),
        nullable=False,
    )
    product_line = relationship("ProductLine", back_populates="images")

🛠️ UYGULAMA 3: Seasonal Events Tablosu (Mevsimsel Kampanyalar)

💡 Yapısal Kısıtlama Planı

🧪 Yapısal Test Kodu (tests/test_seasonal_events.py)

def test_seasonal_event_unique_constraint(db_session):
    from app.models.seasonal_event import SeasonalEvent
    import datetime

    event1 = SeasonalEvent(
        name="Kış İndirimi",
        start_date=datetime.date.today(),
        end_date=datetime.date.today(),
    )
    db_session.add(event1)
    db_session.commit()

    # Aynı isimle ikinci kayıt eklenmeye çalışıldığında benzersizlik hatası fırlatmalı
    event2 = SeasonalEvent(
        name="Kış İndirimi",
        start_date=datetime.date.today(),
        end_date=datetime.date.today(),
    )
    db_session.add(event2)

    with pytest.raises(IntegrityError):
        db_session.commit()

💻 Tablo Modelinin İnşası (app/models/seasonal_event.py)

from sqlalchemy import Column, Date, Integer, String
from app.database import Base


class SeasonalEvent(Base):
    __tablename__ = "seasonal_event"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(100), unique=True, nullable=False)
    start_date = Column(Date, nullable=False)
    end_date = Column(Date, nullable=False)

🛠️ UYGULAMA 4: Attributes & AttributeValue (Özellik Sistemi)

💡 Yapısal Kısıtlama Planı

Dinamik bir ürün filtreleme sistemi kurmak için Attribute (Özellik Tipi) ve AttributeValue (Özellik Değeri) arasında 1-to-Many ilişkisi tasarlanır.

🧪 Yapısal Test Kodu (tests/test_attributes.py)

def test_attribute_and_value_relation(db_session):
    from app.models.attribute import Attribute
    from app.models.attribute_value import AttributeValue

    # Renk adında bir özellik tipi ekleyelim
    color_attr = Attribute(name="Renk", description="Ürün renk varyasyonları")
    db_session.add(color_attr)
    db_session.commit()

    # Bu tipe bağlı değerler ekleyelim
    val1 = AttributeValue(value="Kırmızı", attribute_id=color_attr.id)
    val2 = AttributeValue(value="Mavi", attribute_id=color_attr.id)
    db_session.add_all([val1, val2])
    db_session.commit()

    assert len(color_attr.values) == 2
    assert color_attr.values[0].value == "Kırmızı"

💻 Tablo Modellerinin İnşası

app/models/attribute.py Dosyası:

from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import relationship
from app.database import Base


class Attribute(Base):
    __tablename__ = "attribute"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50), unique=True, nullable=False)
    description = Column(String(255), nullable=True)

    values = relationship("AttributeValue", back_populates="attribute")

app/models/attribute_value.py Dosyası:

from sqlalchemy import Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship
from app.database import Base


class AttributeValue(Base):
    __tablename__ = "attribute_value"

    id = Column(Integer, primary_key=True, index=True)
    value = Column(String(50), nullable=False)

    attribute_id = Column(
        Integer, ForeignKey("attribute.id", ondelete="CASCADE"), nullable=False
    )
    attribute = relationship("Attribute", back_populates="values")

🛠️ UYGULAMA 5: Çoka Çok (Many-to-Many) İlişki Tabloları

Müfredatımızın en kritik yapısal test aşamalarından biri olan ara (association) tabloların kurulumudur.

💻 Modellerin ve Ara Bağlantıların İnşası

1. ProductLine-AttributeValue Many-to-Many Tanımı (app/models/product_line_attribute_value.py)

from sqlalchemy import Column, ForeignKey, Integer, Table
from app.database import Base

# Saf ilişki tablosu (Association Table)
product_line_attribute_value_association = Table(
    "product_line_attribute_value",
    Base.metadata,
    Column(
        "product_line_id",
        Integer,
        ForeignKey("product_line.id", ondelete="CASCADE"),
        primary_key=True,
    ),
    Column(
        "attribute_value_id",
        Integer,
        ForeignKey("attribute_value.id", ondelete="CASCADE"),
        primary_key=True,
    ),
)

2. Product-ProductType Many-to-Many Tanımı (app/models/product_product_type.py)

from sqlalchemy import Column, ForeignKey, Integer, Table
from app.database import Base

product_product_type_association = Table(
    "product_product_type",
    Base.metadata,
    Column(
        "product_id",
        Integer,
        ForeignKey("product.id", ondelete="CASCADE"),
        primary_key=True,
    ),
    Column(
        "product_type_id",
        Integer,
        ForeignKey("product_type.id", ondelete="CASCADE"),
        primary_key=True,
    ),
)

(Bu ara tablolar oluşturulduktan sonra ana koda dönülerek ProductLine modeline attribute_values = relationship("AttributeValue", secondary=product_line_attribute_value_association) bağı eklenir.)

🧪 Entegrasyon ve Veri Bütünlüğü Testi (tests/test_many_to_many.py)

Sistemde çoka çok verilerin silme senaryolarında (Cascade) öksüz kalıp kalmadığını denetleyen kapsamlı testimiz:

def test_product_line_attribute_value_flow(db_session):
    from app.models.product_line import ProductLine
    from app.models.attribute_value import AttributeValue
    from app.models.attribute import Attribute

    # Altyapı Hazırlığı
    attr = Attribute(name="Beden")
    db_session.add(attr)
    db_session.commit()

    val = AttributeValue(value="Large", attribute_id=attr.id)
    db_session.add(val)
    db_session.commit()

    # Ara tabloya kayıt atma simülasyonu
    # Bir product_line nesnesine doğrudan .attribute_values.append() yöntemiyle ekleme yapılabilir.



İleri düzey veri modellerimiz ve aralarındaki çoka çok (Many-to-Many) ilişkiler tamamlandığına göre, şimdi bu yapıyı istemciye güvenli bir şekilde sunma katmanına geçiyoruz.

Önce karmaşık ilişkisel verileri iç içe geçmiş (nested) JSON formatına hatasız dönüştüren Pydantic Serileştirme (Serialization) ve Doğrulama (Validation) Şemalarını yazacak ve test edeceğiz. Ardından, tüm bu ilişkileri tek bir istekte birleştiren Uç Nokta (Endpoint) Entegrasyon Testlerini hayata geçireceğiz.

🧬 BÖLÜM 6: PYDANTIC NESNELERİNİN SERİLEŞTİRME TESTLERİ

💡 Teorik Altyapı ve Mantık

SQLAlchemy veritabanından gelen veriler Python nesne grafikleridir (ilişkili nesneler zinciri). Pydantic v2 ile gelen from_attributes=True ayarı, bu nesneleri otomatik olarak okur ve doğrulanmış temiz JSON çıktılarına serileştirir. Test aşamasında, veritabanından veri çekiyormuş gibi sahte (mock) nesneler üreterek şemaların doğruluğunu denetleriz.

💻 1. Pydantic Şemalarının İnşası (app/schemas/product_composite.py)

Tüm gelişmiş şemaları, iç içe geçecek şekilde tek bir yapıda tasarlayalım:

from typing import List, Optional
from pydantic import BaseModel, Field, ConfigDict
from decimal import Decimal

# --- Özellik Değeri Şeması ---
class AttributeValueResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    
    id: int
    value: str

# --- Ürün Görseli Şeması ---
class ProductImageResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    
    id: int
    url: str
    order: int

# --- Ürün Varyasyon (ProductLine) Şeması ---
class ProductLineResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    
    id: int
    price: Decimal
    sku: str
    order: int
    images: List[ProductImageResponse]
    attribute_values: List[AttributeValueResponse] = Field(..., alias="attribute_values")

# --- Ana Ürün Detay Şeması (İç içe geçmiş tüm ilişkiler dahil) ---
class ProductDetailResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    
    id: int
    name: str
    description: Optional[str] = None
    product_lines: List[ProductLineResponse]

🧪 2. Pydantic Serileştirme Testinin Yazılması (tests/test_pydantic_serialization.py)

Şemalarımızın ilişkisel verileri kayıpsız ve doğru tiplerle JSON verisine dönüştürdüğünü doğrulamak için TDD testini kurguluyoruz:

import pytest
from decimal import Decimal
from app.schemas.product_composite import ProductDetailResponse

def test_product_detail_serialization_and_aliases():
    # 1. Sahte (Mock) ilişkisel nesne yapısı kurgulayalım
    class MockAttributeValue:
        def __init__(self, id, value):
            self.id = id
            self.value = value

    class MockProductImage:
        def __init__(self, id, url, order):
            self.id = id
            self.url = url
            self.order = order

    class MockProductLine:
        def __init__(self, id, price, sku, order, images, attribute_values):
            self.id = id
            self.price = price
            self.sku = sku
            self.order = order
            self.images = images
            self.attribute_values = attribute_values

    class MockProduct:
        def __init__(self, id, name, description, product_lines):
            self.id = id
            self.name = name
            self.description = description
            self.product_lines = product_lines

    # 2. İlişki zincirini bağlayalım (Örn: Kırmızı renkli, görseli olan bir XL tişört varyasyonu)
    mock_attr = MockAttributeValue(id=10, value="Kırmızı")
    mock_img = MockProductImage(id=100, url="https://example.com", order=1)
    
    mock_line = MockProductLine(
        id=50, 
        price=Decimal("299.99"), 
        sku="TSHIRT-RED-XL", 
        order=1, 
        images=[mock_img], 
        attribute_values=[mock_attr]
    )
    
    mock_product = MockProduct(
        id=1, 
        name="Oversize Tişört", 
        description="Pamuklu rahat kesim", 
        product_lines=[mock_line]
    )

    # 3. Serileştirme İşlemini Tetikleyelim
    serialized_data = ProductDetailResponse.model_validate(mock_product).model_dump()

    # 4. Doğrulama (Assertions)
    assert serialized_data["id"] == 1
    assert serialized_data["name"] == "Oversize Tişört"
    assert len(serialized_data["product_lines"]) == 1
    
    # İç içe geçmiş alanların kontrolü
    target_line = serialized_data["product_lines"][0]
    assert target_line["sku"] == "TSHIRT-RED-XL"
    assert target_line["price"] == Decimal("299.99")
    assert target_line["images"][0]["url"] == "https://example.com"
    assert target_line["attribute_values"][0]["value"] == "Kırmızı"

🔗 BÖLÜM 7: TÜM İLİŞKİLERİ TÜKETEN UÇ NOKTA ENTEGRASYON TESTLERİ

Bu aşamada, veritabanına gerçek ilişkisel verileri yazacak, FastAPI istemcisi üzerinden tek bir istek atacak ve veritabanı motorunun join işlemlerini doğru yapıp yapmadığını uçtan uca denetleyeceğiz.

🧪 1. Entegrasyon Testinin Yazılması (tests/test_endpoints_integration.py)

Bu test, istemcinin /products/{product_id} endpoint'ine istek attığında dönen cevabın veritabanındaki tüm ilişkileri (Görseller, Varyasyonlar, Özellikler) doğru getirip getirmediğini doğrular:

import pytest
from app.models.category import Category
# Ürün ana modelimizin mevcut olduğunu varsayıyoruz
from sqlalchemy import Column, Integer, String, ForeignKey
from app.database import Base

# Test hazırlığı için mock veya gerçek Product tablosu bağlantısı
class Product(Base):
    __tablename__ = "product"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, nullable=False)
    description = Column(String)

def test_get_product_detail_with_all_relations_integration(client, db_session):
    from app.models.product_line import ProductLine
    from app.models.product_image import ProductImage
    from app.models.attribute import Attribute
    from app.models.attribute_value import AttributeValue

    # 1. ADIM: Veritabanına Gerçek İlişkisel İlişkileri Yazalım (Arrange)
    # Ana Ürün
    prod = Product(name="Koşu Ayakkabısı", description="Performans odaklı")
    db_session.add(prod)
    db_session.commit()

    # Özellik ve Özellik Değeri (Many-to-Many için)
    attr = Attribute(name="Numara")
    db_session.add(attr)
    db_session.commit()
    
    val = AttributeValue(value="42", attribute_id=attr.id)
    db_session.add(val)
    db_session.commit()

    # Ürün Varyasyonu (ProductLine)
    line = ProductLine(price=1499.50, sku="SHOES-42-RED", order=1, product_id=prod.id)
    # Çoka çok ilişki bağlama
    line.attribute_values.append(val)
    db_session.add(line)
    db_session.commit()

    # Varyasyon Görseli (ProductImage)
    img = ProductImage(url="https://example.com", order=1, product_line_id=line.id)
    db_session.add(img)
    db_session.commit()

    # 2. ADIM: API Uç Noktasını Tetikleyelim (Act)
    response = client.get(f"/products/{prod.id}")

    # 3. ADIM: Sonuçları Uçtan Uca Doğrulayalım (Assert)
    assert response.status_code == 200
    res_data = response.json()
    
    assert res_data["name"] == "Koşu Ayakkabısı"
    assert len(res_data["product_lines"]) == 1
    
    # Varyasyon içerik kontrolü
    assert res_data["product_lines"][0]["sku"] == "SHOES-42-RED"
    # Decimal verinin str/float dönüşüm kontrolü
    assert float(res_data["product_lines"][0]["price"]) == 1499.50
    
    # Görsel ve Many-to-Many ilişki kontrolü
    assert res_data["product_lines"][0]["images"][0]["url"] == "https://example.com"
    assert res_data["product_lines"][0]["attribute_values"][0]["value"] == "42"

def test_get_product_detail_not_found(client):
    # Olmayan ürün senaryosu
    response = client.get("/products/99999")
    assert response.status_code == 404
    assert response.json()["detail"] == "Aranan ürün sistemde bulunamadı."

💻 2. Endpoint Kodunun Yazılması (app/routers/product.py)

Yukarıdaki zorlu entegrasyon testlerini başarıyla yeşile (Green) çevirecek olan optimize edilmiş veritabanı sorgusunu (Eager Loading kullanarak) yazıyoruz:

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session, joinedload
from app.database import get_db
# Testteki Product modelinin import edildiğini varsayıyoruz
from tests.test_endpoints_integration import Product 
from app.models.product_line import ProductLine
from app.schemas.product_composite import ProductDetailResponse

router = APIRouter(prefix="/products", tags=["Products"])

@router.get("/{product_id}", response_model=ProductDetailResponse)
def get_product_detail(product_id: int, db: Session = Depends(get_db)):
    # N+1 sorgu problemini engellemek için joinedload (Eager Loading) kullanıyoruz
    product = db.query(Product).options(
        joinedload(Product.product_lines).joinedload(ProductLine.images),
        joinedload(Product.product_lines).joinedload(ProductLine.attribute_values)
    ).filter(Product.id == product_id).first()

    if not product:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, 
            detail="Aranan ürün sistemde bulunamadı."
        )
        
    return product

🚀 BÖLÜM 8: PROJENİN ANA GİRİŞ KAPISINA ENTEGRASYONU

Tüm geliştirdiğimiz bu router mekanizmalarını app/main.py dosyasına dahil ederek projeyi ayağa kaldırmaya hazır hale getiriyoruz.

app/main.py Son Hali:

from fastapi import FastAPI
from app.routers import category, product

app = FastAPI(
    title="FastAPI TDD Kurumsal Varyasyonlu Ürün API",
    description="İleri düzey ERD ilişkileri ve çoklu DB mimarisine sahip backend sistemi",
    version="1.0.0"
)

# Router'ların uygulamaya dahil edilmesi
app.include_router(category.router)
app.include_router(product.router)

@app.get("/", tags=["Root"])
def root_status():
    return {"status": "Sistem Sağlıklı, Tüm Testler Aktif"}

📊 Son Kontrol ve Raporlama

Aşağıdaki komutu terminalde çalıştırarak yazdığımız karmaşık mimarinin test kapsama (coverage) başarısını görebiliriz:

pytest --cov=app tests/

Böylece müfredatınızda yer alan tüm veri şemalarını, test stratejilerini, serileştirme mimarilerini ve entegrasyon süreçlerini uçtan uca pratik ve derinlemesine uygulamalarla tamamlamış olduk.



*******************************************************************************************

Görselde yer alan FastAPI Complete Course: JWT, SQLModel, Celery & Deployment kurs müfredatını sizden gelen tüm kurallar doğrultusunda (süreleri, "Önizleme" ve "ders" ibrelerini, anlamsız satırları temizleyerek) ve hiçbir şekilde özet yapmadan derinlemesine, teorik açıklamalar, pratik kod örnekleri ve uygulamalarla tasarlayarak parça parça ele alıyorum.

Dosyanın büyüklüğü ve konuların derinliği nedeniyle eğitim serisinin 1. Parçasını (Giriş, "Öğrenecekleriniz" bölümü ve Temel Web Sunucusu İşlemleri) aşağıda eksiksiz bir şekilde başlatıyorum.

FastAPI Kapsamlı Eğitim Müfredatı Detaylı Analizi

Bölüm 1: "Öğrenecekleriniz" Paragrafının Detaylı Açıklaması

Bu kursun ana hedefi, modern ve asenkron bir Python web framework'ü olan FastAPI kullanarak, üretim ortamına (production) hazır, güvenli ve ölçeklenebilir RESTful API'lar geliştirmektir. "Öğrenecekleriniz" başlığı altında verilen maddelerin teknik karşılıkları şu şekildedir:

Bölüm 2: Konuların Tasarımı ve Uygulamalı Detay açıklamaları

Konu 1: Introduction and Project Setup (Giriş ve Proje Kurulumu)

Bir FastAPI projesine başlarken en kritik adım, bağımlılıkların izole edildiği temiz bir sanal ortam (Virtual Environment) oluşturmak ve projenin iskeletini kurmaktır.

Uygulama: Proje Kurulumu ve İlk Çalıştırma

Sisteminizde Python yüklü olduğunu varsayarak terminal üzerinden şu adımları takip ederiz:

# 1. Sanal ortam oluşturma
python -m venv venv

# 2. Sanal ortamı aktif etme (Windows için: venv\Scripts\activate)
source venv/bin/activate

# 3. Gerekli kütüphanelerin yüklenmesi
pip install fastapi uvicorn

Kök dizinde bir main.py dosyası oluşturup projenin ilk ayağa kaldırılma kodunu yazarız:

# main.py
from fastapi import FastAPI

# FastAPI nesnesini somutlaştırıyoruz
app = FastAPI(title="FastAPI Kapsamlı Kurs Projesi")

@app.get("/")
async def root():
    return {"mesaj": "FastAPI Uygulamasına Hoş Geldiniz!"}

Çalıştırma Komutu:

uvicorn main:app --reload

Açıklama: --reload parametresi, kodda bir değişiklik yaptığınızda uvicorn sunucusunun otomatik olarak yenilenmesini sağlar. Bu, geliştirme aşamasında büyük kolaylık sağlar.

Konu 2: Web Server Basics: Path & Query Params, Request Body, Headers (Web Sunucu Temelleri)

İstemciden (Client) sunucuya veri göndermenin 4 temel yolu vardır: Path (Yol) Parametreleri, Query (Sorgu) Parametreleri, Request Body (İstek Gövdesi) ve Headers (Başlıklar).

1. Path Parameters (Yol Parametreleri)

URL yapısının ayrılmaz bir parçası olan, genellikle spesifik bir kaynağı işaret etmek için kullanılan parametrelerdir.

@app.get("/urunler/{urun_id}")
async def urun_detay_getir(urun_id: int):
    # FastAPI, urun_id'nin int olmasını zorunlu kılar. 
    # Eğer url'e /urunler/abc yazılırsa otomatik olarak veri doğrulama hatası (422) fırlatır.
    return {"urun_id": urun_id, "durum": "Aktif", "stok": 120}

2. Query Parameters (Sorgu Parametreleri)

URL'de ? işaretinden sonra gelen anahtar-değer çiftleridir. Genellikle filtreleme, sıralama ve sayfalama (pagination) işlemlerinde kullanılır.

from typing import Optional

@app.get("/urunler")
async def urunleri_listele(kategori: Optional[str] = None, limit: int = 10):
    # kategori opsiyoneldir, limit verilmezse varsayılan olarak 10 alınır.
    return {
        "filtre_kategorisi": kategori,
        "getirilen_adet": limit,
        "sonuc": f"{kategori if kategori else 'Tüm'} kategorisindeki ürünler listelendi."
    }

3. Request Body (İstek Gövdesi)

İstemcinin sunucuya (genellikle POST, PUT, PATCH isteklerinde) göndermek istediği ham veridir. FastAPI bu veriyi yakalamak ve doğrulamak için Pydantic modellerini kullanır.

from pydantic import BaseModel, Field

# Gönderilecek verinin şablonunu (Şemasını) tanımlıyoruz
class UrunEklemeSeması(BaseModel):
    isim: str = Field(..., min_length=2, max_length=50)
    fiyat: float = Field(..., gt=0, description="Fiyat sıfırdan büyük olmalıdır.")
    stok_adet: int

@app.post("/urunler/ekle")
async def yeni_urun_olustur(urun: UrunEklemeSeması):
    # İstemciden gelen JSON verisi otomatik olarak 'urun' nesnesine dönüştürülür.
    return {
        "mesaj": "Ürün başarıyla eklendi",
        "urun_detay": urun.dict()
    }

4. Headers (Başlıklar)

İstekle ilgili meta verileri (Yetkilendirme token'ları, tarayıcı bilgileri vb.) taşımak için kullanılır.

from fastapi import Header

@app.get("/guvenlik/kontrol")
async def user_agent_oku(user_agent: str = Header(None), x_custom_token: str = Header(None)):
    # HTTP başlığındaki 'User-Agent' ve 'X-Custom-Token' otomatik yakalanır.
    return {
        "Kullanıcı Tarayıcısı": user_agent,
        "Özel Token": x_custom_token
    }

Konu 3: Build a CRUD REST API: Response Models, Validation, and Exceptions (CRUD REST API İnşası)

Gerçek senaryolarda, istemciye veritabanındaki her veriyi (Örn: şifre karmaları, silinme logları) dönmek istemeyiz. Burada Response Models (Yanıt Modelleri) devreye girer. Ayrıca hatalı işlemlerde düzgün HTTP durum kodları ile Exception (İstisna) fırlatılması gerekir.

Uygulama: Bellek Üzerinde (In-Memory) CRUD ve İstisna Yönetimi

Aşağıdaki örnekte, henüz veritabanına geçmeden konunun mantığını kavramak için geçici bir Python listesi üzerinde tam bir CRUD senaryosu ve hata yönetimi uyguluyoruz:

from fastapi import HTTPException, status

# Sahte veritabanı simülasyonu
KURS_VERITABANI = [
    {"id": 1, "baslik": "Python Giriş", "egitmen": "Ahmet", "gizli_not": "Maliyet: 100TL"},
    {"id": 2, "baslik": "FastAPI İleri Seviye", "egitmen": "Mehmet", "gizli_not": "Maliyet: 200TL"}
]

# İstemciye sadece görmesi gereken alanları filtrelemek için Pydantic Modeli tanımlıyoruz
class KursYatitSemasi(BaseModel):
    id: int
    baslik: str
    egitmen: str
    # 'gizli_not' alanını buraya eklemediğimiz için dış dünyaya bu veri asla sızmayacak.

@app.get("/kurslar/{kurs_id}", response_model=KursYatitSemasi)
async def kurs_getir(kurs_id: int):
    for kurs in KURS_VERITABANI:
        if kurs["id"] == kurs_id:
            return kurs # response_model sayesinde otomatik filtrelenir.
    
    # Eğer kurs bulunamazsa 404 hatası fırlatıyoruz
    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail=f"{kurs_id} numaralı kurs sistemde bulunamadı!"
    )

Konu 4: Modular Project Structure with FastAPI Routers (FastAPI Routers ile Modüler Proje Yapısı)

Tüm API uç noktalarını tek bir dosyada toplamak projenin büyümesini engeller. FastAPI, kodları iş mantıklarına göre parçalamak için APIRouter sınıfını sunar.

Tasarım Şeması:

src/
│
├── main.py                 # Uygulamanın ana giriş kapısı
└── routers/
    ├── __init__.py
    ├── kullanıcılar.py      # Kullanıcı işlemleri API'ları
    └── kitaplar.py          # Kitap işlemleri API'ları

Uygulama: Router Yapılandırması

src/routers/kitaplar.py dosyasının içeriği:

from fastapi import APIRouter

router = APIRouter(
    prefix="/kitaplar",
    tags=["Kitap İşlemleri"]
)

@router.get("/")
async def tum_kitapları_getir():
    return [{"id": 1, "ad": "Suç ve Ceza"}]

src/main.py dosyasında bu router'ın ana uygulamaya dahil edilmesi:

from fastapi import FastAPI
from routers import kitaplar # Oluşturduğumuz modülü içeri alıyoruz

app = FastAPI()

# Router'ı uygulamaya kaydediyoruz
app.include_router(kitaplar.router)

Bu işlemden sonra, tarayıcıdan /kitaplar/ adresine gidildiğinde modüler olarak tasarladığımız uç nokta çalışacaktır.



Harika, eğitim serimizin en kritik mimari taşlarından biri olan Databases & SQLModel (Part 5, Part 6, Part 7 ve Part 8) bölümüne geçiyoruz.

Bu bölümde veritabanı bağlantılarını asenkron (async) hale getirmeyi, FastAPI'nin modern Lifespan etkinlik yapısını, hem Pydantic şeması hem de SQL tablosu işlevi gören SQLModel yapısını, Dependency Injection (Bağımlılık Enjeksiyonu) mekanizmasını ve veritabanı şema değişikliklerini yöneten Alembic aracını tek bir satırı bile atlamadan, teorik ve pratik olarak ele alıyoruz.

Bölüm 3: Databases & SQLModel Yapısı

Bu bölümün temel amacı, klasik ORM (Object-Relational Mapping) kütüphanelerinin (SQLAlchemy gibi) ve veri doğrulama kütüphanelerinin (Pydantic gibi) yarattığı kod tekrarını sıfıra indiren SQLModel kütüphanesini asenkron olarak projeye entegre etmektir.

Konu 5: Databases with SQLModel: Connection, Lifespan Events, and Models (Veritabanı Bağlantısı ve Lifespan Etkinlikleri)

Eski FastAPI sürümlerinde veritabanı başlatma ve kapatma işlemleri için startup ve shutdown etkinlikleri kullanılıyordu. Modern FastAPI mimarisinde bu işlem, uygulamanın tüm yaşam döngüsünü tek bir bağlam yöneticisi (context manager) altında toplayan Lifespan yapısı ile çözülür.

Uygulama: Asenkron Veritabanı Kurulumu ve Lifespan Yönetimi

Öncelikle gerekli kütüphaneleri ortamımıza dahil etmemiz gerekir:

pip install sqlmodel asyncpg

(Not: asyncpg, PostgreSQL ile asenkron iletişim kurmak için kullanılan güçlü bir sürücüdür.)

src/db.py (Veritabanı Yapılandırma Dosyası):

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from sqlmodel import SQLModel

# Asenkron PostgreSQL bağlantı dizesi
# Gerçek senaryoda bu bilgi .env dosyasından okunmalıdır.
ASYNC_DATABASE_URL = "postgresql+asyncpg://kullanici:sifre@localhost:5432/fastapi_db"

# Asenkron motoru (engine) oluşturuyoruz
async_engine = create_async_engine(
    ASYNC_DATABASE_URL,
    echo=True, # Veritabanında dönen SQL sorgularını terminale yazdırır (Geliştirme için ideal)
    future=True
)

# Tabloları veritabanında otomatik oluşturacak yardımcı fonksiyon
async def init_db():
    async with async_engine.begin() as conn:
        # Kod tabanlı SQLModel modellerini okur ve veritabanında tabloları kurar
        await conn.run_sync(SQLModel.metadata.create_all)

src/main.py (Lifespan Entegrasyonu):

from contextlib import asynccontextmanager
from fastapi import FastAPI
from db import init_db

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Uygulama AYAĞA KALKARKEN çalışacak kısım:
    print("Uygulama başlatılıyor, veritabanı bağlantısı kuruluyor...")
    await init_db() 
    
    yield # Uygulama çalışırken burası askıda bekler
    
    # Uygulama KAPANIRKEN çalışacak kısım:
    print("Uygulama kapatılıyor, bağlantılar güvenli bir şekilde sonlandırılıyor...")

app = FastAPI(lifespan=lifespan)

Konu 6: CRUD with Async SQLModel: Introduction to Dependency Injection (Asenkron CRUD ve Bağımlılık Enjeksiyonu)

FastAPI’nin en güçlü yanlarından biri Depends fonksiyonu ile sağlanan Dependency Injection (Bağımlılık Enjeksiyonu) mekanizmasıdır. Her HTTP isteği geldiğinde yeni bir veritabanı oturumu (Session) açılmasını, işlem bittiğinde ise bu oturumun otomatik olarak kapatılmasını sağlar.

Uygulama: Bağımlılık Oluşturma ve Ürün CRUD İşlemleri

src/db.py dosyasına eklenecek session fonksiyonu:

async def get_session() -> AsyncSession:
    # Asenkron session fabrikası kuruyoruz
    async_session = sessionmaker(
        async_engine, class_=AsyncSession, expire_on_commit=False
    )
    async with async_session() as session:
        yield session # Oturumu teslim eder, endpoint işini bitirince buraya döner ve oturumu kapatır.

src/models.py (SQLModel Tanımı):

from typing import Optional
from sqlmodel import Field, SQLModel

# Bu sınıf hem bir Pydantic şeması hem de bir SQL tablosudur!
class Kurs(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    baslik: str = Field(index=True)
    egitmen: str
    fiyat: float

src/routers/kurslar.py (Asenkron CRUD Router):

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from db import get_session
from models import Kurs

router = APIRouter(prefix="/kurslar", tags=["Kurs Yönetimi"])

# 1. CREATE: Yeni Kurs Ekleme
@router.post("/", status_code=status.HTTP_201_CREATED)
async def kurs_ekle(yeni_kurs: Kurs, session: AsyncSession = Depends(get_session)):
    session.add(yeni_kurs)
    await session.commit() # Veritabanına kaydet
    await session.refresh(yeni_kurs) # Veritabanında oluşan id'yi nesneye yükle
    return yeni_kurs

# 2. READ: Tüm Kursları Listeleme
@router.get("/")
async def kursları_listele(session: AsyncSession = Depends(get_session)):
    sorgu = select(Kurs)
    sonuc = await session.execute(sorgu)
    tum_kurslar = sonuc.scalars().all()
    return tum_kurslar

# 3. READ: Tek Bir Kurs Getirme
@router.get("/{kurs_id}")
async def kurs_detay(kurs_id: int, session: AsyncSession = Depends(get_session)):
    kurs = await session.get(Kurs, kurs_id)
    if not kurs:
        raise HTTPException(status_code=404, detail="Kurs bulunamadı")
    return kurs

Konu 7: Create a User Authentication Model & Database Migrations with Alembic (Kullanıcı Modeli ve Alembic Migrasyonları)

Üretim ortamındaki veritabanlarında SQLModel.metadata.create_all kullanmak tehlikelidir çünkü mevcut tabloları güncelleyemez, sadece sıfırdan oluşturur. Tabloya yeni bir kolon eklendiğinde verileri kaybetmeden veritabanını güncellemek için Alembic kullanılır.

1. Kullanıcı Kimlik Doğrulama Modeli Tasarımı

Kullanıcı kayıt ve giriş işlemlerinde kullanacağımız temel modeli tanımlayalım.

src/models.py (Kullanıcı Tablosu):

import datetime
from sqlmodel import Field, SQLModel

class User(SQLModel, table=True):
    id: int = Field(default=None, primary_key=True)
    email: str = Field(unique=True, index=True, nullable=False)
    password_hash: str = Field(nullable=False) # Şifreleri asla düz metin olarak tutmuyoruz
    is_active: bool = Field(default=True)
    is_superuser: bool = Field(default=False)
    created_at: datetime.datetime = Field(default_factory=datetime.datetime.utcnow)

2. Alembic Kurulumu ve Canlı Migrasyon Adımları

Terminal üzerinden Alembic yapılandırmasını başlatıyoruz:

# Alembic kütüphanesini kurun
pip install alembic

# Projede alembic ortamını başlatın
alembic init alembic

Bu komuttan sonra proje kök dizininde alembic.ini dosyası ve alembic/ klasörü oluşur. Alembic'in asenkron projemizi ve SQLModel yapımızı tanıyabilmesi için iki kritik dosya düzenlemesi yapmalıyız:

alembic.ini dosyasında veritabanı adresini ayarlayın:

sqlalchemy.url = postgresql+asyncpg://kullanici:sifre@localhost:5432/fastapi_db

alembic/env.py dosyasında modellerimizi ve hedefi tanıtın:

# env.py dosyasının üst kısımlarına ekleyin:
from sqlmodel import SQLModel
from models import User, Kurs # Tüm modellerimizi içeri aktarıyoruz

# target_metadata satırını bulup şu şekilde güncelleyin:
target_metadata = SQLModel.metadata

İlk Migrasyonu Üretme ve Veritabanına Uygulama Komutları:

# Değişiklikleri tespit edip otomatik migrasyon dosyası oluşturur
alembic revision --autogenerate -m "kullanici ve kurs tablolari eklendi"

# Oluşan bu şemayı gerçek veritabanına uygular (Tabloları basar)
alembic upgrade head

Konu 8: User Account Creation: Email & Password (Kullanıcı Hesap Oluşturma İşlemleri)

Kullanıcı hesapları oluşturulurken şifrelerin güvenliği en kritik konudur. Şifreler veritabanına kaydedilmeden önce tek yönlü kriptografik algoritmalardan (örneğin bcrypt) geçirilerek "hash"lenmelidir.

Uygulama: Şifre Güvenliği Altyapısı ve Kayıt API'sı

Önce şifreleme kütüphanesini kuralım:

pip install "passlib[bcrypt]"

src/security.py (Güvenlik Yardımcı Dosyası):

from passlib.context import CryptContext

# Bcrypt algoritmasını kullanacak bağlamı tanımlıyoruz
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

src/schemas.py (Giriş/Çıkış Şemaları):
İstemcinin şifresini alırken kullanacağımız ama veritabanındaki hash değerini dışarı sızdırmayacak ara şemalar (DTO - Data Transfer Object):

from pydantic import BaseModel, EmailStr

class UserCreateSchema(BaseModel):
    email: EmailStr
    password: str # Kullanıcının ham şifresi (istek esnasında)

class UserResponseSchema(BaseModel):
    id: int
    email: EmailStr
    is_active: bool
    # Şifre alanı burada yok! Güvenli çıktı sağlar.

src/routers/auth.py (Kullanıcı Kayıt API Uç Noktası):

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from db import get_session
from models import User
from schemas import UserCreateSchema, UserResponseSchema
from security import hash_password

router = APIRouter(prefix="/auth", tags=["Kimlik Doğrulama"])

@router.post("/register", response_model=UserResponseSchema, status_code=status.HTTP_201_CREATED)
async def register_user(user_data: UserCreateSchema, session: AsyncSession = Depends(get_session)):
    # 1. Email adresi sistemde zaten var mı kontrol et
    sorgu = select(User).where(User.email == user_data.email)
    sonuc = await session.execute(sorgu)
    mevcut_kullanici = sonuc.scalars().first()
    
    if mevcut_kullanici:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Bu e-posta adresiyle zaten bir kullanıcı kayıtlı."
        )
    
    # 2. Şifreyi hash'le ve yeni kullanıcı nesnesini oluştur
    yeni_kullanici = User(
        email=user_data.email,
        password_hash=hash_password(user_data.password) # Güvenli hashleme
    )
    
    # 3. Veritabanına kaydet
    session.add(yeni_kullanici)
    await session.commit()
    await session.refresh(yeni_kullanici)
    
    return yeni_kullanici # Şema sayesinde sadece güvenli alanlar döner



Harika! Projemizin en kritik güvenlik duvarını öreceğimiz Authentication & User Management (Part 9, Part 10, Part 11, Part 12 ve Part 13) bölümüne geçiyoruz.

Bu bölümde endüstri standardı olan JWT (JSON Web Tokens) mimarisini kuracak, kısa ömürlü Access Token ile uzun ömürlü Refresh Token mekanizmasını inşa edecek, API uç noktalarımızı HTTP Bearer Auth ile koruma altına alacağız. Ayrıca kullanıcı çıkış yaptığında token'ları anında geçersiz kılmak için Redis tabanlı bir Token Revocation (İptal) sistemi tasarlayacak ve en sonunda Rol Tabanlı Erişim Kontrolü (RBAC) uygulayacağız.

Bölüm 4: Authentication & User Management

Bu katmanda amaç; kullanıcının kimliğini doğrulamak, ona bir kimlik kartı (Token) teslim etmek ve bu kartın yetki düzeyine (Rollerine) göre API içerisindeki kapılardan geçişini kontrol etmektir.

Konu 9: JWT Authentication: Create Access and Refresh Tokens (JWT Yetkilendirmesi: Erişim ve Yenileme Token'ları Üretimi)

JWT, taraflar arasında güvenli bir şekilde JSON nesnesi olarak veri taşımaya yarayan açık bir standarttır (RFC 7519).

Uygulama: JWT Üretim Altyapısının Kurulması

Gerekli kütüphaneyi ortamımıza dahil edelim:

pip install PyJWT

src/config.py (Güvenlik Ayarları):

import os

class SecurityConfig:
    # Gerçek üretim ortamında bu anahtarlar çok güçlü rastgele dizgiler olmalı ve .env dosyasından okunmalıdır.
    JWT_SECRET = "ust_dizey_gizli_anahtar_987654321"
    JWT_ALGORITHM = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES = 15
    REFRESH_TOKEN_EXPIRE_DAYS = 7

src/auth_utils.py (Token Oluşturma Fonksiyonları):

import datetime
import jwt
from config import SecurityConfig

def create_jwt_token(data: dict, expires_delta: datetime.timedelta) -> str:
    """Genel JWT üreten yardımcı fonksiyon"""
    payload = data.copy()
    expire = datetime.datetime.utcnow() + expires_delta
    
    # 'exp' (expiration time) JWT standartlarında rezerve edilmiş bir anahtardır.
    payload.update({"exp": expire})
    
    encoded_jwt = jwt.encode(
        payload, 
        SecurityConfig.JWT_SECRET, 
        algorithm=SecurityConfig.JWT_ALGORITHM
    )
    return encoded_jwt

def generate_tokens(user_id: int) -> dict:
    """Kullanıcı için hem Access hem de Refresh Token üretir"""
    access_token_expires = datetime.timedelta(minutes=SecurityConfig.ACCESS_TOKEN_EXPIRE_MINUTES)
    refresh_token_expires = datetime.timedelta(days=SecurityConfig.REFRESH_TOKEN_EXPIRE_DAYS)
    
    access_token = create_jwt_token(
        data={"sub": str(user_id), "token_type": "access"}, 
        expires_delta=access_token_expires
    )
    
    refresh_token = create_jwt_token(
        data={"sub": str(user_id), "token_type": "refresh"}, 
        expires_delta=refresh_token_expires
    )
    
    return {
        "access_token": access_token,
        "refresh_token": refresh_token,
        "token_type": "bearer"
    }

Konu 10: JWT Authentication: Protect Endpoints with HTTP Bearer Auth (HTTP Bearer Auth ile Uç Noktaları Korumak)

Bir istemci korumalı bir API uç noktasına istek attığında, token bilgisini HTTP başlığında (Authorization: Bearer <token>) göndermek zorundadır. FastAPI, bu başlığı yakalamak için yerleşik HTTPBearer güvenlik şemasını sunar.

Uygulama: Koruyucu Bağımlılık (Dependency) Yazımı

src/auth_utils.py dosyasına eklenecek doğrulama katmanı:

import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from config import SecurityConfig

# FastAPI'nin HTTPBearer şemasını somutlaştırıyoruz
security_scheme = HTTPBearer()

async def get_current_user_id(credentials: HTTPAuthorizationCredentials = Depends(security_scheme)) -> int:
    """
    Gelen istekteki Bearer token'ı yakalar, çözer ve doğrular.
    Başarılıysa token içindeki kullanıcı ID'sini (sub) döner.
    """
    token = credentials.credentials
    try:
        # Token şifresini çöz ve geçerliliğini (ve süresini) kontrol et
        payload = jwt.decode(
            token, 
            SecurityConfig.JWT_SECRET, 
            algorithms=[SecurityConfig.JWT_ALGORITHM]
        )
        
        # Token tipinin 'access' olduğundan emin oluyoruz
        if payload.get("token_type") != "access":
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Geçersiz token türü. Lütfen Access Token kullanın."
            )
            
        user_id = payload.get("sub")
        if user_id is None:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Token içeriği geçersiz (sub bulunamadı)."
            )
        return int(user_id)
        
    except jwt.ExpiredSignatureError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token süresi dolmuş. Lütfen token'ı yenileyin."
        )
    except jwt.PyJWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Geçersiz kimlik doğrulama token'ı."
        )

Korumalı Uç Nokta Örneği (src/routers/profil.py):

from fastapi import APIRouter, Depends
from auth_utils import get_current_user_id

router = APIRouter(prefix="/profil", tags=["Kullanıcı Profili"])

# Bu endpoint'e erişmek isteyen biri geçerli token sunmak zorundadır
@router.get("/me")
async def profil_detayi(current_user_id: int = Depends(get_current_user_id)):
    return {
        "mesaj": "Gizli profile erişim sağlandı.",
        "kullanici_id": current_user_id
    }

Konu 11: JWT Authentication: Renew User Access Using Refresh Tokens (Refresh Token ile Erişimi Yenileme)

Access token'ın süresi bittiğinde istemci (örneğin mobil uygulama) arka planda /auth/refresh uç noktasına elindeki Refresh Tokenı göndererek sessizce yeni bir Access Token talep eder.

Uygulama: Token Yenileme API'sı

src/routers/auth.py dosyasına eklenecek yenileme uç noktası:

import jwt
from fastapi import APIRouter, HTTPException, status, Body
from auth_utils import create_jwt_token, SecurityConfig
import datetime

@router.post("/refresh")
async def refresh_access_token(refresh_token: str = Body(..., embed=True)):
    try:
        # Refresh token'ı çözüyoruz
        payload = jwt.decode(
            refresh_token, 
            SecurityConfig.JWT_SECRET, 
            algorithms=[SecurityConfig.JWT_ALGORITHM]
        )
        
        if payload.get("token_type") != "refresh":
            raise HTTPException(status_code=400, detail="Bu bir refresh token değil.")
            
        user_id = payload.get("sub")
        
        # Sadece yeni bir kısa ömürlü Access Token üretiyoruz
        access_token_expires = datetime.timedelta(minutes=SecurityConfig.ACCESS_TOKEN_EXPIRE_MINUTES)
        new_access_token = create_jwt_token(
            data={"sub": user_id, "token_type": "access"},
            expires_delta=access_token_expires
        )
        
        return {
            "access_token": new_access_token,
            "token_type": "bearer"
        }
        
    except jwt.PyJWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED, 
            detail="Geçersiz veya süresi dolmuş yenileme token'ı."
        )

Konu 12: JWT Authentication: Revoke Access Tokens Using Redis (Redis ile Token İptali / Oturumu Kapatma)

JWT'ler doğası gereği "stateless" (durumsuz) yapılardır; sunucu veritabanına bakmadan token'ın doğruluğunu anlar. Ancak kullanıcı "Çıkış Yap" (Logout) dediğinde, o token süresi dolana kadar hala geçerlidir. Bu güvenlik açığını kapatmak için iptal edilen token'ları (Blocklist) çok hızlı okuma/yazma yapan Redis üzerinde saklarız.

Uygulama: Redis Entegrasyonu ve Token Blocklist Kontrolü

Önce Redis kütüphanesini kuralım:

pip install redis

src/redis_client.py (Redis Bağlantısı):

import redis

# Redis sunucusuna bağlanıyoruz (Varsayılan port 6379)
redis_db = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

def block_token(jti: str, expire_seconds: int):
    """Token'ın benzersiz kimliğini (jti veya token ham halini) Redis'e ekler"""
    redis_db.setex(name=jti, time=expire_seconds, value="true")

def is_token_blocked(jti: str) -> bool:
    """Token Redis blocklist'inde var mı kontrol eder"""
    return redis_db.exists(jti) == 1

src/auth_utils.py içerisindeki get_current_user_id fonksiyonuna eklenecek Redis Kontrolü:

# get_current_user_id fonksiyonunun içine, jwt.decode işleminden HEMEN SONRA eklenmelidir:
# Basitlik açısından token dizgesinin kendisini anahtar olarak kontrol edelim:

if is_token_blocked(token):
    raise HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Bu token iptal edilmiş (Kullanıcı çıkış yaptı)."
    )

Kullanıcı Çıkış Yapma Uç Noktası (src/routers/auth.py):

@router.post("/logout")
async def logout(credentials: HTTPAuthorizationCredentials = Depends(security_scheme)):
    token = credentials.credentials
    # Token'ı 15 dakika boyunca (Access token ömrü kadar) Redis'te karantinaya alıyoruz
    block_token(jti=token, expire_seconds=15 * 60)
    return {"mesaj": "Başarıyla çıkış yapıldı, token iptal edildi."}

Konu 13: Role-Based Access Control Using Dependency Injection (Rol Tabanlı Erişim Kontrolü - RBAC)

Sistemdeki bazı uç noktalara sadece belirli yetkideki kişilerin (Örn: admin veya instructor) erişmesi istenir. Bunu sağlamak için FastAPI'nin dinamik sınıf tabanlı bağımlılık yapısını (Callable Dependencies) kullanırız.

1. Model Güncellemesi

Öncelikle kullanıcı modelimize rol alanı eklememiz gerekir.

src/models.py (Genişletilmiş User Tablosu):

# User sınıfına yeni bir alan ekliyoruz (Alembic ile veritabanına yansıtılmalı)
class User(SQLModel, table=True):
    # ... eski alanlar (id, email, password_hash)
    role: str = Field(default="user") # Seçenekler: user, instructor, admin

2. RBAC Bağımlılık Sınıfının Tasarımı

Her endpoint için ayrı fonksiyon yazmak yerine, izin verilen rolleri dinamik olarak kabul eden bir sınıf tasarlıyoruz.

src/rbac.py (Rol Kontrol Mekanizması):

from fastapi import Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from db import get_session
from models import User
from auth_utils import get_current_user_id

class RoleChecker:
    def __init__(self, allowed_roles: list[str]):
        # İzin verilen rolleri sınıf başlatılırken alıyoruz
        self.allowed_roles = allowed_roles

    async def __call__(self, user_id: int = Depends(get_current_user_id), session: AsyncSession = Depends(get_session)):
        """İstek geldiğinde FastAPI bu metodu tetikler"""
        # Veritabanından kullanıcının güncel rolünü sorgula
        user = await session.get(User, user_id)
        
        if not user or user.role not in self.allowed_roles:
            raise HTTPException(
                status_code=status.HTTP_03_FORBIDDEN, # 403 Forbidden
                detail="Bu işlemi gerçekleştirmek için yetkiniz yetersiz."
            )
        return user # Yetki tamamsa kullanıcı nesnesini teslim et

3. Uygulama: Rol Korumalı Endpoint Kullanımı

src/routers/kurslar.py (Sadece Eğitmen ve Adminlerin Kurs Ekleyebilmesi):

from fastapi import APIRouter, Depends
from rbac import RoleChecker
from models import Kurs, User

router = APIRouter(prefix="/kurslar", tags=["Kurs Yönetimi"])

# Sadece admin ve instructor rollerine izin veren denetleyici nesnesi
allow_management = RoleChecker(["admin", "instructor"])

@router.post("/ekle")
async def kurs_ekle(yeni_kurs: Kurs, current_admin: User = Depends(allow_management)):
    # Eğer istek atan kullanıcı admin veya instructor değilse yukarıdaki Depends adımında elenecektir.
    return {
        "mesaj": "Yetki doğrulandı, kurs başarıyla sisteme eklendi.",
        "ekleyen_yetkili": current_admin.email
    }



Veri tabanımızın kalbini oluşturacak ve tabloları birbirine akıllı bağlarla bağlayacağımız Databases Relationships & Advanced SQLModel (Part 14 ve Part 15) bölümüne geçiyoruz.

Bu bölümde, ilişkisel veri tabanlarının en önemli konusu olan İlişki Tanımlamalarını (Relationships) ele alacağız. Bir eğitmenin birden fazla kursa sahip olduğu Bire-Çok (One-to-Many) ilişki modelini kuracak, SQLModel kütüphanesinin en güçlü özelliklerinden olan Relationship nesnesini kullanacak ve verilerin birbirini sonsuz döngüye sokmasını engelleyen gelişmiş İçiçe Pydantic Şemaları (Advanced Schemas) tasarlayacağız.

Bölüm 5: Databases Relationships & Advanced SQLModel

SQLModel kullanırken ilişkileri iki aşamalı düşünmek gerekir:

  1. Foreign Key (Yabancı Anahtar): SQL düzeyinde tabloları birbirine bağlayan gerçek veri tabanı sütunudur.

  2. Relationship Property (İlişki Özelliği): Python/ORM düzeyinde, sanki nesnenin bir özelliğiymiş gibi ilişkili diğer nesnelere (kod yazarken kolayca) erişmemizi sağlayan sanal bağlantıdır.

Konu 14: Model and Schema Relationships: One-to-Many SQLModel (Bire-Çok İlişkilerin Modellenmesi)

Senaryomuzda bir Eğitmen (User) birden fazla Kurs açabilir. Yani User ve Kurs arasında Bire-Çok (One-to-Many) bir ilişki mevcuttur.

Uygulama: SQLModel ile İlişkili Modellerin Tanımlanması

src/models.py (Güncellenmiş Modeller):

from typing import List, Optional
from sqlmodel import Field, Relationship, SQLModel

class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    email: str = Field(unique=True, index=True, nullable=False)
    password_hash: str = Field(nullable=False)
    role: str = Field(default="user")
    
    # ORM İlişkisi: Bir kullanıcının açtığı tüm kurslara 'user.kurslar' diyerek erişebileceğiz.
    # back_populates, Kurs modelindeki 'egitmen' özelliğine işaret eder.
    kurslar: List["Kurs"] = Relationship(back_populates="egitmen")


class Kurs(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    baslik: str = Field(index=True)
    fiyat: float
    
    # SQL Düzeyinde Foreign Key (Yabancı Anahtar) Tanımı
    # Bu alan, kursun hangi eğitmene ait olduğunun ID bilgisini tutar.
    egitmen_id: int = Field(foreign_key="user.id", nullable=False)
    
    # ORM İlişkisi: Bir kursun eğitmen bilgilerine 'kurs.egitmen' diyerek doğrudan erişebileceğiz.
    egitmen: User = Relationship(back_populates="kurslar")

Konu 15: More Database and Schema Relationships (Gelişmiş Şemalar ve Döngü Engelleme)

İlişkili modelleri doğrudan API yanıtı (Response) olarak döndürmeye çalıştığımızda çok büyük bir sorunla karşılaşırız: Sonsuz Döngü (Infinite Recursion Loop).
Kurs nesnesi Eğitmen nesnesini yüklemeye çalışır $\rightarrow$ Eğitmen nesnesi içindeki Kurslar listesini yüklemeye çalışır $\rightarrow$ O kurslar tekrar Eğitmeni yükler... Bu döngü sistemin çökmesine neden olur.

Bu sorunu çözmek ve temiz bir API çıktısı sunmak için Gelişmiş İçiçe Pydantic Şemaları (Nested Schemas) tasarlamamız gerekir.

Uygulama: Gelişmiş Şemaların Tasarımı

src/schemas.py (Döngü Engelleyici Temiz Şemalar):

from typing import List
from pydantic import BaseModel, EmailStr

# --- TEMEL ŞEMALAR (İlişkisiz, Yalın Veriler) ---

class UserBaseSchema(BaseModel):
    id: int
    email: EmailStr
    role: str

class KursBaseSchema(BaseModel):
    id: int
    baslik: str
    fiyat: float
    egitmen_id: int

# --- GELİŞMİŞ İLİŞKİLİ ŞEMALAR (İçiçe Yapılar) ---

class KursDetayResponseSchema(KursBaseSchema):
    """Bir kurs çağrıldığında, yanında eğitmeninin bilgisini de döner"""
    egitmen: UserBaseSchema # İçiçe pydantic şeması kullanımı

class EgitmenDetayResponseSchema(UserBaseSchema):
    """Bir eğitmen çağrıldığında, açtığı tüm kursların listesini de döner"""
    kurslar: List[KursBaseSchema]

Uygulama: İlişkili Verileri Veri Tabanından Çekme ve API Entegrasyonu

İlişkili verileri asenkron veri tabanından çekerken SQLAlchemy varsayılan olarak ilişkili tabloları otomatik yüklemez (Lazy Loading aktiftir ve asenkron mimaride hata verir). Bu yüzden sorgu esnasında ilişkili tabloları agresif bir şekilde yüklememiz (Eager Loading / selectinload) gerekir.

src/routers/kurslar.py (İlişkili Veri Getiren API):

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from sqlalchemy.orm import selectinload
from db import get_session
from models import Kurs
from schemas import KursDetayResponseSchema

router = APIRouter(prefix="/kurslar", tags=["Kurs İlişkileri"])

@router.get("/{kurs_id}", response_model=KursDetayResponseSchema)
async def ilişkili_kurs_detayi_getir(kurs_id: int, session: AsyncSession = Depends(get_session)):
    # selectinload kullanarak Kurs verisini çekerken, ona bağlı olan 'egitmen' verisini de tek seferde çekiyoruz.
    sorgu = select(Kurs).where(Kurs.id == kurs_id).options(selectinload(Kurs.egitmen))
    sonuc = await session.execute(sorgu)
    kurs = sonuc.scalars().first()
    
    if not kurs:
        raise HTTPException(status_code=404, detail="Aranan kurs bulunamadı.")
        
    return kurs # KursDetayResponseSchema şemasına tam uyumlu ve güvenli bir JSON çıktısı üretilir.

Üretilecek Örnek API JSON Çıktısı (Sonsuz döngüden uzak, temiz çıktı):

{
  "id": 10,
  "baslik": "FastAPI İleri Seviye",
  "fiyat": 299.99,
  "egitmen_id": 3,
  "egitmen": {
    "id": 3,
    "email": "ahmet@egitim.com",
    "role": "instructor"
  }
}



Uygulamamızın kararlılığını, güvenliğini ve izlenebilirliğini (observability) en üst seviyeye çıkaracağımız Error Handling & Middleware (Part 16 ve Part 17) bölümüne geçiyoruz.

Bu bölümde, API boyunca standart bir hata formatı yakalamak için Özel API İstisnaları (Custom API Exceptions) geliştirecek, gelen her isteği ve oluşan hataları takip etmek amacıyla bir Custom Logging (Özel Günlükleme) mekanizması kuracak ve son olarak web uygulamalarının (React, Vue, Angular vb.) API'ımıza güvenli bir şekilde erişebilmesi için hayati önem taşıyan CORS (Cross-Origin Resource Sharing) ayarlarını yapılandıracağız.

Bölüm 6: Error Handling & Middleware

Bir API'ın profesyonelliği, sadece başarılı senaryolarda değil, hata anlarında istemciye ne kadar açıklayıcı ve standart bir yanıt döndüğüyle de ölçülür. Middleware (Ara Yazılım) katmanı ise isteklerin ve yanıtların arasına girerek küresel operasyonlar yürütmemizi sağlar.

Konu 16: Error Handling: Custom API Exceptions (Özel API İstisnaları)

FastAPI'nin yerleşik HTTPException sınıfı iş görse de, kurumsal projelerde tüm hataların istemciye aynı JSON şablonuyla dönmesi istenir. Bunun için kendi özel hata sınıflarımızı türetir ve bunları FastAPI'nin küresel Exception Handler (İstisna Yakalayıcı) mekanizmasına kaydederiz.

Uygulama: Özel Hata Mimarisi ve Yakalayıcı Yönetimi

src/exceptions.py (Özel İstisna Sınıfları ve Handler Tanımı):

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

# Tüm özel iş mantığı hatalarımız için bir taban sınıf oluşturuyoruz
class ProjeTabanHata(Exception):
    def __init__(self, mesaj: str, durum_kodu: int = 400):
        self.mesaj = mesaj
        self.durum_kodu = durum_kodu

# Taban sınıftan türeyen spesifik hata sınıfları
class KaynakBulunamadiHatasi(ProjeTabanHata):
    def __init__(self, mesaj: str = "Aranan kaynak bulunamadı."):
        super().__init__(mesaj, durum_kodu=404)

class YetersizBakiyeHatasi(ProjeTabanHata):
    def __init__(self, mesaj: str = "Bu işlem için bakiyeniz yetersiz."):
        super().__init__(mesaj, durum_kodu=400)


def kaydet_ozel_hata_yakalayicilar(app: FastAPI):
    """Yazdığımız özel hataları FastAPI'ye küresel olarak tanıtan fonksiyon"""
    
    @app.exception_handler(ProjeTabanHata)
    async def proje_hata_yakalayici(request: Request, exc: ProjeTabanHata):
        # İstemciye dönecek standart hata şablonunu tasarlıyoruz
        return JSONResponse(
            status_code=exc.durum_kodu,
            content={
                "basarili": False,
                "hata": {
                    "mesaj": exc.mesaj,
                    "kod": exc.__class__.__name__,
                    "zaman_damgasi": datetime.datetime.utcnow().isoformat()
                }
            }
        )

Uygulama İçi Kullanım Örneği (src/routers/kurslar.py):

from exceptions import KaynakBulunamadiHatasi

@router.get("/detay/{kurs_id}")
async def kurs_kontrol_et(kurs_id: int):
    # Simülasyon: Kurs bulunamadı varsayalım
    kurs_var_mi = False
    
    if not kurs_var_mi:
        # Kendi yazdığımız özel hatayı fırlatıyoruz
        raise KaynakBulunamadiHatasi(f"{kurs_id} ID'li kurs sistemde mevcut değil.")

İstemciye Dönen Standart JSON Çıktısı:

{
  "basarili": false,
  "hata": {
    "mesaj": "45 ID'li kurs sistemde mevcut değil.",
    "kod": "KaynakBulunamadiHatasi",
    "zaman_damgasi": "2026-08-29T15:23:00"
  }
}

Konu 17: Middleware: Custom Logging and CORS (Özel Günlükleme ve CORS Yapılandırması)

1. Custom Logging Middleware (Özel Günlükleme Ara Yazılımı)

API'ımıza gelen her HTTP isteğinin hangi adrese geldiğini, ne kadar sürede tamamlandığını ve hangi HTTP durum koduyla döndüğünü log dosyalarına veya terminale basmak, üretim ortamındaki hataları ayıklamak (debug) için kritiktir.

src/middleware.py (Zamanlama ve Loglama Ara Yazılımı):

import time
import logging
from fastapi import Request

# Python yerleşik logging mekanizmasını yapılandırıyoruz
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger("API_Gozlemci")

async def log_istek_sureleri_middleware(request: Request, call_next):
    """Gelen isteklerin çalışma süresini hesaplayan ve loglayan ara yazılım"""
    baslangic_zamani = time.time()
    
    # İsteğin yolunu ve metodunu yakalıyoruz
    istek_yolu = request.url.path
    istek_metodu = request.method
    
    # İsteği bir sonraki aşamaya (route handler'a) gönderiyoruz ve yanıtı alıyoruz
    yanit = await call_next(request)
    
    gecen_sure = time.time() - baslangic_zamani
    
    # Bilgileri terminale/log dosyasına yazıyoruz
    logger.info(
        f"Metod: {istek_metodu} | Yol: {istek_yolu} | Durum: {yanit.status_code} | Süre: {gecen_sure:.4f} sn"
    )
    
    # İstemciye özel bir performans başlığı (header) da ekleyebiliriz
    yanit.headers["X-Islem-Suresi"] = str(gecen_sure)
    
    return yanit

2. CORS (Cross-Origin Resource Sharing) Yapılandırması

Tarayıcılar, güvenlik nedeniyle bir web sitesinin (Örn: http://localhost:3000 adresindeki bir React uygulamasının) farklı bir kökteki API'a (Örn: http://projem.com) doğrudan istek atmasını engeller. API'ımızın güvenli dış kaynaklara kapılarını açması için CORS ayarlarını yapmalıyız.

Uygulama: Main Dosyasında Tüm Parçaların Birleştirilmesi

src/main.py (Entegrasyon):

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from exceptions import kaydet_ozel_hata_yakalayicilar
from middleware import log_istek_sureleri_middleware

app = FastAPI()

# --- 1. CORS AYARLARI ---
# API'ımıza erişmesine izin verdiğimiz dış kökenlerin (origin) listesi
origins = [
    "http://localhost:3000",      # Yerel React/Vue geliştirme sunucusu
    "https://projem.com",     # Canlıdaki ön yüz (frontend) domaini
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,            # İzin verilen adres listesi
    allow_credentials=True,           # Çerezlerin (Cookies) taşınmasına izin ver
    allow_methods=["*"],              # Tüm HTTP metodlarına (GET, POST, PUT, DELETE vb.) izin ver
    allow_headers=["*"],              # Tüm HTTP başlıklarına izin ver
)

# --- 2. CUSTOM LOGGING MIDDLEWARE ENTEGRASYONU ---
app.middleware("http")(log_istek_sureleri_middleware)

# --- 3. CUSTOM ERROR HANDLER ENTEGRASYONU ---
kaydet_ozel_hata_yakalayicilar(app)

@app.get("/test")
async def test_endpoint():
    return {"durum": "Her şey yolunda, log mekanizması devrede!"}

Bu yapılandırma tamamlandığında, terminalinizde her istekten sonra şu şekilde profesyonel log satırları akmaya başlayacaktır:

2026-08-29 15:23:45,123 - INFO - Metod: GET | Yol: /test | Durum: 200 | Süre: 0.0024 sn
2026-08-29 15:24:02,541 - INFO - Metod: GET | Yol: /kurslar/detay/45 | Durum: 404 | Süre: 0.0081 sn



Güvenli bir web uygulamasının en kritik kullanıcı döngülerini oluşturan Security Features & User Flows (Part 18 ve Part 19) bölümüne geçiyoruz.

Bu bölümde, sahte hesap açılışlarını engellemek için E-posta ile Hesap Doğrulama (Email Verification) sistemini inşa edecek ve şifresini unutan kullanıcılar için güvenli, süreli ve tek kullanımlık token tabanlı Şifre Sıfırlama (Password Reset) akışını tasarlayacağız.

Bölüm 7: Security Features & User Flows

Kullanıcı kayıt akışlarında e-posta doğrulaması ve şifre sıfırlama süreçleri tasarlanırken güvenlik açığı yaratmamak adına şu kurallara dikkat edilmelidir:

  1. Süreli Token Kullanımı: E-postaya giden onaylama ve şifre sıfırlama linkleri sonsuza kadar geçerli olmamalıdır (Örn: 24 saat veya 15 dakika geçerlilik).

  2. Güvenli Arka Plan İletişimi: E-posta gönderme işlemi zaman alan bir iştir. Bu işlemler (müfredatın sonraki bölümünde göreceğimiz Celery entegrasyonu öncesi) kullanıcıyı arayüzde bekletmemek adına asenkron veya arka plan görevi (Background Tasks) olarak planlanmalıdır.

Konu 18: User Account Verification via Email (E-posta ile Hesap Doğrulama)

Kullanıcı ilk kayıt olduğunda sistem durumu is_active = False olarak işaretlenir. Kullanıcının e-posta adresine benzersiz, kriptografik bir doğrulama token'ı gönderilir. Kullanıcı bu linke tıkladığında hesabı aktif hale getirilir.

Uygulama: E-posta Doğrulama Akışının Kurulması

Gerekli e-posta kütüphanesini kuralım:

pip install fastapi-mail

src/models.py (Kullanıcı Tablosu Doğrulama Güncellemesi):

from sqlmodel import Field, SQLModel
from typing import Optional

class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    email: str = Field(unique=True, index=True)
    password_hash: str
    is_active: bool = Field(default=False) # İlk kayıt anında hesap pasiftir!

src/auth_utils.py (Doğrulama Token'ı Üretme ve Çözme):

import jwt
import datetime
from config import SecurityConfig

def create_verification_token(email: str) -> str:
    """E-posta doğrulaması için 24 saat geçerli özel token üretir"""
    expire = datetime.datetime.utcnow() + datetime.timedelta(hours=24)
    payload = {"sub": email, "action": "email_verification", "exp": expire}
    return jwt.encode(payload, SecurityConfig.JWT_SECRET, algorithm=SecurityConfig.JWT_ALGORITHM)

src/routers/auth.py (Kayıt ve Onaylama API Uç Noktaları):

from fastapi import APIRouter, Depends, HTTPException, status, BackgroundTasks
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from db import get_session
from models import User
from schemas import UserCreateSchema
from security import hash_password
from auth_utils import create_verification_token, SecurityConfig
import jwt

router = APIRouter(prefix="/auth", tags=["Hesap Doğrulama"])

# Sahte e-posta gönderme fonksiyonu (Simülasyon)
def send_simulation_email(email: str, token: str):
    verification_link = f"https://projem.com{token}"
    print(f"\n[E-POSTA SİMÜLASYONU] Kime: {email}")
    print(f"Lütfen hesabınızı onaylamak için tıklayın: {verification_link}\n")

@router.post("/register")
async def register_user(user_data: UserCreateSchema, background_tasks: BackgroundTasks, session: AsyncSession = Depends(get_session)):
    # Email kontrolü
    sorgu = select(User).where(User.email == user_data.email)
    sonuc = await session.execute(sorgu)
    if sonuc.scalars().first():
        raise HTTPException(status_code=400, detail="E-posta zaten kayıtlı.")
    
    yeni_kullanici = User(email=user_data.email, password_hash=hash_password(user_data.password), is_active=False)
    session.add(yeni_kullanici)
    await session.commit()
    
    # Doğrulama token'ı oluşturma
    token = create_verification_token(user_data.email)
    
    # E-posta gönderme işlemini arka plan görevi olarak ekliyoruz (API'ı yavaşlatmaz)
    background_tasks.add_task(send_simulation_email, user_data.email, token)
    
    return {"mesaj": "Kayıt başarılı. Lütfen e-posta adresinize gönderilen onay linkine tıklayın."}

@router.get("/verify-email")
async def verify_email(token: str, session: AsyncSession = Depends(get_session)):
    try:
        payload = jwt.decode(token, SecurityConfig.JWT_SECRET, algorithms=[SecurityConfig.JWT_ALGORITHM])
        
        if payload.get("action") != "email_verification":
            raise HTTPException(status_code=400, detail="Geçersiz işlem türü.")
            
        email = payload.get("sub")
        
        # Kullanıcıyı bul ve aktifleştir
        sorgu = select(User).where(User.email == email)
        sonuc = await session.execute(sorgu)
        user = sonuc.scalars().first()
        
        if not user:
            raise HTTPException(status_code=404, detail="Kullanıcı bulunamadı.")
            
        if user.is_active:
            return {"mesaj": "Hesap zaten daha önce doğrulanmış."}
            
        user.is_active = True # Kullanıcı artık sistemi kullanabilir
        await session.commit()
        
        return {"mesaj": "E-posta adresiniz başarıyla doğrulandı! Giriş yapabilirsiniz."}
        
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=400, detail="Doğrulama linkinin süresi dolmuş (24 saat).")
    except jwt.PyJWTError:
        raise HTTPException(status_code=400, detail="Geçersiz doğrulama token'ı.")

Konu 19: Handling Password Resets (Şifre Sıfırlama Süreçlerinin Yönetimi)

Kullanıcı şifresini unuttuğunda, eski şifreyi ona göstermek kesinlikle yasaktır (Zaten bcrypt ile tek yönlü hash'lendiği için geri döndürülemez). Güvenli akışta kullanıcı e-postasını girer, ona 15 dakika ömürlü bir sıfırlama token'ı gönderilir. Kullanıcı bu token ile yeni şifresini belirler.

Uygulama: Güvenli Şifre Sıfırlama Akışı

src/auth_utils.py (Şifre Sıfırlama Token'ı):

def create_password_reset_token(email: str) -> str:
    """Şifre sıfırlama için sadece 15 dakika geçerli çok sıkı bir token üretir"""
    expire = datetime.datetime.utcnow() + datetime.timedelta(minutes=15)
    payload = {"sub": email, "action": "password_reset", "exp": expire}
    return jwt.encode(payload, SecurityConfig.JWT_SECRET, algorithm=SecurityConfig.JWT_ALGORITHM)

src/routers/auth.py (Şifre Unuttum ve Yeni Şifre Kaydetme API'ları):

from pydantic import BaseModel, EmailStr
from security import hash_password

class PasswordResetRequestSchema(BaseModel):
    email: EmailStr

class PasswordResetConfirmSchema(BaseModel):
    token: str
    new_password: str

@router.post("/password-reset-request")
async def password_reset_request(data: PasswordResetRequestSchema, background_tasks: BackgroundTasks, session: AsyncSession = Depends(get_session)):
    sorgu = select(User).where(User.email == data.email)
    sonuc = await session.execute(sorgu)
    user = sonuc.scalars().first()
    
    # GÜVENLİK İPUCU: Kullanıcı yoksa bile "E-posta kayıtlı değil" demiyoruz. 
    # Saldırganların sistemde hangi e-postaların kayıtlı olduğunu anlamasını engellemek için her durumda aynı cevabı dönüyoruz.
    if user:
        token = create_password_reset_token(user.email)
        # Gerçek senaryoda e-posta gönderilir
        background_tasks.add_task(print, f"[ŞİFRE SIFIRLAMA] Link: https://projem.com{token}")
        
    return {"mesaj": "Eğer e-posta adresi sistemimizde kayıtlıysa, şifre sıfırlama talimatları gönderilecektir."}


@router.post("/password-reset-confirm")
async def password_reset_confirm(data: PasswordResetConfirmSchema, session: AsyncSession = Depends(get_session)):
    try:
        payload = jwt.decode(data.token, SecurityConfig.JWT_SECRET, algorithms=[SecurityConfig.JWT_ALGORITHM])
        
        if payload.get("action") != "password_reset":
            raise HTTPException(status_code=400, detail="Geçersiz şifre sıfırlama işlemi.")
            
        email = payload.get("sub")
        
        sorgu = select(User).where(User.email == email)
        sonuc = await session.execute(sorgu)
        user = sonuc.scalars().first()
        
        if not user:
            raise HTTPException(status_code=404, detail="Kullanıcı bulunamadı.")
            
        # Yeni şifreyi güvenli bir şekilde hash'leyip güncelliyoruz
        user.password_hash = hash_password(data.new_password)
        await session.commit()
        
        return {"mesaj": "Şifreniz başarıyla güncellendi. Yeni şifrenizle giriş yapabilirsiniz."}
        
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=400, detail="Şifre sıfırlama linkinin süresi dolmuş (15 dakika).")
    except jwt.PyJWTError:
        raise HTTPException(status_code=400, detail="Geçersiz şifre sıfırlama token'ı.")



Sistem mimarimizin performansını ve ölçeklenebilirliğini zirveye çıkaracak olan, ağır ve uzun süren süreçleri ana uygulama akışından koparıp kuyruklara teslim edeceğimiz Background Tasks & Asynchronous Jobs (Part 20) bölümüne geçiyoruz.

Bu bölümde, FastAPI içindeki yerleşik BackgroundTasks yapısı ile endüstri standardı olan dağıtık görev kuyruğu Celery ve mesaj aracısı (message broker) Redis mimarisi arasındaki farkları ele alacak, tam teşekküllü bir Celery işçisi (worker) tasarlayacak ve asenkron görev yönetimini kodlayacağız.

Bölüm 8: Background Tasks & Asynchronous Jobs

Bir web isteği sırasında (Request-Response yaşam döngüsünde) kullanıcının tarayıcıda veya mobil uygulamada bekletilme süresi ideal olarak 200 milisaniyenin altında olmalıdır. Ancak büyük raporlar üretmek, PDF dosyaları oluşturmak, resimleri optimize etmek veya toplu e-postalar göndermek saniyeler, hatta dakikalar alabilir. Bu tür "ağır" iş yükleri ana sunucu iş parçacığını (thread) bloke etmemek için arka plana devredilir.

Konu 20: Background Tasks with FastAPI and Celery + Redis (Celery ve Redis ile Arka Plan Görevleri)

FastAPI'nin yerleşik bir BackgroundTasks mekanizması vardır. Ancak bu mekanizma aynı Python sürecinde (process) çalışır. Eğer sunucu çökerse kuyruktaki görevler kaybolur ve CPU'yu yoğun kullanan işlerde ana uygulamayı yavaşlatabilir.

Celery ise tamamen bağımsız süreçlerde çalışan, görevleri kalıcı olarak Redis veya RabbitMQ gibi kuyruklarda saklayan ve binlerce görevi paralel işleyebilen profesyonel bir asenkron görev kuyruğudur.

Mimari Tasarım Şeması

[İstemci] ──(İstek atar)──> [FastAPI] ──(Görevi Tetikler)──> [Redis Kuyruğu]
                                                                  │
[İstemci] <──(İş ID'si döner)─ [FastAPI]                           │ (Görevi Çeker)
                                                                  ▼
                                                          [Celery Worker] (Ağır İşi Yapar)

Uygulama: Celery ve Redis Altyapısının Kurulması

Öncelikle gerekli kütüphaneleri projemize ekleyelim:

pip install celery redis

src/celery_app.py (Celery Yapılandırma Dosyası):

from celery import Celery

# Celery uygulamasını somutlaştırıyoruz. 
# broker: Görevlerin birikeceği kuyruk (Redis)
# backend: Tamamlanan görevlerin sonuçlarının saklanacağı yer (Redis)
celery_app = Celery(
    "fastapi_tasks",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/0"
)

# Celery ayarlarını yapılandırıyoruz
celery_app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    timezone="Europe/Istanbul",
    enable_utc=True,
)

# Ağır işi yapacak olan örnek bir Celery görevi (Task) tanımlıyoruz
@celery_app.task(name="tasks.agir_rapor_olustur_task")
def agir_rapor_olustur_task(kullanici_email: str, rapor_tipi: str):
    """
    Bu fonksiyon FastAPI sürecinin tamamen DIŞINDA, 
    bağımsız bir Celery Worker sürecinde çalışacaktır.
    """
    import time
    print(f"[Celery] {kullanici_email} için {rapor_tipi} raporu hazırlanmaya başlandı...")
    
    # 10 saniye süren ağır bir veritabanı veya dosya işlemi simülasyonu
    time.sleep(10) 
    
    print(f"[Celery] Rapor hazırlandı ve {kullanici_email} adresine mail atıldı.")
    return {"durum": "Başarılı", "rapor_tipi": rapor_tipi, "sure_sn": 10}

Uygulama: FastAPI Endpoint Katmanında Görevi Tetikleme

src/routers/raporlar.py (Rapor Tetikleme ve Durum Kontrol API'ları):

from fastapi import APIRouter, status
from celery.result import AsyncResult
from celery_app import agir_rapor_olustur_task

router = APIRouter(prefix="/raporlar", tags=["Arka Plan Görevleri (Celery)"])

@router.post("/tetikle", status_code=status.HTTP_202_ACCEPTED)
async def rapor_uretimini_baslat(email: str, tip: str):
    # Görevi senkron çağırmıyoruz ( agir_rapor_olustur_task(email, tip) ŞEKLİNDE DEĞİL! )
    # .delay() kullanarak görevi anında Redis kuyruğuna fırlatıyoruz.
    gorev = agir_rapor_olustur_task.delay(email, tip)
    
    # İstemciye 10 saniye beklemeden ANINDA 202 (Kabul Edildi) durum koduyla iş id'sini dönüyoruz.
    return {
        "mesaj": "Rapor üretim görevi arka plana atıldı.",
        "gorev_id": gorev.id,
        "durum_kontrol_adresi": f"/raporlar/durum/{gorev.id}"
    }

@router.get("/durum/{gorev_id}")
async def rapor_durumu_sorgula(gorev_id: str):
    # Redis backend üzerinden görevin güncel durumunu sorguluyoruz
    gorev_sonucu = AsyncResult(gorev_id)
    
    return {
        "gorev_id": gorev_id,
        "durum": gorev_sonucu.state, # PENDING, STARTED, SUCCESS, FAILURE
        "sonuc": gorev_sonucu.result if gorev_sonucu.ready() else None
    }

Sistem Nasıl Çalıştırılır? (Terminal Operasyonları)

Bu mimarinin çalışabilmesi için terminalinizde 3 ayrı sürecin eş zamanlı açık olması gerekir:

# 1. Terminal: Redis Sunucusunun Başlatılması (Eğer bilgisayarınızda kuruluysa)
redis-server

# 2. Terminal: FastAPI Sunucusunun Başlatılması
uvicorn main:app --reload

# 3. Terminal: Celery İşçisinin (Worker) Başlatılması (Proje kök dizininde)
celery -A celery_app.celery_app worker --loglevel=info

Artık /raporlar/tetikle uç noktasına bir istek attığınızda, FastAPI sunucunuz milisaniyeler içinde yanıt verecek, Celery Worker terminalinizde ise 10 saniye boyunca işlemin yapıldığını ve başarıyla tamamlandığını gözlemleyebileceksiniz.



API'larımızın kalitesini, dökümantasyon standartlarını ve güvenilirliğini en üst seviyeye çıkaracağımız Documentation & Testing (Part 21 ve Part 22) bölümüne geçiyoruz.

Bu bölümde, FastAPI'nin sunduğu otomatik dökümantasyon araçlarını (Swagger UI ve Redoc) profesyonelce özelleştirecek, ardından projenin asenkron katmanlarını Pytest kullanarak test edecek, dış bağımlılıkları (veritabanı, dış API'lar) Unittest Mock ile izole edecek ve son olarak mülkiyet tabanlı (property-based) test aracı Schemathesis ile API'ımızı otomatik olarak hatalara karşı tarayacağız.

Bölüm 9: Documentation & Testing

Bir API'ın başarısı, geliştiriciler tarafından ne kadar kolay anlaşıldığı (Dökümantasyon) ve kod değişikliklerinde sistemin kırılıp kırılmadığı (Test Otomasyonu) ile doğrudan ilişkilidir.

Konu 21: API Documentation with Swagger and Redoc (Swagger ve Redoc ile Profesyonel Dökümantasyon)

FastAPI, OpenAPI standartlarını kullanarak dökümantasyonu otomatik üretir. Ancak kurumsal bir görünüm için başlıkları, açıklamaları, sürümleri ve güvenlik şemalarını özelleştirmemiz gerekir.

Uygulama: FastAPI Uygulamasının Dökümantasyon İçin Özelleştirilmesi

src/main.py (Gelişmiş OpenAPI Ayarları):

from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi

app = FastAPI(
    title="Akademi REST API Portalı",
    description="JWT, SQLModel ve Celery entegrasyonuna sahip, yüksek performanslı asenkron eğitim yönetim API dökümantasyonu.",
    version="1.4.2",
    docs_url="/gelistirici-docs",  # Varsayılan /docs adresini güvenlik için değiştiriyoruz
    redoc_url="/mimari-docs"        # Varsayılan /redoc adresini değiştiriyoruz
)

def ozel_openapi_semasi():
    # Şema zaten bir kere üretilmişse tekrar hesaplamamak için önbellekten (cache) döner
    if app.openapi_schema:
        return app.openapi_schema
    
    openapi_schema = get_openapi(
        title=app.title,
        version=app.version,
        description=app.description,
        routes=app.routes,
    )
    
    # Şemaya kurumsal logo veya ekstra meta veriler ekleme
    openapi_schema["info"]["x-logo"] = {
        "url": "https://firmamiz.com"
    }
    
    app.openapi_schema = openapi_schema
    return app.openapi_schema

# FastAPI'nin varsayılan şema üreticisini kendi fonksiyonumuzla değiştiriyoruz
app.openapi = ozel_openapi_semasi

Konu 22: API Testing with Unittest Mock, Pytest & Schemathesis (Pytest, Mock ve Schemathesis ile Test Otomasyonu)

Asenkron FastAPI projelerinde test yazarken, gerçek veritabanını kirletmemek için bellek içi (In-Memory) bir test veritabanı kurmak veya dış servis çağrılarını taklit etmek (Mocking) gerekir.

1. Pytest ve Asenkron İstemci Kurulumu

Gerekli test kütüphanelerini kuralım:

pip install pytest pytest-asyncio httpx

src/tests/conftest.py (Test Bağımlılıkları ve Yapılandırma Dosyası):

import pytest
from httpx import AsyncClient
from sqlmodel import SQLModel, create_engine
from sqlmodel.pool import StaticPool
from src.main import app
from src.db import get_session

# Testler için bellekte (In-Memory) çalışacak sahte bir SQLite veritabanı motoru oluşturuyoruz
TEST_DATABASE_URL = "sqlite:///:memory:"

@pytest.fixture(name="test_session")
def test_session_fixture():
    engine = create_engine(
        TEST_DATABASE_URL, 
        connect_args={"check_same_thread": False}, 
        poolclass=StaticPool
    )
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session
    SQLModel.metadata.drop_all(engine)

@pytest.fixture(name="client")
def client_fixture(test_session):
    # Ana uygulamadaki gerçek 'get_session' bağımlılığını, test veritabanı oturumumuzla değiştiriyoruz
    def override_get_session():
        yield test_session
        
    app.dependency_overrides[get_session] = override_get_session
    
    # Testler için senkron/asenkron istek atabilecek HTTP istemcisini teslim ediyoruz
    with TestClient(app) as c:
        yield c
    
    # Test bitince override işlemini temizliyoruz
    app.dependency_overrides.clear()

2. Uygulama: Pytest ve Mock Kullanarak Bir Uç Noktayı Test Etme

src/tests/test_kurslar.py (Kurs Ekleme ve Sahte Servis Testi):

from unittest.mock import patch

def test_yeni_kurs_ekleme_basarili(client):
    # Girdi verimizi hazırlıyoruz
    kurs_verisi = {"baslik": "Sıfırdan İleri Seviye Go Dili", "fiyat": 149.99, "egitmen_id": 1}
    
    # HTTP POST isteği atıyoruz
    response = client.post("/kurslar/", json=kurs_verisi)
    
    # Doğrulama (Assertion) adımları
    assert response.status_code == 201
    data = response.json()
    assert data["baslik"] == "Sıfırdan İleri Seviye Go Dili"
    assert "id" in data


def test_dis_servis_entegrasyonu_ile_kurs_fiyati(client):
    """
    Dışarıdaki bir döviz kuru API'sına bağımlı olan bir endpoint'i,
    dış API'ya gerçekten istek atmadan Mock (taklit) kullanarak test ediyoruz.
    """
    # 'src.routers.kurslar.doviz_kuru_getir' fonksiyonunun her çağrıldığında sabit 34.5 dönmesini sağlıyoruz
    with patch("src.routers.kurslar.doviz_kuru_getir") as mock_doviz:
        mock_doviz.return_value = 34.5
        
        response = client.get("/kurslar/fiyat-tl/1")
        assert response.status_code == 200
        # Mock sayesinde dış API çökmüş olsa bile testimiz kararlı çalışır

3. Schemathesis ile API'ı Hatalara Karşı Taramak (Property-Based Testing)

Schemathesis, API dökümantasyonunuzu (OpenAPI şemasını) okur ve uç noktalarınıza binlerce kombinasyonda rastgele, hatalı, sınırları zorlayan (Örn: metin beklenen yere sayı, aşırı uzun dizgiler) veriler göndererek sisteminizi çökertmeye çalışır. API'ınızın beklenmedik durumlarda 500 (Internal Server Error) hatası vermesini engellemek için mükemmel bir araçtır.

Kurulum ve Çalıştırma:

pip install schemathesis

# API sunucunuz uvicorn ile localhost:8000 üzerinde çalışırken şu komutu tetikleyin:
schemathesis run http://localhost:8000/openapi.json

Schemathesis, tüm API yollarınızı otomatik olarak bulur ve veri doğrulama açıklarınızı tarayıp raporlar.



Geliştirdiğimiz tüm bu katmanları (FastAPI, PostgreSQL, Redis ve Celery Worker) bir araya getirip production ortamına, yani canlı sunuculara taşımaya hazır hale getireceğimiz Final Testing & Deployment (Part 23) bölümüne geçiyoruz.

Bu bölümde, projenin canlı ortam ayarlarını yönetmek için Gelişmiş Çevre Değişkenleri (Environment Variables) mimarisini kuracak, üretim ortamında (production) güvenliği ve performansı optimize etmek için gereken Uvicorn/Gunicorn yapılandırmasını yapacak ve son olarak tüm bu ekosistemi tek bir komutla ayağa kaldırabileceğimiz endüstri standardı Docker & Docker Compose mimarisini tasarlayacağız.

Bölüm 10: Final Testing & Deployment

Bir uygulamayı production ortamına alırken geliştirme ortamından (development) farklı olarak en çok dikkat edilmesi gereken unsurlar şunlardır:

  1. Gizlilik ve Güvenlik: Şifreler, veritabanı adresleri ve gizli anahtarlar asla kaynak kodun içinde (main.py veya models.py) düz metin olarak yer almamalıdır. Çevre değişkenleri (.env) kullanılmalıdır.

  2. Süreç Yönetimi (Process Management): uvicorn main:app --reload komutu geliştirme amaçlıdır. Canlı ortamda çoklu iş parçacığı (multi-worker) yönetimi yapabilen, sunucu çöktüğünde uygulamayı otomatik yeniden başlatan yapılar kurulmalıdır.

Konu 23: Deploying FastAPI, PostgreSQL, Celery & Redis (Canlı Dağıtım ve Production Yapılandırması)

1. Gelişmiş Çevre Ayarları (Pydantic-Settings Entegrasyonu)

Canlı ortam yapılandırmalarını yönetmek için en kararlı yol pydantic-settings kütüphanesini kullanmaktır.

pip install pydantic-settings

src/config.py (Production Ayar Yönetimi):

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    # Varsayılan değerler atanır, ancak .env dosyasında ne yazıyorsa o geçerli olur
    PROJECT_NAME: str = "Akademi API Production"
    SECRET_KEY: str
    ALGORITHM: str = "HS256"
    
    # Veritabanı ve Redis bağlantı adresleri
    DATABASE_URL: str
    REDIS_URL: str = "redis://redis:6379/0"

    # .env dosyasını otomatik okuması için Pydantic yapılandırması
    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")

# Tek bir settings nesnesi üzerinden tüm uygulamada ayarlara erişeceğiz
settings = Settings()

Canlı Ortam İçin Kök Dizinde Bulunacak .env Dosyası:

SECRET_KEY=9a8b7c6d5e4f3210zyxwvutsrqponmlk_canli_anahtar
DATABASE_URL=postgresql+asyncpg://db_user:guvenli_sifre@postgres_db:5432/canli_akademi_db
REDIS_URL=redis://redis_service:6379/0

2. Docker ile Konteynerleştirme (Containerization)

Tüm sistem bileşenlerinin (FastAPI, Celery, Postgres, Redis) farklı sunucularda veya aynı sunucu içinde birbirini etkilemeden izole çalışabilmesi için Docker imajları hazırlarız.

Proje Kök Dizini İçin Dockerfile:

# 1. Aşama: Python taban imajını seçiyoruz
FROM python:3.11-slim

# 2. Aşama: Çalışma dizinini ayarlıyoruz
WORKDIR /app

# 3. Aşama: Python'un logları anlık ekrana basmasını sağlıyoruz
ENV PYTHONUNBUFFERED=1

# 4. Aşama: Bağımlılıkları kopyalayıp yüklüyoruz
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 5. Aşama: Tüm proje kodlarını konteyner içine kopyalıyoruz
COPY . .

# 6. Aşama: Varsayılan olarak API'ı ayağa kaldıracak komut (Celery için bu geçersiz kılınacak)
CMD ["gunicorn", "src.main:app", "-w", "4", "-k", "uvicorn.workers.UvicornWorker", "-b", "0.0.0.0:8000"]

Mimarinin Açıklaması: Canlı ortamda gunicorn orkestrasyon aracını öne koyup arkasına 4 adet asenkron uvicorn işçisi (UvicornWorker) bağlıyoruz. Bu sayede sunucu, gelen binlerce eş zamanlı isteği (concurrent requests) işlemcideki çekirdeklere dağıtarak bloke olmadan yanıtlayabilir.

3. Çoklu Servis Yönetimi: Docker Compose Mimarisi

Tüm mikro servis bileşenlerini tek bir orkestrasyon dosyasıyla ayağa kaldırmak için projenin kök dizininde docker-compose.yml dosyası tasarlarız.

docker-compose.yml:

version: '3.8'

services:
  # 1. Servis: PostgreSQL Veritabanı
  postgres_db:
    image: postgres:15-alpine
    container_name: production_postgres
    environment:
      POSTGRES_USER: db_user
      POSTGRES_PASSWORD: guvenli_sifre
      POSTGRES_DB: canli_akademi_db
    volumes:
      - postgres_data:/var/lib/postgresql/data
    ports:
      - "5432:5432"

  # 2. Servis: Mesaj Aracısı ve İptal Listesi (Redis)
  redis_service:
    image: redis:7-alpine
    container_name: production_redis
    ports:
      - "6379:6379"

  # 3. Servis: FastAPI Web API Uygulamamız
  web_api:
    build: .
    container_name: production_fastapi
    ports:
      - "8000:8000"
    env_file:
      - .env
    depends_on:
      - postgres_db
      - redis_service

  # 4. Servis: Ağır İşleri Sırtlanacak Celery İşçisi
  celery_worker:
    build: .
    container_name: production_celery
    # Dockerfile içindeki varsayılan Gunicorn komutunu ezerek Celery Worker'ı başlatıyoruz
    command: celery -A src.celery_app.celery_app worker --loglevel=info
    env_file:
      - .env
    depends_on:
      - redis_service
      - postgres_db

volumes:
  postgres_data:

Üretim Ortamında Canlıya Alım (Deployment) Komutu:

Sunucunuza (AWS, DigitalOcean, Azure vb.) Docker ve Docker Compose yükledikten sonra projenizin olduğu dizinde tek bir komut çalıştırmanız yeterlidir:

docker-compose up --build -d

Açıklama: -d (detached) parametresi tüm bu devasa ekosistemin arka planda (arka plan süreci olarak) sessizce çalışmasını sağlar. Artık sunucunuzun IP adresinin 8000 portuna gelen istekler Gunicorn/Uvicorn tarafından karşılanacak, veri tabanı Postgres'e yazılacak, kuyruklar Redis üzerinden bağımsız Celery konteynerine aktarılacaktır.

Konu: FastAPI with MongoDB & Beanie (NoSQL Alternatif Ekstra Bölüm)

Kurs müfredatının son bölümünde SQL veritabanlarına güçlü bir alternatif olarak döküman tabanlı MongoDB ve onun için geliştirilmiş Pydantic uyumlu modern asenkron ODM (Object Document Mapper) kütüphanesi olan Beanie yer almaktadır. SQLModel ile yaptığımız tüm doğrulamalar ve Pydantic entegrasyonu, Beanie sayesinde MongoDB'de de birebir korunur.

Uygulama: NoSQL Geliştirme Temelleri

pip install beanie motor

(Not: motor, MongoDB için asenkron Python sürücüsüdür.)

src/nosql_models.py (Beanie Modeli):

from beanie import Document
from pydantic import Field

# Beanie Document sınıfından türetilen modeller doğrudan MongoDB koleksiyonu (collection) olur
class SistemLogDocument(Document):
    islem_adi: str
    kullanici: str
    detay: str
    etki_seviyesi: int = Field(default=1)

    class Settings:
        name = "sistem_loglari" # MongoDB'deki koleksiyon adı

src/main.py dosyasına eklenecek MongoDB Lifespan Başlatıcısı:

from motor.motor_asyncio import AsyncIOMotorClient
from beanie import init_beanie
from nosql_models import SistemLogDocument

# Lifespan içine eklenecek asenkron MongoDB başlatma adımı:
async def init_nosql():
    client = AsyncIOMotorClient("mongodb://localhost:27017")
    # Beanie'yi bağlayıp modellerimizi MongoDB'ye tanıtıyoruz
    await init_beanie(database=client.akademi_nosql_db, document_models=[SistemLogDocument])



Projemizi modern bulut mühendisliği standartlarına ulaştıracak, manuel müdahaleleri sıfırlayacak CI/CD (Sürekli Entegrasyon / Sürekli Dağıtım) boru hattı (pipeline) mimarisine ve canlı sistemlerin takibi için hayati önem taşıyan Docker Log İzleme operasyonlarına geçiyoruz.

Bölüm 11: Bulut Sunucusuna Geçiş (CI/CD) & Docker Log Yönetimi

Bir uygulamayı her güncellemede sunucuya bağlanıp manuel olarak git pull ve docker-compose up yaparak canlıya almak; hem insan hatasına çok açıktır hem de sistem kesintilerine (downtime) yol açabilir. Sektör standardı, kod GitHub'a pushlandığı anda testlerin otomatik çalışması, imajın üretilmesi ve sunucuya otomatik yüklenmesidir.

1. GitHub Actions ile CI/CD Entegrasyonu

Bu senaryoda, kodlarımızı barındırdığımız GitHub platformunun yerleşik otomasyon aracı olan GitHub Actions'ı kullanacağız. Amacımız, ana daldaki (main branch) her kod güncellemesinde projenin otomatik olarak bulut sunucumuza (Örn: AWS EC2, DigitalOcean Damlacığı, Ubuntu Sunucusu) aktarılmasıdır.

Uygulama: GitHub Actions Workflow Yapılandırması

Projenizin kök dizininde .github/workflows/deploy.yml adında bir dosya oluşturun:

name: FastAPI Canlıya Alım (Deployment) Boru Hattı

# Sadece main dalına (branch) bir kod pushlandığında çalışır
on:
  push:
    branches:
      - main

jobs:
  test_ve_deploy:
    runs-on: ubuntu-latest

    steps:
      # 1. Adım: GitHub üzerindeki kodu otomasyon sunucusuna çeker
      - name: Kodu Kopyala (Checkout)
        uses: actions/checkout@v4

      # 2. Adım: Python ortamını hazırlar ve Pytest testlerini koşturur
      - name: Python Ortamı Kur ve Test Et
        uses: actions/setup-python@v5
        with:
          python-node: '3.11'
      - name: Bağımlılıkları Yükle ve Testleri Çalıştır
        run: |
          pip install -r requirements.txt
          pytest src/tests/

      # 3. Adım: Testler başarılıysa, SSH üzerinden bulut sunucumuza bağlanır ve komutları çalıştırır
      - name: Bulut Sunucusuna Bağlan ve Güncelle (Deploy)
        uses: appleboy/ssh-action@v1.0.3
        with:
          host: ${{ secrets.SUNUCU_IP_ADRESI }}      # Sunucunuzun IP adresi
          username: ${{ secrets.SUNUCU_KULLANICI }}  # Örn: ubuntu, root
          key: ${{ secrets.SUNUCU_SSH_KEY }}        # Sunucuya erişim sağlayan Private SSH Key
          script: |
            cd /var/www/fastapi-akademi
            git pull origin main
            # Sadece değişen servisleri arka planda sessizce yeniden inşa edip kaldırır
            docker-compose up --build -d web_api celery_worker

Güvenlik Notu: Yukarıdaki secrets. ile başlayan alanlar kesinlikle koda açık yazılmaz. GitHub projenizin ayarlarından Settings -> Secrets and variables -> Actions kısmına gidilerek güvenli bir şekilde sunucu bilgileri olarak tanımlanır.

2. Canlı Sunucuda Docker Loglarının İzlenmesi (Logging Operations)

Uygulama canlıya alındıktan sonra arka planda ne olduğunu anlamanın, hataları (500 hataları, veritabanı kopmaları, Celery çöküşleri) yakalamanın tek yolu Docker loglarını doğru filtrelemektir. Sunucu terminaline SSH ile bağlandıktan sonra şu pratik komut şablonlarını kullanırız:

A. Canlı Akışı İzleme (Real-time Tail)

Herhangi bir servisin loglarını, sanki bir video akışı gibi canlı olarak ekrandan izlemek için -f (follow) parametresi kullanılır.

# Sadece FastAPI web sunucusunun loglarını canlı takip etme
docker logs -f production_fastapi

# Celery işçisinin arka planda hangi görevleri yaptığını anlık izleme
docker logs -f production_celery

B. Sadece Son Satırları Sınırlama (Tail Limitation)

Eğer konteyner aylardır çalışıyorsa, doğrudan docker logs demek ekrana milyonlarca satır basabilir ve terminali kilitleyebilir. --tail parametresi ile sadece son gelişmeleri okuruz.

# FastAPI'nin en son ürettiği 100 log satırını getir
docker logs --tail 100 production_fastapi

C. Zaman Damgası ile İzleme (Timestamps)

Kod loglarınızın içinde tarih bilgisi yazmıyorsa, Docker'ın kendi zaman damgalarını ekrana yazdırmasını sağlayabilirsiniz.

docker logs -t --tail 50 production_fastapi

D. Gelişmiş Filtreleme (Grep Kombinasyonları)

Log yığınının içinden sadece "Hata" veya "Kritik" durumları ayıklamak için Linux'un güçlü aracı grep ile Docker loglarını birleştiririz.

# FastAPI logları içinde sadece HTTP 500 (Internal Server Error) veren satırları filtrele
docker logs production_fastapi 2>&1 | grep "500"

# Celery logları içinde hataya (ERROR / FAILURE) neden olan görevleri bulma
docker logs production_celery 2>&1 | grep -i "error"

(Not: 2>&1 ifadesi, Docker'ın standart hata çıktılarını (stderr) standart çıktılara (stdout) yönlendirerek grep komutunun hatasız arama yapmasını sağlar.)