FastAPI ile Kişisel API Oluşturma

Modern veri bilimi ve yazılım geliştirme dünyasında, projelerinizi API’ler aracılığıyla kullanılabilir hale getirmek giderek daha önemli hale geliyor. Bu makale, Python tabanlı FastAPI çerçevesini kullanarak kişisel projeleriniz için nasıl hızlı, güvenilir ve kullanımı kolay bir API oluşturabileceğinizi adım adım anlatıyor.

FastAPI Nedir ve Neden Kullanmalıyız?

FastAPI, Python 3.6+ için geliştirilmiş modern, hızlı ve verimli bir web çerçevesidir. Adından da anlaşılacağı gibi, hızlı API geliştirmeye odaklanmıştır ve birçok özelliği otomatik olarak sağlar.

FastAPI’nin Avantajları:

1. Hız: Adından da anlaşılacağı üzere, performans açısından oldukça hızlıdır. Node.js ve Go ile yarışabilecek düzeyde hızlı çalışır.
2. Otomatik Dokümantasyon: Swagger UI ve ReDoc ile otomatik interaktif API dokümantasyonu oluşturur.
3. Veri Doğrulama: Pydantic sayesinde güçlü veri doğrulama mekanizması sunar.
4. Modern Python Özellikleri: Type hints, async/await gibi modern Python özelliklerini tam olarak destekler.
5. Kullanım Kolaylığı: Django veya Flask kadar kullanımı kolay, ancak performansı çok daha yüksektir.

Gerekli Ortamın Hazırlanması

FastAPI ile çalışmaya başlamadan önce gerekli ortamı hazırlayalım. İdeal olarak, yeni bir sanal ortam oluşturup gereksinimleri bu ortama kurmanızı öneririm.

# Sanal ortam oluşturma
python -m venv fastapi-env

# Sanal ortamı aktifleştirme (Windows)
fastapi-env\Scripts\activate

# Sanal ortamı aktifleştirme (Linux/MacOS)
source fastapi-env/bin/activate

# Gerekli paketleri kurma
pip install fastapi uvicorn sqlalchemy

Burada:

• fastapi: Ana çerçeve
• uvicorn: ASGI sunucusu (FastAPI uygulamanızı çalıştırmak için)
• sqlalchemy: Veritabanı işlemleri için ORM (İlerleyen bölümlerde kullanacağız)

Basit Bir API Oluşturma

Hemen basit bir API oluşturarak başlayalım. Aşağıdaki kodu main.py dosyasına kaydedelim:

from fastapi import FastAPI

# FastAPI uygulaması oluşturma
app = FastAPI(
    title="Kişisel Veri API",
    description="Veri projeleriniz için kişisel API",
    version="0.1.0"
)

# Kök endpoint (root endpoint)
@app.get("/")
async def root():
    return {"message": "Kişisel Veri API'ye Hoş Geldiniz!"}

# Basit bir veri endpoint'i
@app.get("/veriler")
async def get_data():
    # Gerçek bir projede burada veritabanından veri çekilebilir
    return {
        "veriler": [
            {"id": 1, "isim": "Ali", "değer": 42},
            {"id": 2, "isim": "Ayşe", "değer": 38},
            {"id": 3, "isim": "Mehmet", "değer": 55}
        ]
    }

Şimdi API’mizi çalıştıralım:

uvicorn main:app --reload

Tarayıcımızdan http://127.0.0.1:8000 adresine giderek API’mizin çalıştığını görebiliriz. Ayrıca http://127.0.0.1:8000/docs adresinden otomatik oluşturulan Swagger dokümantasyonunu inceleyebiliriz.

Veri Modelleri ve Doğrulama

FastAPI’nin en güçlü yanlarından biri, Pydantic kullanarak veri doğrulama yapabilmesidir. Aşağıda, veri modellerini tanımlayarak API’mizi geliştirelim:

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

app = FastAPI(
    title="Kişisel Veri API",
    description="Veri projeleriniz için kişisel API",
    version="0.1.0"
)

# Veri modeli tanımlama
class VeriÖğesi(BaseModel):
    id: int
    isim: str
    değer: float
    açıklama: Optional[str] = None
    
    class Config:
        schema_extra = {
            "example": {
                "id": 1,
                "isim": "Örnek Veri",
                "değer": 42.5,
                "açıklama": "Bu bir örnek veri öğesidir"
            }
        }

# Veri oluşturma modeli (ID otomatik oluşturulacak)
class VeriOluştur(BaseModel):
    isim: str = Field(..., min_length=2, max_length=50)
    değer: float = Field(..., gt=0)
    açıklama: Optional[str] = Field(None, max_length=200)

# Örnek veri listesi (gerçek projede veritabanı kullanılacak)
veri_listesi = [
    {"id": 1, "isim": "Ali", "değer": 42.0, "açıklama": "İlk veri"},
    {"id": 2, "isim": "Ayşe", "değer": 38.5, "açıklama": None},
    {"id": 3, "isim": "Mehmet", "değer": 55.2, "açıklama": "Üçüncü veri"}
]

@app.get("/")
async def root():
    return {"message": "Kişisel Veri API'ye Hoş Geldiniz!"}

@app.get("/veriler", response_model=List[VeriÖğesi])
async def get_all_data():
    return veri_listesi

@app.get("/veriler/{veri_id}", response_model=VeriÖğesi)
async def get_data_by_id(veri_id: int):
    for veri in veri_listesi:
        if veri["id"] == veri_id:
            return veri
    raise HTTPException(status_code=404, detail=f"ID:{veri_id} olan veri bulunamadı")

@app.post("/veriler", response_model=VeriÖğesi, status_code=201)
async def create_data(veri: VeriOluştur):
    # Yeni ID oluştur (gerçek uygulamada veritabanı tarafından yapılır)
    yeni_id = max([v["id"] for v in veri_listesi]) + 1
    
    # Yeni veri öğesi oluştur
    yeni_veri = {
        "id": yeni_id,
        "isim": veri.isim,
        "değer": veri.değer,
        "açıklama": veri.açıklama
    }
    
    # Listeye ekle
    veri_listesi.append(yeni_veri)
    return yeni_veri

Bu kodda:

• VeriÖğesi ve VeriOluştur adında iki Pydantic model tanımladık
• Tüm verileri listeleme, ID’ye göre tek veri getirme ve yeni veri oluşturma endpoint’leri ekledik
• Veri doğrulama için alan kısıtlamaları ekledik (minimum uzunluk, maksimum uzunluk, vb.)
• HTTP durum kodlarını ve hata işlemeyi ekledik

FastAPI ile API Dokümantasyonu Otomatik Oluşturma

FastAPI’nin en güzel özelliklerinden biri, API dokümantasyonunu otomatik olarak oluşturmasıdır. Uygulamanızı çalıştırdıktan sonra aşağıdaki URL’leri ziyaret ederek dokümantasyonu görebilirsiniz:

• Swagger UI: http://127.0.0.1:8000/docs
• ReDoc: http://127.0.0.1:8000/redoc

Bu dokümantasyon sayfaları, API’nizi test etmenize ve diğer geliştiricilerin kolayca anlamasına olanak tanır.

Veri Tabanı Entegrasyonu

Gerçek dünya uygulamalarında, verileri bellekte tutmak yerine bir veritabanında saklamak isteriz. SQLAlchemy ORM kullanarak veritabanı entegrasyonu ekleyelim:

from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy import create_engine, Column, Integer, String, Float, Text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from pydantic import BaseModel, Field
from typing import List, Optional

# Veritabanı bağlantısı
SQLALCHEMY_DATABASE_URL = "sqlite:///./veriler.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# SQLAlchemy modeli
class VeriTablosu(Base):
    __tablename__ = "veriler"
    
    id = Column(Integer, primary_key=True, index=True)
    isim = Column(String(50), index=True)
    değer = Column(Float)
    açıklama = Column(Text, nullable=True)

# Veritabanını oluştur
Base.metadata.create_all(bind=engine)

# Pydantic modelleri
class VeriÖğesi(BaseModel):
    id: int
    isim: str
    değer: float
    açıklama: Optional[str] = None
    
    class Config:
        orm_mode = True

class VeriOluştur(BaseModel):
    isim: str = Field(..., min_length=2, max_length=50)
    değer: float = Field(..., gt=0)
    açıklama: Optional[str] = Field(None, max_length=200)

# FastAPI uygulaması
app = FastAPI(
    title="Kişisel Veri API",
    description="SQLite veritabanı ile veri projeleriniz için kişisel API",
    version="0.2.0"
)

# Bağımlılık - Veritabanı oturumu
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/")
async def root():
    return {"message": "Kişisel Veri API'ye Hoş Geldiniz!"}

@app.get("/veriler", response_model=List[VeriÖğesi])
def get_all_data(db: Session = Depends(get_db)):
    return db.query(VeriTablosu).all()

@app.get("/veriler/{veri_id}", response_model=VeriÖğesi)
def get_data_by_id(veri_id: int, db: Session = Depends(get_db)):
    veri = db.query(VeriTablosu).filter(VeriTablosu.id == veri_id).first()
    if veri is None:
        raise HTTPException(status_code=404, detail=f"ID:{veri_id} olan veri bulunamadı")
    return veri

@app.post("/veriler", response_model=VeriÖğesi, status_code=201)
def create_data(veri: VeriOluştur, db: Session = Depends(get_db)):
    db_veri = VeriTablosu(
        isim=veri.isim,
        değer=veri.değer,
        açıklama=veri.açıklama
    )
    db.add(db_veri)
    db.commit()
    db.refresh(db_veri)
    return db_veri

@app.put("/veriler/{veri_id}", response_model=VeriÖğesi)
def update_data(veri_id: int, veri: VeriOluştur, db: Session = Depends(get_db)):
    db_veri = db.query(VeriTablosu).filter(VeriTablosu.id == veri_id).first()
    if db_veri is None:
        raise HTTPException(status_code=404, detail=f"ID:{veri_id} olan veri bulunamadı")
    
    # Verileri güncelle
    db_veri.isim = veri.isim
    db_veri.değer = veri.değer
    db_veri.açıklama = veri.açıklama
    
    db.commit()
    db.refresh(db_veri)
    return db_veri

@app.delete("/veriler/{veri_id}", status_code=204)
def delete_data(veri_id: int, db: Session = Depends(get_db)):
    db_veri = db.query(VeriTablosu).filter(VeriTablosu.id == veri_id).first()
    if db_veri is None:
        raise HTTPException(status_code=404, detail=f"ID:{veri_id} olan veri bulunamadı")
    
    db.delete(db_veri)
    db.commit()
    return None

Bu kod:

• SQLite veritabanı kullanarak verileri kalıcı olarak saklıyor
• ORM modelleri ve Pydantic modelleri arasında dönüşüm yapıyor
• CRUD (Create, Read, Update, Delete) işlemlerinin tamamını içeriyor
• Bağımlılık enjeksiyonu ile veritabanı oturumlarını yönetiyor

API’nizi Dağıtma ve Yayınlama

API’nizi geliştirdikten sonra, onu internette yayınlamak isteyeceksiniz. İşte bazı yaygın dağıtım seçenekleri:

1. Docker Kullanarak Dağıtım

Öncelikle bir Dockerfile oluşturalım:

FROM python:3.9

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Ve bir requirements.txt dosyası:

fastapi>=0.68.0
pydantic>=1.8.0
uvicorn>=0.15.0
sqlalchemy>=1.4.23

Şimdi Docker imajını oluşturup çalıştırabiliriz:

# İmajı oluştur
docker build -t kisisel-api .

# Container'ı çalıştır
docker run -d -p 8000:8000 kisisel-api

2. Heroku’ya Dağıtım

Heroku’ya dağıtım için bir Procfile dosyası oluşturun:

web: uvicorn main:app --host=0.0.0.0 --port=${PORT:-8000}

Sonra Heroku CLI ile dağıtın:

heroku create kisisel-api
git push heroku main

3. AWS, Google Cloud veya Azure’a Dağıtım

Bu platformlar daha fazla ölçeklenebilirlik ve özellik sunabilir. Her bulut platformu farklı dağıtım yöntemleri sunar:

• AWS için AWS Elastic Beanstalk veya AWS Lambda + API Gateway
• Google Cloud için Google App Engine veya Cloud Run
• Azure için Azure App Service

İleri Seviye FastAPI Özellikleri

FastAPI, basit API’lerin ötesinde birçok gelişmiş özellik sunar. İşte bazı ileri seviye özellikleri:

1. API Kimlik Doğrulama

Kimlik doğrulama ve yetkilendirme ekleyelim:

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel

# JWT için gizli anahtar
SECRET_KEY = "gercek_bir_projede_bu_anahtari_guvenli_bir_sekilde_saklayin"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# Şifre hashing
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# OAuth2 ile token doğrulama
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# Kullanıcı modeli
class Kullanıcı(BaseModel):
    kullanıcı_adı: str
    email: str
    tam_ad: str
    aktif: bool = True

# Token modeli
class Token(BaseModel):
    access_token: str
    token_type: str

# Token veri modeli
class TokenData(BaseModel):
    kullanıcı_adı: str = None

# Örnek kullanıcı veritabanı (gerçek uygulamada bir DB kullanın)
fake_users_db = {
    "ahmet": {
        "kullanıcı_adı": "ahmet",
        "tam_ad": "Ahmet Yılmaz",
        "email": "ahmet@example.com",
        "hashed_password": pwd_context.hash("gizlisifre"),
        "aktif": True,
    }
}

def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return user_dict

def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        return False
    if not verify_password(password, user["hashed_password"]):
        return False
    return user

def create_access_token(data: dict, expires_delta: timedelta = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Geçersiz kimlik bilgileri",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(kullanıcı_adı=username)
    except JWTError:
        raise credentials_exception
    user = get_user(fake_users_db, username=token_data.kullanıcı_adı)
    if user is None:
        raise credentials_exception
    return user

async def get_current_active_user(current_user = Depends(get_current_user)):
    if not current_user["aktif"]:
        raise HTTPException(status_code=400, detail="Pasif kullanıcı")
    return current_user

app = FastAPI()

@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Kullanıcı adı veya şifre yanlış",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user["kullanıcı_adı"]}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/users/me")
async def read_users_me(current_user = Depends(get_current_active_user)):
    return {
        "kullanıcı_adı": current_user["kullanıcı_adı"],
        "email": current_user["email"],
        "tam_ad": current_user["tam_ad"]
    }

2. FastAPI Dosya Yükleme

FastAPI ile dosya yükleme işlemi yapalım:

from fastapi import FastAPI, File, UploadFile
import shutil
from pathlib import Path

app = FastAPI()

@app.post("/dosya-yukle/")
async def create_upload_file(dosya: UploadFile = File(...)):
    # Dosya adı ve uzantısı
    dosya_adı = dosya.filename
    
    # Dosya yolunu oluştur
    dosya_yolu = Path(f"yuklemeler/{dosya_adı}")
    dosya_yolu.parent.mkdir(parents=True, exist_ok=True)
    
    # Dosyayı kaydet
    with dosya_yolu.open("wb") as buffer:
        shutil.copyfileobj(dosya.file, buffer)
    
    return {
        "dosya_adı": dosya_adı,
        "dosya_boyutu": dosya_yolu.stat().st_size,
        "dosya_tipi": dosya.content_type
    }

3. FastAPI Görev Zamanlaması

FastAPI ile düzenli çalışan görevler oluşturmak için fastapi-utils paketini kullanabiliriz:

from fastapi import FastAPI
from fastapi_utils.tasks import repeat_every
import time
from datetime import datetime

app = FastAPI()

@app.on_event("startup")
@repeat_every(seconds=60)  # Her dakika çalışır
def periodic_task() -> None:
    # Düzenli çalışacak göreviniz
    now = datetime.now().strftime("%H:%M:%S")
    print(f"Zamanlanmış görev çalışıyor: {now}")
    # Burada veri analizi, veritabanı temizleme gibi görevler yapabilirsiniz

Sonuç

Bu makalede, FastAPI kullanarak nasıl hızlı, güvenli ve ölçeklenebilir bir kişisel API oluşturabileceğinizi öğrendiniz. FastAPI’nin sunduğu modern özellikleri kullanarak, veri projelerinizi kolayca bir API’ye dönüştürebilir ve başkalarıyla paylaşabilirsiniz.

FastAPI, Python dünyasında hızla popülerlik kazanan bir çerçevedir ve hem basit hem de karmaşık projeler için mükemmel bir seçimdir. Veri bilimi, makine öğrenimi ve web uygulamaları geliştirme konularında çalışıyorsanız, FastAPI’yi projelerinize dahil etmeyi düşünmelisiniz.

Daha fazla bilgi için FastAPI resmi dokümantasyonunu inceleyebilirsiniz.

---