Asyncio ve FastAPI ile Asenkron Servis Rehberi

Asyncio ve FastAPI: Üretime Uygun Asenkron Servis Rehberi

Bu rehber, orta seviye Python geliştiricilerinin asyncio olay döngüsünü doğru anlamasını ve FastAPI tabanlı bir servisi bu bilinçle kurmasını hedefler. Örnekler Python 3.11+ ve FastAPI 0.110+ sürümleriyle uyumlu olacak şekilde yazıldı; API'ler bu aralıkta kararlıdır. Kod parçaları çalıştırılabilir olacak şekilde sadeleştirildi, ancak üretim ortamına almadan önce kendi bağımlılık sürümlerinizle doğrulamanız gerekir.

Önkoşullar: Python'da fonksiyon, dekoratör, tip ipuçları ve sanal ortam bilgisi. HTTP ve veritabanı istemcilerinin temel kullanımı.

Eşzamanlılık Neden Gerekli

Bir web servisinin ölçeklenmesini belirleyen şey CPU hızı değil, çoğu zaman G/Ç beklemesidir. Veritabanı sorgusu, dış API çağrısı veya dosya okuma sırasında iş parçacığı bloke olur. Asenkron modelde bu bekleme süresi, olay döngüsünün başka görevleri çalıştırması için boşaltılır.

Kritik ayrım şudur: asyncio tek iş parçacığı üzerinde eşzamanlılık sağlar, paralellik sağlamaz. CPU'yu doyuran iş yükleri (görüntü işleme, ağır matematik) için asyncio çözüm değildir; bu işler ProcessPoolExecutor veya ayrı servisler gerektirir.

Olay Döngüsü ve Coroutine Temelleri

async def ile tanımlanan fonksiyon çağrıldığında çalışmaz, bir coroutine nesnesi döner. Çalışması için await ile beklenmesi veya bir göreve dönüştürülüp olay döngüsüne verilmesi gerekir.

import asyncio

async def fetch_user(user_id: int) -> dict:
    await asyncio.sleep(0.1)  # G/Ç beklemesini temsil eder
    return {"id": user_id}

async def main() -> None:
    user = await fetch_user(1)
    print(user)

asyncio.run(main())

asyncio.run olay döngüsünü kurar, coroutine'i çalıştırır ve döngüyü kapatır. Uygulama kodunda bir kez çağrılır; iç içe asyncio.run çağrısı hata verir.

Eşzamanlı Çalıştırma: gather ve TaskGroup

Birden fazla bağımsız G/Ç işlemini sırayla beklemek zaman kaybıdır. İki temel araç vardır.

asyncio.gather verilen awaitable'ları eşzamanlı çalıştırır ve sonuçları sırayla döner:

async def load_dashboard(user_id: int) -> dict:
    profile, orders, notifications = await asyncio.gather(
        fetch_profile(user_id),
        fetch_orders(user_id),
        fetch_notifications(user_id),
    )
    return {"profile": profile, "orders": orders, "notifications": notifications}

Python 3.11 ile gelen TaskGroup daha güvenli bir alternatiftir: görevlerden biri hata verirse diğerleri iptal edilir ve hata bir ExceptionGroup olarak yükselir.

async def load_dashboard(user_id: int) -> dict:
    async with asyncio.TaskGroup() as tg:
        profile = tg.create_task(fetch_profile(user_id))
        orders = tg.create_task(fetch_orders(user_id))
    return {"profile": profile.result(), "orders": orders.result()}

Yeni kodda TaskGroup tercih edilmelidir; iptal semantiği gather'a göre daha öngörülebilirdir.

Bloke Eden Çağrıların Yeri

En sık yapılan hata, senkron bir kütüphaneyi doğrudan async def içinde çağırmaktır. Örneğin requests.get(...) çağrısı olay döngüsünü tamamen bloke eder ve tüm istekler bekler.

# YANLIŞ: olay döngüsünü bloke eder
async def bad_handler():
    return requests.get("https://example.com").text

# DOĞRU: varsayılan yürütücüde çalıştırır
async def good_handler():
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(None, requests.get, "https://example.com")

Daha temiz çözüm, asenkron bir HTTP istemcisi (httpx, aiohttp) kullanmaktır. Aynı kural veritabanı sürücüleri için de geçerlidir: psycopg2 yerine asyncpg veya psycopg'nin asenkron bağlantısı tercih edilir.

FastAPI'de Async Yol Fonksiyonları

FastAPI, async def ve normal def tanımlı yolları farklı şekilde işler.

• async def yol fonksiyonu doğrudan olay döngüsünde çalışır. İçinde bloke eden çağrı yapılırsa tüm sunucu yavaşlar.
• def yol fonksiyonu iş parçacığı havuzunda çalıştırılır. Senkron kütüphanelerle çalışırken bu güvenli seçenektir.

from fastapi import FastAPI
import httpx

app = FastAPI()

@app.get("/repos/{owner}/{name}")
async def get_repo(owner: str, name: str) -> dict:
    async with httpx.AsyncClient(timeout=5.0) as client:
        resp = await client.get(f"https://api.github.com/repos/{owner}/{name}")
        resp.raise_for_status()
        return resp.json()

AsyncClient'ı her istekte oluşturmak yerine uygulama yaşam döngüsünde bir kez kurup paylaşmak bağlantı havuzundan yararlanmanızı sağlar:

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.http = httpx.AsyncClient(timeout=5.0)
    yield
    await app.state.http.aclose()

app = FastAPI(lifespan=lifespan)

Bağımlılıklar ve Oturum Yönetimi

Veritabanı oturumu gibi kaynaklar bağımlılık olarak enjekte edilip iş bitince kapatılmalıdır. Asenkron sürücülerde bağlantı havuzu genellikle uygulama düzeyinde yaşar; istek düzeyinde yalnızca oturum açılır.

from fastapi import Depends

async def get_session():
    async with async_session_factory() as session:
        yield session

@app.get("/items")
async def list_items(session = Depends(get_session)):
    result = await session.execute(...)
    return result.scalars().all()

yield içeren bağımlılıklar istek tamamlandığında temizleme kodunu çalıştırır; bu, bağlantı sızıntısını önler.

Zaman Aşımları ve İptal

Asenkron kodda bir istek sonsuza kadar bekleyebilir. Her dış çağrıya zaman aşımı koyun. Python 3.11 ile asyncio.timeout bağlam yöneticisi kullanılabilir:

async def call_with_timeout():
    try:
        async with asyncio.timeout(3.0):
            return await slow_operation()
    except TimeoutError:
        return {"error": "timeout"}

İstemci seviyesinde de zaman aşımı ayarlamak gerekir; uygulama seviyesindeki zaman aşımı alt katmandaki soket beklemesini kesmez, sadece beklemeyi sonlandırır.

Yaygın Hatalar

• async def içinde time.sleep, requests, senkron ORM çağrısı kullanmak. Olay döngüsü bloke olur, gecikme tüm isteklere yansır.
• await unutmak. Coroutine çalışmaz, sessizce yok sayılır ve RuntimeWarning üretir.
• asyncio.gather sonucunu sırasız varsaymak. Sonuçlar veriliş sırasına göre döner; karışık sıra beklemeyin.
• Görev referansını tutmamak. asyncio.create_task sonucu bir yerde saklanmazsa görev çöp toplayıcı tarafından iptal edilebilir.
• Aynı olay döngüsünde CPU yoğun iş çalıştırmak. Bu işleri ayrı sürece veya kuyruğa taşıyın.

Güvenlik Etkileri

• Zaman aşımı olmayan dış çağrılar, kaynak tüketimi ve hizmet dışı bırakma (DoS) için kapı aralar. Tüm ağ çağrılarına üst sınır koyun.
• Kullanıcıdan gelen URL veya dosya yolu doğrudan dış istek veya dosya erişimine verilirse SSRF ve yol geçişi riski doğar. Beyaz liste uygulayın.
• Arka planda başlatılan görevlerde (BackgroundTasks, create_task) hata yakalanmazsa sessizce kaybolur. Görevleri izleyin ve hataları loglayın.
• Eşzamanlı erişilen paylaşılan durum (sayaç, önbellek) için kilitleme veya atomik yapılar kullanın; asyncio tek iş parçacığı olsa da await noktaları arasında durum değişebilir.

Test ve Doğrulama

Asenkron kod test edilirken test çalıştırıcısının olay döngüsünü yönetmesi gerekir. pytest-asyncio veya anyio eklentisi bu işi yapar. Testlerde gerçek ağ çağrısı yerine sahte istemci kullanmak, testleri hem hızlı hem kararlı kılar. Örneklerin çalıştığını iddia etmiyorum; kendi ortamınızda doğrulayın.

Sonuç

Asenkron programlamanın özü tek bir kurala indirgenebilir: olay döngüsünü asla bloke etme. FastAPI bu kuralı kolaylaştırır ama zorunlu kılmaz. Yol fonksiyonlarını async mi yoksa senkron mu tanımlayacağınızı kullandığınız kütüphanelere göre seçin, dış çağrılara zaman aşımı koyun, görevleri sahiplenin ve CPU yoğun işleri döngü dışına taşıyın. Bu dört alışkanlık, asenkron servislerin üretimdeki en yaygın sorunlarını baştan engeller.