Python Async Programlama: Eksiksiz Rehber

Python betiğiniz API yanıtlarını, veritabanı sorgularını veya dosya işlemlerini beklerken, bu süre çoğu zaman boşa gider. Python async programlama sayesinde kodunuz aynı anda birden fazla görevi ele alabilir. Böylece bir işlem beklerken diğerleri ilerler; atıl anlar üretken çalışmaya dönüşür ve dakikalar süren beklemeleri çoğu zaman saniyelere indirir.

Bu rehberde, mini projeler üzerinden ilerleyerek Python'da async programlamanın temelini öğreteceğim. Coroutine'lerin, olay döngüsünün (event loop) ve async I/O'nun kodunuzu nasıl çok daha duyarlı hale getirdiğini göreceksiniz.

Eşzamansız web API'leri oluşturmayı öğrenmek istiyorsanız, FastAPI üzerindeki bu kursa mutlaka göz atın.

Python Async Programlama Nedir?

Geleneksel eşzamanlı (senkron) Python'da kodunuz satır satır çalışır. Örneğin bir API çağırdığınızda, programınız yanıtı bekler. Bu iki saniye sürerse, programınızın tamamı iki saniye boyunca boşta kalır. Eşzamansız programlama, kodunuzun bir API çağrısı başlatıp ardından diğer görevlerle devam etmesine olanak tanır.

Yanıt geldiğinde, kodunuz kaldığı yerden devam eder. Her işlemin bitmesini beklemek yerine, birden fazla işlemi aynı anda yürütebilirsiniz. Bu, kodunuzun veritabanları, API'ler veya dosya sistemleri gibi harici sistemlerden yanıt beklediği durumlarda en çok fark yaratır. 

Bunun çalışması için Python'un async sistemi birkaç temel kavram kullanır:

• Coroutine'ler: def yerine async def ile tanımlanan fonksiyonlar. Yürütmeyi duraklatıp sürdürebilirler; bu da bekleme içeren işlemler için idealdir.

• await: Python'a şunu söyler: "Bu coroutine'i, bu işlem tamamlanana kadar duraklat; bu arada başka kodlar çalışabilsin."

• Olay döngüsü (event loop): Tüm coroutine'lerinizi yöneten ve hangisinin ne zaman çalışacağına karar veren motor.

• Görevler (tasks): Eşzamanlı yürütme için sarmalanmış coroutine'ler. Aynı anda birden fazla işlemi çalıştırmak için asyncio.create_task() ile oluşturursunuz.

Async programlamanın neleri yapabildiği (ve yapamadığı) konusunun karışmaması için şunları aklınızda tutun:

• Async, I/O ağırlıklı işlerde en iyi sonucu verir — HTTP istekleri, veritabanı sorguları ve dosya işlemleri gibi kodunuzun harici sistemleri beklediği durumlarda.

• Async, CPU ağırlıklı işlere yardımcı olmaz — karmaşık hesaplamalar veya veri işleme gibi kodunuzun aktif olarak hesap yaptığı durumlarda değil.

Bu kavramları içselleştirmenin en iyi yolu gerçek async kodu yazmaktır. Sıradaki bölümde, ilk async fonksiyonunuzu oluşturacak ve coroutine'lerin olay döngüsüyle nasıl birlikte çalıştığını tam olarak göreceksiniz.

İlk Python Async Fonksiyonunuz

Async koda geçmeden önce, bir şey yapmadan önce bekleyen normal bir senkron fonksiyona bakalım:

import time

def greet_after_delay():
    print("Starting...")
    time.sleep(2)  # Blocks for 2 seconds
    print("Hello!")

greet_after_delay()Altyapı:

Starting...
Hello!Altyapı:

Fonksiyon çalışır, ancak time.sleep(2) tüm programınızı bloklar. Bu iki saniye boyunca başka hiçbir şey çalışamaz.

Şimdi async sürümü:

import asyncio

async def greet_after_delay():
    print("Starting...")
    await asyncio.sleep(2)  # Pauses, but doesn't block
    print("Hello!")

asyncio.run(greet_after_delay())Altyapı:

Starting...
Hello!Altyapı:

Çıktı aynı görünür, ancak perde arkasında farklı bir şey olur. Üç değişiklik bunu async yaptı:

1. def yerine async def ile coroutine olarak bildirildi.

2. time.sleep(2) yerine await asyncio.sleep(2) ile bloklamadan duraklatıldı.

3. asyncio.run() olay döngüsünü başlatıp coroutine'i çalıştırdı.

asyncio.sleep()'in kendisinin de bir async fonksiyon olduğuna ve bu yüzden await gerektirdiğine dikkat edin. Bu kilit bir kuraldır: her async fonksiyon await ile çağrılmalıdır. İster asyncio.sleep() gibi yerleşik bir fonksiyon olsun, ister sizin yazdığınız bir fonksiyon; await'i unutursanız gerçekte çalışmaz.

Şu an için async sürüm daha hızlı görünmüyor. Bunun nedeni yalnızca bir görevimizin olması. Asıl fayda, aynı anda birden çok coroutine çalıştırdığınızda ortaya çıkar; bunu bir sonraki bölümde ele alacağız.

Bilmeniz gereken bir başka önemli şey: async bir fonksiyonu, normal bir fonksiyon gibi doğrudan çağıramazsınız. Deneyelim:

result = greet_after_delay()
print(result)
print(type(result))Altyapı:

<coroutine object greet_after_delay at 0x...>
<class 'coroutine'>Altyapı:

greet_after_delay() çağrısı, sonucu değil bir coroutine nesnesi döndürür. Fonksiyon aslında çalışmaz. Bunu yürütmek için asyncio.run() veya başka bir fonksiyon içinde await kullanmanız gerekir.

Olay döngüsü nasıl çalışır

Olay döngüsü async programlamanın motorudur. Coroutine'lerinizi yönetir ve neyin ne zaman çalışacağına karar verir. Async greet_after_delay() fonksiyonunu çalıştırdığınızda adım adım olanlar:

1. asyncio.run() bir olay döngüsü oluşturur.

2. Olay döngüsü greet_after_delay()'i başlatır.

3. "Starting..." yazdırılır.

4. await asyncio.sleep(2) satırına gelir → coroutine duraklar.

5. Olay döngüsü kontrol eder: "Çalışacak başka görev var mı?" (şimdilik yok).

6. 2 saniye geçer, sleep tamamlanır.

7. Olay döngüsü greet_after_delay()'i sürdürür.

8. "Hello!" yazdırılır.

9. Fonksiyon biter → olay döngüsü çıkar.

Async'i ilginç kılan kısım 5. adımdır. Tek bir coroutine ile yapılacak başka bir şey yoktur. Ancak birden fazla coroutine olduğunda, olay döngüsü biri beklerken diğer işlere geçer. İki saniyelik bir sleep sırasında boşta kalmak yerine başka kodları çalıştırabilir.

Olay döngüsünü bir trafik yöneticisi gibi düşünün. Tek tek arabaları hızlandırmaz. Biri durduğunda, diğerlerine yol vererek akışı sürdürür.

Yaygın async hatası: await'i unutmak

Acemilerin sık yaptığı bir hata, başka bir async fonksiyon içinden coroutine çağırırken await'i unutmak:

import asyncio

async def get_message():
    await asyncio.sleep(1)
    return "Hello!"

async def main():
    message = get_message()  # Missing await!
    print(message)

asyncio.run(main())Altyapı:

<coroutine object get_message at 0x...>
RuntimeWarning: coroutine 'get_message' was never awaitedAltyapı:

await olmadan, dönüş değeri yerine coroutine nesnesi elde edersiniz. Python ayrıca coroutine'in hiç çalıştırılmadığı konusunda uyarır.

Çözüm basit:

async def main():
    message = await get_message()  # Added await
    print(message)

asyncio.run(main())Altyapı:

Hello!Altyapı:

Bir coroutine'in beklenmediğine dair bir RuntimeWarning gördüğünüzde, her async fonksiyon çağrısında await kullandığınızı kontrol edin.

Eşzamanlı Python Async Görevleri 

Önceki bölümde senkron bir fonksiyonu async'e çevirdik. Ancak daha hızlı değildi. Bunun nedeni yalnızca bir coroutine çalıştırmış olmamız. Async'in gerçek gücü, birden fazla coroutine'i aynı anda çalıştırdığınızda ortaya çıkar.

Ardışık await neden hâlâ ardışık çalışır

Birden çok async fonksiyon çağırmanın onları otomatik olarak eşzamanlı çalıştıracağını düşünebilirsiniz. Ancak greet_after_delay()'i üç kez çağırdığımızda ne olduğuna bakın:

import asyncio
import time

async def greet_after_delay(name):
    print(f"Starting {name}...")
    await asyncio.sleep(2)
    print(f"Hello, {name}!")

async def main():
    start = time.perf_counter()
    
    await greet_after_delay("Alice")
    await greet_after_delay("Bob")
    await greet_after_delay("Charlie")
    
    elapsed = time.perf_counter() - start
    print(f"Total time: {elapsed:.2f} seconds")

asyncio.run(main())Altyapı:

Starting Alice...
Hello, Alice!
Starting Bob...
Hello, Bob!
Starting Charlie...
Hello, Charlie!
Total time: 6.01 secondsAltyapı:

İki saniyelik üç görev için altı saniye. Her await, bir sonraki satıra geçmeden önce ilgili coroutine'in bitmesini bekler. Kod async olsa da ardışık çalışır.

asyncio.gather() kullanarak görevleri eşzamanlı çalıştırma 

Coroutine'leri aynı anda çalıştırmak için asyncio.gather() kullanın. Birden fazla coroutine alır ve bunları eşzamanlı yürütür:

async def main():
    start = time.perf_counter()
    
    await asyncio.gather(
        greet_after_delay("Alice"),
        greet_after_delay("Bob"),
        greet_after_delay("Charlie"),
    )
    
    elapsed = time.perf_counter() - start
    print(f"Total time: {elapsed:.2f} seconds")

asyncio.run(main())Altyapı:

Starting Alice...
Starting Bob...
Starting Charlie...
Hello, Alice!
Hello, Bob!
Hello, Charlie!
Total time: 2.00 secondsAltyapı:

Altı yerine iki saniye. Üç coroutine de anında başladı, eşzamanlı olarak uyudu ve birlikte tamamlandı. Tek bir değişiklikle 3 kat hızlanma.

Çıktı sırasına dikkat edin: tüm "Starting..." mesajları, herhangi bir "Hello..." mesajından önce yazdırılıyor. Bu, coroutine'lerin birbirini beklemek yerine aynı iki saniyelik zaman penceresinde çalıştığını gösterir.

asyncio.gather(), sonuçları, coroutine'leri hangi sırayla verdiyseniz o sırada bir liste halinde döndürür. Coroutine'leriniz değer döndürüyorsa, bunları yakalayabilirsiniz:

async def fetch_number(n):
    await asyncio.sleep(1)
    return n * 10

async def main():
    results = await asyncio.gather(
        fetch_number(1),
        fetch_number(2),
        fetch_number(3),
    )
    print(results)

asyncio.run(main())Altyapı:

[10, 20, 30]Altyapı:

Sonuçlar, gather()'a verilen coroutine'lerin sırasına uygun olarak [10, 20, 30] şeklinde döner.

aiohttp ile Python Async HTTP İstekleri

Şimdiye dek gecikmeleri simüle etmek için asyncio.sleep() kullandık. Şimdi gerçek HTTP istekleri yapalım. requests kütüphanesine yönelebilirsiniz, ancak burada işe yaramaz. requests senkron çalışır ve olay döngüsünü bloklar; async'in amacını boşa çıkarır.

Bunun yerine, bu amaç için geliştirilmiş bir async HTTP istemcisi olan aiohttp'u kullanın.

aiohttp'ya giriş

aiohttp ile bir URL nasıl çekilir:

import aiohttp
import asyncio

async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def main():
    html = await fetch("https://example.com")
    print(f"Fetched {len(html)} characters")

asyncio.run(main())Altyapı:

Fetched 513 charactersAltyapı:

İç içe iki async with bloğuna dikkat edin. Her biri farklı bir kaynağı yönetir ve bunların ne yaptığını anlamak aiohttp'yu doğru kullanmanın anahtarıdır.

aiohttp'da ClientSession nasıl çalışır

aiohttp ile istek yaptığınızda adım adım olanlar:

1. aiohttp.ClientSession() bir bağlantı havuzu oluşturur (başta boştur).

2. session.get(url) havuzu kontrol eder: "Bu host'a (web sitesinin sunucusu) açık bir bağlantı var mı?"

3. Bağlantı yoksa yeni bir TCP bağlantısı ve bir SSL el sıkışması oluşturulur.

4. Bir HTTP isteği gönderilir ve yanıt başlıklarını bekleriz.

5. Yanıt nesnesi bağlantıyı tutar.

6. await response.text() gövde verisini ağdan okur.

7. İçteki async with bloğundan çıkış: Bağlantı havuza geri döner (açık kalır!).

8. Aynı host'a yapılacak bir sonraki istek, havuzdaki bağlantıyı yeniden kullanır (3. adımı atlar).

9. Dıştaki async with bloğundan çıkış: Havuzdaki tüm bağlantılar kapanır.

7 ve 8. adımlar kilit önemdedir. Bağlantı havuzu, istekler arasında bağlantıları canlı tutar. Aynı host'a bir istek daha yaptığınızda, TCP ve SSL el sıkışması tamamen atlanır.

Bu önemlidir çünkü yeni bir bağlantı kurmak yavaştır. Bir TCP el sıkışması sunucuya gidiş-dönüş gerektirir. Bir SSL el sıkışması iki tane daha. Gecikmeye bağlı olarak, ilk baytı göndermeden önce 100-300 ms anlamına gelebilir.

Tüm istekler için paylaşılan bir oturum kullanma

Şimdi her istek için yeni bir oturum oluşturmanın neden sorun olduğunu görebilirsiniz:

# Wrong: new session for each request
async def fetch_bad(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def main():
    urls = ["https://example.com"] * 10
    results = await asyncio.gather(*[fetch_bad(url) for url in urls])Altyapı:

fetch_bad()'in her çağrısı, boş bir havuzla yeni bir oturum oluşturur. Tüm istekler aynı host'a gitse bile her istek tam el sıkışma maliyetini öder.

Çözüm, tek bir oturum oluşturup fetch fonksiyonunuza geçirmektir:

# Right: reuse a single session
async def fetch_good(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = ["https://example.com"] * 10
    async with aiohttp.ClientSession() as session:
        results = await asyncio.gather(*[fetch_good(session, url) for url in urls])Altyapı:

Paylaşılan bir oturumla, ilk istek bağlantıyı kurar; kalan dokuz istek onu yeniden kullanır. On el sıkışma yerine bir.

Async HTTP isteği örneği: Hacker News kazıma

Bunu Hacker News API ile uygulayalım. Bu API, hikâyeleri çekmek için birden fazla istek gerektirdiğinden eşzamansız davranışı göstermek için idealdir. Python'da REST API'lerle çalışmaya yeniyseniz, temel kavramlar için Python API'leri: API Oluşturma ve Kullanma Rehberi'ne göz atın.

Hacker News API yapısı:

• https://hacker-news.firebaseio.com/v0/topstories.json hikâye ID'lerinin bir listesini (sadece sayılar) döndürür

• https://hacker-news.firebaseio.com/v0/item/{id}.json tek bir hikâye için ayrıntıları döndürür

10 hikâye almak için 11 istek gerekir: biri ID listesi için, ardından her hikâye için birer tane. Async programlamanın parladığı nokta tam da burası.

Önce, ilk hikâyeyi çekmeyi denersek API ne döndürüyor görelim:

import aiohttp
import asyncio

HN_API = "https://hacker-news.firebaseio.com/v0"

async def main():
    async with aiohttp.ClientSession() as session:
        # Get top story IDs
        async with session.get(f"{HN_API}/topstories.json") as response:
            story_ids = await response.json()
        
        print(f"Found {len(story_ids)} stories")
        print(f"First 5 IDs: {story_ids[:5]}")
        
        # Fetch first story details
        first_id = story_ids[0]
        async with session.get(f"{HN_API}/item/{first_id}.json") as response:
            story = await response.json()
        
        print(f"\nStory structure:")
        for key, value in story.items():
            print(f"  {key}: {repr(value)[:50]}")

asyncio.run(main())Altyapı:

Found 500 stories
First 5 IDs: [46051449, 46055298, 46021577, 46053566, 45984864]

Story structure:
  by: 'mikeayles'
  descendants: 22
  id: 46051449
  kids: [46054027, 46053889, 46053275, 46053515, 46053002,
  score: 217
  text: 'I got DOOM running in KiCad by rendering it with 
  time: 1764108815
  title: 'Show HN: KiDoom – Running DOOM on PCB Traces'
  type: 'story'
  url: 'https://www.mikeayles.com/#kidoom'Altyapı:

API 500 hikâye ID'si döndürür ve her hikâyede title, url, score ve by (yazar) gibi alanlar bulunur. 

Birden çok sonucu ardışık vs. eşzamanlı çekme

Şimdi 10 hikâyeyi ardışık olarak çekelim:

import aiohttp
import asyncio
import time

HN_API = "https://hacker-news.firebaseio.com/v0"

async def fetch_story(session, story_id):
    async with session.get(f"{HN_API}/item/{story_id}.json") as response:
        return await response.json()

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get(f"{HN_API}/topstories.json") as response:
            story_ids = await response.json()
        
        start = time.perf_counter()
        stories = []
        for story_id in story_ids[:10]:
            story = await fetch_story(session, story_id)
            stories.append(story)
        elapsed = time.perf_counter() - start
        
        print(f"Sequential: Fetched {len(stories)} stories in {elapsed:.2f} seconds")

asyncio.run(main())Altyapı:

Sequential: Fetched 10 stories in 2.41 secondsAltyapı:

Şimdi aynı hikâyeleri eşzamanlı olarak çekelim:

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get(f"{HN_API}/topstories.json") as response:
            story_ids = await response.json()
        
        start = time.perf_counter()
        tasks = [fetch_story(session, story_id) for story_id in story_ids[:10]]
        stories = await asyncio.gather(*tasks)
        elapsed = time.perf_counter() - start
        
        print(f"Concurrent: Fetched {len(stories)} stories in {elapsed:.2f} seconds")
        print("\nTop 3 stories:")
        for story in stories[:3]:
            print(f"  - {story.get('title', 'No title')}")

asyncio.run(main())Altyapı:

Concurrent: Fetched 10 stories in 0.69 seconds

Top 3 stories:
  - Show HN: KiDoom – Running DOOM on PCB Traces
  - AWS is 10x slower than a dedicated server for the same price [video]
  - Surprisingly, Emacs on Android is pretty goodAltyapı:

Eşzamanlı sürüm 3,5 kat daha hızlı. Her isteğin bitmesini bekleyip sonra yenisini başlatmak yerine, 10 isteğin tamamı aynı anda çalışır. Async programlama gerçek ağ I/O'sunda burada karşılığını verir.

Python Async Hata Yönetimi ve Oran Sınırlama

Verileri eşzamanlı çekerken birkaç şey ters gidebilir. Çok fazla istekle sunucuyu bunaltabilirsiniz. Bazı istekler sonsuza kadar asılı kalabilir. Bazıları ise doğrudan başarısız olabilir. Ve hatalar olduğunda, bir kurtarma stratejisine ihtiyacınız vardır.

Bu bölüm, isteklerin gönderimini kontrol etmekten zaman sınırları koymaya, hataları ele almaya ve uygun olduğunda yeniden denemeye kadar endişeleri oluş sırasıyla ele alır. Python'da istisna yönetiminin temellerine tazeleme gerekiyorsa, Python'da İstisna ve Hata Yönetimi'ne bakın. Aşağıdaki temel kurulum boyunca bunu kullanacağız:

import aiohttp
import asyncio
import time

HN_API = "https://hacker-news.firebaseio.com/v0"

async def fetch_story(session, story_id):
    async with session.get(f"{HN_API}/item/{story_id}.json") as response:
        return await response.json()Altyapı:

Semaphore'larla oran sınırlama

Önceki bölümde 10 isteği birden gönderdik. Bu sorun olmadı. Peki 500 hikâye çekmeniz gerekirse ne olur? Ya da 10.000 sayfa kazımanız?

Çoğu API, oran sınırlaması uygular. Saniyede 10 isteğe veya 100 eşzamanlı bağlantıya izin verebilirler. Bu sınırları aşarsanız engellenir, yavaşlatılır veya yasaklanırsınız. API sınır dayatmasa bile, binlerce isteği aynı anda göndermek kendi sisteminizi veya sunucuyu zorlayabilir.

Her an "uçuşta" (in flight) olan istek sayısını kontrol etmenin bir yoluna ihtiyacınız var. Semaphore tam da bunu yapar.

Semaphore, izin sistemi gibi çalışır. Üç izniniz olduğunu hayal edin. İstek yapmak isteyen her görev önce bir izin almalıdır. Bittiğinde izni iade eder, böylece yeni bir istek onu kullanabilir. İzin yoksa, görev biri serbest kalana kadar bekler.

3 izin ve 4 veya daha fazla görevle nasıl işler:

1. Üç izin mevcuttur.

2. A Görevi bir izin alır (2 kaldı), isteğine başlar.

3. B Görevi bir izin alır (1 kaldı), isteğine başlar.

4. C Görevi bir izin alır (0 kaldı), isteğine başlar.

5. D Görevi izin ister, ancak yoktur—bekler.

6. A Görevi biter, iznini iade eder (1 mevcut).

7. D Görevi o izni alır ve isteğine başlar.

8. Tüm görevler bitene kadar bu şekilde devam eder.

5. adımdaki bekleme verimlidir. Görev "izin boşaldı mı?" diye dönüp durmaz. Duraklar ve başka koda çalışması için izin verir. Olay döngüsü, bir izin kullanılabilir olduğunda onu uyandırır.

Koda bakalım. asyncio'da asyncio.Semaphore(n) ile bir semaphore oluşturursunuz; n izin sayısıdır. Kullanmak için kodunuzu async with semaphore: içine sarın. Bloğa girerken bir izin alınır, çıkarken otomatik olarak serbest bırakılır:

async def fetch_story_limited(session, story_id, semaphore):
    async with semaphore:  # Acquire permit (or wait if none available)
        async with session.get(f"{HN_API}/item/{story_id}.json") as response:
            return await response.json()
    # Permit automatically released hereAltyapı:

30 hikâyeyi semaphore ile ve semaphoresuz çekmeyi karşılaştıralım:

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get(f"{HN_API}/topstories.json") as response:
            story_ids = (await response.json())[:30]

        # Without rate limiting: all 30 at once
        start = time.perf_counter()
        await asyncio.gather(*[fetch_story(session, sid) for sid in story_ids])
        print(f"No limit: {time.perf_counter() - start:.2f}s (30 concurrent)")

        # With Semaphore(5): max 5 at a time
        semaphore = asyncio.Semaphore(5)
        start = time.perf_counter()
        await asyncio.gather(*[fetch_story_limited(session, sid, semaphore) for sid in story_ids])
        print(f"Semaphore(5): {time.perf_counter() - start:.2f}s (5 concurrent)")

asyncio.run(main())Altyapı:

No limit: 0.62s (30 concurrent)
Semaphore(5): 1.50s (5 concurrent)Altyapı:

Semaphore sürümü daha yavaştır çünkü istekleri beşerli partiler halinde işler. Ancak bu bir ödünleşimdir: hızdan feragat edip öngörülebilir ve sunucu dostu davranış kazanırsınız.

Unutmayın: semaphore, zaman birimi başına istekleri değil, eşzamanlı istek sayısını sınırlar. Semaphore(10) "aynı anda en fazla 10 istek" demektir, "saniyede 10 istek" değil. Kesin zaman tabanlı oran sınırlaması (ör. tam olarak saniyede 10 istek) gerekiyorsa, semaphore'u partiler arası gecikmelerle birleştirebilir veya aiolimiter gibi bir kütüphane kullanabilirsiniz.

asyncio.wait_for() ile zaman aşımları

Eşzamanlılık kontrolünde olsanız bile, tekil istekler takılabilir. Sunucu bağlantınızı kabul eder ama hiç yanıt vermez. Zaman aşımı olmadan programınız süresiz bekler.

asyncio.wait_for() herhangi bir coroutine'i bir son tarih ile sarar. Coroutine'i ve saniye cinsinden bir timeout verirsiniz. İşlem zamanında tamamlanmazsa asyncio.TimeoutError fırlatır:

async def slow_operation():
    print("Starting slow operation...")
    await asyncio.sleep(5)
    return "Done"

async def main():
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=2.0)
        print(f"Success: {result}")
    except asyncio.TimeoutError:
        print("Operation timed out after 2 seconds")

asyncio.run(main())Altyapı:

Starting slow operation...
Operation timed out after 2 secondsAltyapı:

Zaman aşımı dolduğunda wait_for() coroutine'i iptal eder. TimeoutError'ı yakalayıp ne yapacağınıza karar verebilirsiniz: isteği atlamak, varsayılan bir değer döndürmek veya yeniden denemek.

Eşzamanlı isteklerde, her birini ayrı ayrı sarmalayın. İşte hata yerine bir hata sözlüğü döndüren bir yardımcı:

async def fetch_story_with_timeout(session, story_id, timeout=5.0):
    try:
        coro = fetch_story(session, story_id)
        return await asyncio.wait_for(coro, timeout=timeout)
    except asyncio.TimeoutError:
        return {"error": f"Story {story_id} timed out"}Altyapı:

Bir coroutine iptal edildiğinde (zaman aşımı veya başka bir nedenle), Python içinde asyncio.CancelledError fırlatır. Coroutine'iniz dosya tanıtıcıları veya bağlantılar gibi kaynaklar tutuyorsa, try/finally kullanarak iptal durumunda bile temizliğin yapıldığından emin olun:

async def fetch_with_cleanup(session, url):
    print("Starting fetch...")
    try:
        async with session.get(url) as response:
            return await response.text()
    finally:
        print("Cleanup complete")  # Runs even on cancellationAltyapı:

asyncio.gather() ile hata yönetimi

Zaman aşımları yavaş istekleri yakalar. Ancak bazı istekler hatayla anında başarısız olur. Bir partideki tek bir istek başarısız olursa ne olduğunu görelim.

Önce, geçersiz ID'lerde istisna fırlatan bir fetch_story() sürümüne ihtiyacımız var:

async def fetch_story_strict(session, story_id):
    story = await fetch_story(session, story_id)
    if story is None:
        raise ValueError(f"Story not found: {story_id}")
    return storyAltyapı:

Şimdi dört geçerli hikâye ve bir geçersiz ID çekelim:

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get(f"{HN_API}/topstories.json") as response:
            story_ids = await response.json()

        ids_to_fetch = story_ids[:4] + [99999999999]  # 4 valid + 1 invalid

        try:
            stories = await asyncio.gather(
                *[fetch_story_strict(session, sid) for sid in ids_to_fetch]
            )
            print(f"Got {len(stories)} stories")
        except ValueError as e:
            print(f"ERROR: {e}")

asyncio.run(main())Altyapı:

ERROR: Story not found: 99999999999Altyapı:

Tek bir geçersiz ID ile, dört başarılı sonucu da kaybediyoruz. Varsayılan olarak gather() hızlı-arıza (fail-fast) davranışı kullanır: bir istisna her şeyi iptal eder ve yukarı taşır.

Kısmi sonuçları korumak için return_exceptions=True ekleyin. Bu, gather()'ın davranışını değiştirir: istisna fırlatmak yerine, başarılı değerlerin yanında sonuç listesinde öğe olarak döndürür:

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get(f"{HN_API}/topstories.json") as response:
            story_ids = await response.json()

        ids_to_fetch = story_ids[:4] + [99999999999]

        results = await asyncio.gather(
            *[fetch_story_strict(session, sid) for sid in ids_to_fetch],
            return_exceptions=True  # Don't raise, return exceptions in list
        )

        # Separate successes from failures using isinstance()
        stories = [r for r in results if not isinstance(r, Exception)]
        errors = [r for r in results if isinstance(r, Exception)]

        print(f"Got {len(stories)} stories, {len(errors)} failed")

asyncio.run(main())Altyapı:

Got 4 stories, 1 failedAltyapı:

isinstance(result, Exception) kontrolü, başarılı sonuçları hatalardan ayırmanıza olanak tanır. Sonrasında çalışanları işleyebilir, hataları günlüğe alabilir veya yeniden deneyebilirsiniz.

Üstel geri çekilmeyle (exponential backoff) yeniden deneme mantığı

Bazı hatalar geçicidir. Sunucu kısa süreliğine aşırı yüklenmiş olabilir veya ağdaki bir aksaklık bağlantınızı düşürebilir. Bu durumlarda yeniden denemek mantıklıdır.

Ancak anında yeniden denemek işleri daha da kötüleştirebilir. Sunucu zorlanıyorsa, onu yeniden denemelerle bombardımana tutmak sorunu büyütür. Üstel geri çekilme, her deneme arasında bekleme süresini uzatarak bunu çözer.

Desen, bekleme sürelerini hesaplamak için 2 ** attempt kullanır: 0. deneme bir saniye (2⁰), 1. deneme iki saniye (2¹), 2. deneme dört saniye (2²) bekler, vb. Bu, sunucuya toparlanması için giderek daha fazla zaman tanır:

async def fetch_with_retry(session, story_id, max_retries=3):
    for attempt in range(max_retries):
        try:
            story = await fetch_story(session, story_id)
            if story is None:
                raise ValueError(f"Story {story_id} not found")
            return story
        except (aiohttp.ClientError, ValueError):  # Catch specific exceptions
            if attempt == max_retries - 1:
                print(f"Story {story_id}: Failed after {max_retries} attempts")
                return None

            backoff = 2 ** attempt  # 1s, 2s, 4s...
            print(f"Story {story_id}: Attempt {attempt + 1} failed, retrying in {backoff}s...")
            await asyncio.sleep(backoff)Altyapı:

Dikkat ederseniz, genel bir except yerine belirli istisnaları (aiohttp.ClientError, ValueError) yakalıyoruz. Bu, yalnızca geçici olabilecek hatalarda yeniden denediğimizden emin olur. Kötü koddaki bir KeyError yeniden denemeyi tetiklememelidir.

Geçerli ve geçersiz ID'lerin karışımıyla test edelim:

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get(f"{HN_API}/topstories.json") as response:
            story_ids = await response.json()

        test_ids = [story_ids[0], 99999999999, story_ids[1], 88888888888, story_ids[2]]

        results = await asyncio.gather(*[fetch_with_retry(session, sid) for sid in test_ids])

        successful = [r for r in results if r is not None]
        print(f"\nSuccessful: {len(successful)}, Failed: {len(test_ids) - len(successful)}")

asyncio.run(main())Altyapı:

Story 99999999999: Attempt 1 failed, retrying in 1s...
Story 88888888888: Attempt 1 failed, retrying in 1s...
Story 99999999999: Attempt 2 failed, retrying in 2s...
Story 88888888888: Attempt 2 failed, retrying in 2s...
Story 99999999999: Failed after 3 attempts
Story 88888888888: Failed after 3 attempts

Successful: 3, Failed: 2Altyapı:

Canlı sistemlerde, birden fazla başarısız isteğin tam olarak aynı anda yeniden denememesini sağlamak için titreşim (küçük rastgele gecikmeler) de eklersiniz. Ayrıca yalnızca geçici hataları (ör. 503 gibi sunucu tarafı ağ sorunları) yeniden dener, kalıcı olanlarda (ör. 404 veya 401) hemen vazgeçersiniz.

aiosqlite ile Python Async Veritabanı Depolama

Hacker News hikâyelerini uygun oran sınırlaması, zaman aşımları ve hata yönetimiyle çekiyorduk. Şimdi bunları bir veritabanına kaydedelim.

sqlite3 gibi normal bir senkron veritabanı kütüphanesi kullanmak, sorgular sırasında olay döngüsünü bloklayarak eşzamansız programlamanın amacını boşa çıkarır. Kodunuz veritabanını beklerken başka coroutine'ler çalışamaz. Async uygulamalar için async bir veritabanı kütüphanesine ihtiyaç vardır.

aiosqlite, Python'un yerleşik sqlite3'ünü async bir arayüzle sarmalar. Veritabanı işlemlerini bir iş parçacığı havuzunda çalıştırır, böylece olay döngüsünü bloklamaz. SQLite sunucu kurulumu gerektirmez — sadece bir dosyadır — bu nedenle bu kodu hemen çalıştırabilirsiniz. Python'da veritabanlarıyla çalışmaya yeniyseniz, Introduction to Databases in Python kursu, aiosqlite'ın üzerine inşa ettiği senkron temelleri kapsar.

Veritabanını kurma

Desen tanıdık gelmeli. Tıpkı aiohttp.ClientSession gibi, bağlantıyı yönetmek için async with kullanırsınız:

import aiosqlite

async def init_db(db_path):
    async with aiosqlite.connect(db_path) as db:
        await db.execute("""
            CREATE TABLE IF NOT EXISTS stories (
                id INTEGER PRIMARY KEY,
                title TEXT,
                url TEXT,
                score INTEGER,
                fetched_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        await db.commit()

asyncio.run(init_db("stories.db"))Altyapı:

Temel fonksiyonlar:

• aiosqlite.connect(path) bir veritabanı dosyası açar (veya oluşturur).

• await db.execute(sql) bir SQL ifadesi çalıştırır.

• await db.commit() değişiklikleri diske kaydeder.

Hikâyeleri kaydetme

İşte tek bir hikâyeyi kaydeden fonksiyon:

async def save_story(db, story):
    await db.execute(
        "INSERT OR REPLACE INTO stories (id, title, url, score) VALUES (?, ?, ?, ?)",
        (story["id"], story.get("title", ""), story.get("url", ""), story.get("score", 0))
    )Altyapı:

? yer tutucuları SQL enjeksiyonunu önler — değerleri SQL'e eklemek için asla f-string kullanmayın. INSERT OR REPLACE, hikâyeleri tekrar çekersek mevcut kayıtları günceller.

Python'da Tam Async Boru Hattı: Getir ve Kaydet

Şimdi bu eğitimdeki her şeyi bir araya getirerek eksiksiz bir boru hattı oluşturalım. 20 Hacker News hikâyesini oran sınırlamasıyla çekip bir veritabanına kaydedeceğiz:

import aiohttp
import aiosqlite
import asyncio

HN_API = "https://hacker-news.firebaseio.com/v0"

async def fetch_story(session, story_id):
    async with session.get(f"{HN_API}/item/{story_id}.json") as response:
        return await response.json()

async def fetch_story_limited(session, story_id, semaphore):
    async with semaphore:
        story = await fetch_story(session, story_id)
        if story:
            return story
        return None

async def save_story(db, story):
    await db.execute(
        "INSERT OR REPLACE INTO stories (id, title, url, score) VALUES (?, ?, ?, ?)",
        (story["id"], story.get("title", ""), story.get("url", ""), story.get("score", 0))
    )

async def main():
    # Initialize database
    async with aiosqlite.connect("hn_stories.db") as db:
        await db.execute("""
            CREATE TABLE IF NOT EXISTS stories (
                id INTEGER PRIMARY KEY,
                title TEXT,
                url TEXT,
                score INTEGER,
                fetched_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        
        # Fetch stories
        async with aiohttp.ClientSession() as session:
            async with session.get(f"{HN_API}/topstories.json") as response:
                story_ids = await response.json()
            
            semaphore = asyncio.Semaphore(5)
            tasks = [fetch_story_limited(session, sid, semaphore) for sid in story_ids[:20]]
            stories = await asyncio.gather(*tasks)
        
        # Save to database
        for story in stories:
            if story:
                await save_story(db, story)
        await db.commit()
        
        # Query and display
        cursor = await db.execute("SELECT id, title, score FROM stories ORDER BY score DESC LIMIT 5")
        rows = await cursor.fetchall()
        
        print(f"Saved {len([s for s in stories if s])} stories. Top 5 by score:")
        for row in rows:
            print(f"  [{row[2]}] {row[1][:50]}")

asyncio.run(main())Altyapı:

Saved 20 stories. Top 5 by score:
  [671] Google Antigravity exfiltrates data via indirect p
  [453] Trillions spent and big software projects are stil
  [319] Ilya Sutskever: We're moving from the age of scali
  [311] Show HN: We built an open source, zero webhooks pa
  [306] FLUX.2: Frontier Visual IntelligenceAltyapı:

Boru hattı, her bölümdeki desenleri kullanır: bağlantı havuzu için ClientSession, oran sınırlaması için Semaphore(5), eşzamanlı getirme için gather() ve şimdi de async depolama için aiosqlite. Her bileşen, diğerlerini bloklamadan kendi işini yapar.

Bu iş akışını her çalıştırdığınızda, o günün en popüler hikâyelerini alırsınız.

Sonuç

Bu eğitim sizi temel async/await söz diziminden eksiksiz bir veri boru hattına taşıdı. Coroutine'lerin nasıl duraklayıp devam ettiğini, olay döngüsünün eşzamanlı görevleri nasıl yönettiğini ve asyncio.gather()'ın birden çok işlemi nasıl aynı anda çalıştırdığını öğrendiniz. aiohttp ile gerçek HTTP istekleri eklediniz, semaphore'larla eşzamanlılığı kontrol ettiniz, zaman aşımları ve yeniden denemelerle hataları yönettiniz ve aiosqlite ile sonuçları bir veritabanında sakladınız.

Kodunuz harici sistemleri beklediğinde async kullanın: HTTP API'leri, veritabanları, dosya I/O'su veya ağ soketleri. Veri işleme veya sayı hesaplama gibi CPU-yoğun işler için async yardımcı olmaz — bunun yerine multiprocessing veya concurrent.futures'a bakın. Daha ileri gitmek için asyncio dokümantasyonunu keşfedebilir ve async web API'leri oluşturmak için FastAPI'yi düşünebilirsiniz. 

Bu bilginin üzerine inşa ederek akıllı uygulamalar tasarlamayı öğrenmek istiyorsanız, Geliştiriciler için Associate AI Engineer kariyer yoluna mutlaka göz atın.