FastAPI ve Redis: Yüksek Performanslı Uygulamalar

by
with no comment
Teknoloji

FastAPI ve Redis: Yüksek Performanslı Uygulamalar Geliştirme

Günümüzde web uygulamalarının hızı ve performansı, kullanıcı deneyimini doğrudan etkileyen kritik faktörlerdir. Bu nedenle, geliştiriciler yüksek trafikli ve karmaşık uygulamalar için ölçeklenebilir ve hızlı çözümler aramaktadır. FastAPI, modern web API’leri oluşturmak için tasarlanmış yüksek performanslı bir Python framework’üdür. Redis ise, bellek içi veri yapısı deposu olarak bilinir ve caching, session yönetimi ve gerçek zamanlı analiz gibi çeşitli senaryolarda sıklıkla kullanılır. Bu makalede, FastAPI ve Redis’in nasıl entegre edilebileceğini, bu entegrasyonun avantajlarını ve pratik uygulama örneklerini inceleyeceğiz.

İçindekiler

FastAPI Nedir?

FastAPI, modern web API’leri oluşturmak için kullanılan, yüksek performanslı (Starlette ve Pydantic üzerine inşa edilmiş) bir Python web framework’üdür. Temel özellikleri şunlardır:

  • Hız: Yüksek performans sunar, Go ve NodeJS’e yakın hızlarda çalışır.
  • Kolay Kullanım: Geliştirici dostu bir arayüze sahiptir ve hızlı prototipleme imkanı sunar.
  • Otomatik Dokümantasyon: Swagger UI ve ReDoc gibi araçlarla otomatik API dokümantasyonu oluşturur.
  • Veri Doğrulama: Pydantic ile veri tiplerini tanımlama ve doğrulama imkanı sunar.
  • Asenkron Destek: Asenkron programlama (async/await) ile uyumlu çalışır.

Redis Nedir?

Redis (Remote Dictionary Server), bellek içi veri yapısı deposudur. Verileri RAM üzerinde sakladığı için çok hızlıdır ve çeşitli senaryolarda kullanılabilir:

  • Caching: Sık erişilen verileri önbelleğe alarak uygulama performansını artırır.
  • Session Yönetimi: Kullanıcı oturumlarını saklamak ve yönetmek için kullanılır.
  • Pub/Sub: Gerçek zamanlı mesajlaşma ve olay tabanlı uygulamalar için kullanılır.
  • Leaderboard ve Sıralama: Sıralı kümeler (sorted sets) kullanarak sıralama işlemleri gerçekleştirir.
  • Rate Limiting: API erişimlerini sınırlamak için kullanılır.

FastAPI ve Redis Entegrasyonu Neden Önemli?

FastAPI ve Redis’in birlikte kullanılması, web uygulamalarının performansını önemli ölçüde artırabilir. Özellikle aşağıdaki senaryolarda büyük avantaj sağlar:

  • Yüksek Trafik: Yüksek trafikli uygulamalarda, Redis caching mekanizması sayesinde veritabanına yapılan yükü azaltır.
  • Gerçek Zamanlı Uygulamalar: Redis pub/sub özellikleri ile gerçek zamanlı veri akışı ve mesajlaşma uygulamaları geliştirilebilir.
  • Oturum Yönetimi: Redis, kullanıcı oturumlarını hızlı ve güvenli bir şekilde saklar.
  • API Hız Sınırlama: API’lerin kötüye kullanımını engellemek için Redis ile hız sınırlama uygulanabilir.

Kurulum ve Hazırlık

FastAPI ve Redis entegrasyonu için öncelikle gerekli kütüphaneleri yüklememiz gerekiyor.


pip install fastapi uvicorn redis

Bu komut, FastAPI, Uvicorn (ASGI sunucusu) ve Redis Python kütüphanesini yükleyecektir. Ayrıca, Redis sunucusunun kurulu ve çalışır durumda olduğundan emin olun.

Temel Entegrasyon: Redis Bağlantısı ve Veri Okuma/Yazma

FastAPI uygulamamızda Redis’e bağlanmak ve veri okuma/yazma işlemlerini gerçekleştirmek için aşağıdaki adımları izleyebiliriz:


from fastapi import FastAPI
import redis

app = FastAPI()

# Redis bağlantı bilgileri
REDIS_HOST = "localhost"
REDIS_PORT = 6379

# Redis bağlantısı oluştur
redis_client = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, decode_responses=True)

@app.get("/")
async def read_root():
    # Redis'ten veri oku
    value = redis_client.get("mykey")
    if value:
        return {"message": f"Redis'ten okunan değer: {value}"}
    else:
        return {"message": "Redis'te 'mykey' bulunamadı"}

@app.post("/set/{value}")
async def set_value(value: str):
    # Redis'e veri yaz
    redis_client.set("mykey", value)
    return {"message": f"'mykey' değeri {value} olarak ayarlandı"}

Bu kod örneği, Redis’e bağlanmayı, veri okuma ve yazma işlemlerini göstermektedir. decode_responses=True parametresi, Redis’ten gelen verilerin otomatik olarak string olarak çözümlenmesini sağlar.

Caching Mekanizmaları

Redis’in en yaygın kullanım alanlarından biri caching’dir. FastAPI uygulamalarında, sık erişilen verileri Redis’te önbelleğe alarak veritabanına yapılan yükü azaltabiliriz.


from fastapi import FastAPI, Depends
import redis
from typing import Optional

app = FastAPI()

# Redis bağlantı bilgileri
REDIS_HOST = "localhost"
REDIS_PORT = 6379

# Redis bağlantısı oluştur
redis_client = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, decode_responses=True)

# Örnek veri tabanı fonksiyonu (gerçekte bir veritabanı sorgusu olabilir)
async def get_data_from_database(item_id: int):
    # Burada veritabanından veri çekme işlemini simüle ediyoruz.
    await asyncio.sleep(1) # Veritabanı sorgusu simülasyonu
    return {"id": item_id, "data": f"Veri {item_id}"}

async def get_cached_data(item_id: int, redis_client: redis.Redis = Depends(lambda: redis_client)) -> Optional[dict]:
    # Redis'ten veriyi çek
    cached_data = redis_client.get(f"item:{item_id}")
    if cached_data:
        return json.loads(cached_data)
    return None

async def set_cached_data(item_id: int, data: dict, redis_client: redis.Redis = Depends(lambda: redis_client)):
    # Redis'e veriyi kaydet
    redis_client.set(f"item:{item_id}", json.dumps(data), ex=60)  # 60 saniye boyunca cache'te tut

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    # Önce cache'e bak
    cached_data = await get_cached_data(item_id)
    if cached_data:
        print("Cache'ten okundu")
        return cached_data
    
    # Cache'te yoksa veritabanından çek
    data = await get_data_from_database(item_id)
    
    # Veriyi cache'e kaydet
    await set_cached_data(item_id, data)
    print("Veritabanından okundu")
    return data

Bu örnekte, get_cached_data fonksiyonu ile Redis’ten veri çekiliyor. Eğer veri cache’te bulunamazsa, get_data_from_database fonksiyonu ile veritabanından çekiliyor ve set_cached_data fonksiyonu ile Redis’e kaydediliyor. ex=60 parametresi, verinin 60 saniye boyunca cache’te tutulmasını sağlar.

Session Yönetimi

Redis, kullanıcı oturumlarını saklamak ve yönetmek için de kullanılabilir. FastAPI uygulamalarında, kullanıcı oturumlarını Redis’te saklayarak, oturum bilgilerine hızlı bir şekilde erişebiliriz.

Session yönetimi için genellikle `itsdangerous` veya benzeri bir kütüphane kullanılır. Ancak basit bir örnek için aşağıdaki gibi bir yaklaşım izlenebilir:


from fastapi import FastAPI, Request, Response
from uuid import uuid4
import redis

app = FastAPI()

# Redis bağlantı bilgileri
REDIS_HOST = "localhost"
REDIS_PORT = 6379

# Redis bağlantısı oluştur
redis_client = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, decode_responses=True)

SESSION_COOKIE_NAME = "session_id"
SESSION_EXPIRY = 3600  # Oturumun geçerlilik süresi (saniye)

@app.middleware("http")
async def session_middleware(request: Request, call_next):
    session_id = request.cookies.get(SESSION_COOKIE_NAME)
    
    if not session_id:
        session_id = str(uuid4())
        request.state.new_session = True  # Yeni bir oturum oluşturulduğunu belirt
    else:
        request.state.new_session = False

    request.state.session_id = session_id
    request.state.session = redis_client.get(f"session:{session_id}")

    response = await call_next(request)

    if request.state.new_session:
        response.set_cookie(SESSION_COOKIE_NAME, session_id, max_age=SESSION_EXPIRY, httponly=True)

    return response

@app.get("/session_data")
async def get_session_data(request: Request):
    session_id = request.state.session_id
    session_data = redis_client.get(f"session:{session_id}")
    if session_data:
        return {"session_id": session_id, "data": json.loads(session_data)}
    else:
        return {"session_id": session_id, "data": {}}

@app.post("/set_session_data/{key}/{value}")
async def set_session_data(request: Request, key: str, value: str):
    session_id = request.state.session_id
    session_data = redis_client.get(f"session:{session_id}")

    if session_data:
        session_dict = json.loads(session_data)
    else:
        session_dict = {}

    session_dict[key] = value
    redis_client.setex(f"session:{session_id}", SESSION_EXPIRY, json.dumps(session_dict))
    return {"message": f"'{key}' değeri '{value}' olarak oturuma eklendi"}

Bu örnek, her istekte session ID’sini kontrol eder ve eğer yoksa yeni bir session ID oluşturur. Oturum verileri Redis’te saklanır ve her istekte okunur. Oturum verileri setex fonksiyonu ile belirtilen süre boyunca (SESSION_EXPIRY) Redis’te saklanır.

Rate Limiting (Hız Sınırlama)

API’lerin kötüye kullanımını engellemek için hız sınırlama (rate limiting) önemlidir. Redis, atomik operasyonları sayesinde hız sınırlama için ideal bir çözümdür.


from fastapi import FastAPI, Depends, HTTPException
from fastapi.requests import Request
import redis
import time

app = FastAPI()

# Redis bağlantı bilgileri
REDIS_HOST = "localhost"
REDIS_PORT = 6379

# Redis bağlantısı oluştur
redis_client = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, decode_responses=True)

# Hız sınırlama ayarları
REQUEST_LIMIT = 5  # Belirli bir süre içinde izin verilen istek sayısı
TIME_WINDOW = 60  # Süre penceresi (saniye)

async def rate_limit(request: Request):
    client_ip = request.client.host
    key = f"rate_limit:{client_ip}"

    # Redis'te istek sayısını al
    current_requests = redis_client.get(key)

    if current_requests is None:
        # İlk istek ise, sayacı oluştur ve süreyi ayarla
        redis_client.set(key, 1, ex=TIME_WINDOW)
    else:
        # İsteği artır
        if int(current_requests) >= REQUEST_LIMIT:
            raise HTTPException(status_code=429, detail="Çok fazla istekte bulundunuz. Lütfen daha sonra tekrar deneyin.")
        redis_client.incr(key)

@app.get("/protected_route", dependencies=[Depends(rate_limit)])
async def protected_route():
    return {"message": "Bu rota hız sınırlaması ile korunmaktadır."}

Bu örnek, her istekte client IP adresini kullanarak Redis’te bir sayaç tutar. Eğer client belirli bir süre içinde izin verilen istek sayısını aşarsa, 429 Too Many Requests hatası döndürülür. incr komutu, Redis’teki sayacı atomik olarak artırır.

Örnek Uygulama: Basit Bir API

Aşağıdaki örnek, FastAPI ve Redis’i kullanarak basit bir API oluşturmayı göstermektedir. Bu API, kullanıcı bilgilerini saklamak ve okumak için Redis’i kullanır.


from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import redis
import json

app = FastAPI()

# Redis bağlantı bilgileri
REDIS_HOST = "localhost"
REDIS_PORT = 6379

# Redis bağlantısı oluştur
redis_client = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, decode_responses=True)

class User(BaseModel):
    id: int
    name: str
    email: str

@app.post("/users/")
async def create_user(user: User):
    user_data = user.dict()
    user_id = user_data["id"]
    redis_client.set(f"user:{user_id}", json.dumps(user_data))
    return {"message": f"Kullanıcı {user.name} oluşturuldu"}

@app.get("/users/{user_id}")
async def read_user(user_id: int):
    user_data = redis_client.get(f"user:{user_id}")
    if user_data:
        return json.loads(user_data)
    else:
        raise HTTPException(status_code=404, detail="Kullanıcı bulunamadı")

Bu API, /users/ endpoint’i ile yeni kullanıcılar oluşturabilir ve /users/{user_id} endpoint’i ile kullanıcı bilgilerini okuyabilir. Kullanıcı bilgileri Redis’te JSON formatında saklanır.

İleri Seviye Konular: Asenkron Operasyonlar ve Connection Pooling

Daha karmaşık uygulamalarda, Redis işlemlerini asenkron olarak gerçekleştirmek ve connection pooling kullanmak performansı artırabilir. Asenkron operasyonlar, uygulamanın Redis işlemlerini beklerken diğer işlemleri yapmasına olanak tanır. Connection pooling ise, Redis bağlantılarının yeniden kullanılmasını sağlayarak bağlantı oluşturma maliyetini azaltır.

Asenkron Redis işlemleri için `aioredis` kütüphanesi kullanılabilir.


import asyncio
from fastapi import FastAPI
import aioredis

app = FastAPI()

# Redis bağlantı bilgileri
REDIS_HOST = "localhost"
REDIS_PORT = 6379

# Asenkron Redis bağlantısı oluştur
async def create_redis_pool():
    return await aioredis.create_redis_pool(f"redis://{REDIS_HOST}:{REDIS_PORT}", minsize=5, maxsize=10)

@app.on_event("startup")
async def startup_event():
    app.state.redis = await create_redis_pool()

@app.on_event("shutdown")
async def shutdown_event():
    app.state.redis.close()
    await app.state.redis.wait_closed()

async def get_redis(app):
    return app.state.redis

@app.get("/")
async def read_root(redis=Depends(get_redis)):
    value = await redis.get("mykey")
    if value:
        return {"message": f"Redis'ten okunan değer: {value.decode('utf-8')}"}
    else:
        return {"message": "Redis'te 'mykey' bulunamadı"}

@app.post("/set/{value}")
async def set_value(value: str, redis=Depends(get_redis)):
    await redis.set("mykey", value)
    return {"message": f"'mykey' değeri {value} olarak ayarlandı"}

Bu örnek, `aioredis` kütüphanesi kullanarak asenkron Redis bağlantısı oluşturur ve connection pooling kullanır. minsize ve maxsize parametreleri, connection pool’unun minimum ve maksimum boyutunu belirler.

Güvenlik Konuları

FastAPI ve Redis entegrasyonunda dikkat edilmesi gereken bazı güvenlik konuları şunlardır:

  • Redis Erişimi: Redis sunucusuna yetkisiz erişimi engellemek için parola belirleyin ve güvenlik duvarı kuralları ile erişimi sınırlayın.
  • Veri Şifreleme: Hassas verileri Redis’te saklarken şifreleyin.
  • Oturum Güvenliği: Oturum ID’lerini güvenli bir şekilde saklayın ve oturum hijacking saldırılarına karşı önlemler alın.
  • Giriş Doğrulama: Kullanıcı girişlerini ve API isteklerini doğrulayarak enjeksiyon saldırılarını engelleyin.

Sonuç

FastAPI ve Redis entegrasyonu, yüksek performanslı ve ölçeklenebilir web uygulamaları geliştirmek için güçlü bir kombinasyon sunar. Caching, session yönetimi, rate limiting ve gerçek zamanlı uygulamalar gibi çeşitli senaryolarda Redis’in avantajlarından yararlanılabilir. Bu makalede, FastAPI ve Redis entegrasyonunun temel adımlarını, pratik uygulama örneklerini ve güvenlik konularını inceledik. Bu bilgileri kullanarak, kendi uygulamalarınızda FastAPI ve Redis’i entegre edebilir ve performansınızı önemli ölçüde artırabilirsiniz.

Leave A Comment

Your email address will not be published. Required fields are marked *