Python asyncio ve async/await: Eşzamanlı Programlama

# Senkron — sıralı bekleme
import time
 
def get_data(url):
    time.sleep(0.5)  # ağ gecikmesi simülasyonu
    return f"Veri: {url}"
 
# 10 URL için → 5 saniye
urls = [f"https://api.example.com/{i}" for i in range(10)]
results = [get_data(u) for u in urls]

# Asenkron — eşzamanlı bekleme
import asyncio
import httpx
 
async def get_data(client, url):
    response = await client.get(url)
    return response.json()
 
async def main():
    async with httpx.AsyncClient() as client:
        tasks = [get_data(client, url) for url in urls]
        results = await asyncio.gather(*tasks)
 
# 10 URL için → ~0.5 saniye
asyncio.run(main())

async def merhaba():
    print("Merhaba")
    await asyncio.sleep(1)
    print("Dünya")
 
# Yanlış: doğrudan çağırmak işe yaramaz
# merhaba()  →  <coroutine object merhaba at 0x...>
 
# Doğru: await veya asyncio.run ile çalıştır
asyncio.run(merhaba())

import asyncio
 
async def task(isim, sure):
    print(f"{isim} başladı")
    await asyncio.sleep(sure)  # ← burada loop diğer task'lara geçer
    print(f"{isim} bitti ({sure}s)")
 
async def main():
    # Tüm task'lar eşzamanlı çalışır
    await asyncio.gather(
        task("A", 2),
        task("B", 1),
        task("C", 3),
    )
    # Toplam süre: max(2,1,3) = 3s, 6s değil
 
asyncio.run(main())
# B başladı, A başladı, C başladı
# B bitti (1s)
# A bitti (2s)
# C bitti (3s)

# gather — coroutine'leri alır, hepsini bekler
results = await asyncio.gather(coro1(), coro2(), coro3())
 
# create_task — arka planda başlatır, sonuç sonra alınır
async def main():
    task1 = asyncio.create_task(coro1())
    task2 = asyncio.create_task(coro2())
 
    # ... başka işler yap ...
 
    result1 = await task1
    result2 = await task2

# return_exceptions=True — tek hata tüm grubu kesmez
results = await asyncio.gather(
    fetch_url("https://example.com"),
    fetch_url("https://hatalı.invalid"),
    return_exceptions=True,
)
 
for r in results:
    if isinstance(r, Exception):
        print(f"Hata: {r}")
    else:
        print(f"Veri: {r[:50]}")

import asyncio
import httpx
from typing import Any
 
async def fetch_json(client: httpx.AsyncClient, url: str) -> dict[str, Any]:
    try:
        response = await client.get(url, timeout=10.0)
        response.raise_for_status()
        return response.json()
    except httpx.TimeoutException:
        return {"error": "timeout", "url": url}
    except httpx.HTTPStatusError as e:
        return {"error": e.response.status_code, "url": url}
 
async def fetch_all(urls: list[str]) -> list[dict]:
    limits = httpx.Limits(max_connections=20, max_keepalive_connections=10)
 
    async with httpx.AsyncClient(limits=limits) as client:
        tasks = [fetch_json(client, url) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)
 
# 100 URL'yi eşzamanlı çek
urls = [f"https://jsonplaceholder.typicode.com/posts/{i}" for i in range(1, 101)]
results = asyncio.run(fetch_all(urls))
print(f"{len(results)} sonuç alındı")

# Sunucuyu aşırı yüklememek için eşzamanlı istek sayısını sınırla
async def fetch_limited(sem: asyncio.Semaphore, client: httpx.AsyncClient, url: str):
    async with sem:  # maksimum 10 eşzamanlı istek
        return await fetch_json(client, url)
 
async def main():
    sem = asyncio.Semaphore(10)
    async with httpx.AsyncClient() as client:
        tasks = [fetch_limited(sem, client, url) for url in urls]
        return await asyncio.gather(*tasks)

# Async context manager
class AsyncDatabaseConnection:
    async def __aenter__(self):
        self.conn = await asyncpg.connect(DATABASE_URL)
        return self.conn
 
    async def __aexit__(self, *args):
        await self.conn.close()
 
async def get_users():
    async with AsyncDatabaseConnection() as conn:
        return await conn.fetch("SELECT * FROM users")
 
# Async generator — büyük veriyi parça parça işle
async def read_lines(dosya: str):
    async with aiofiles.open(dosya, 'r', encoding='utf-8') as f:
        async for line in f:
            yield line.strip()
 
async def main():
    async for satir in read_lines("buyuk_dosya.txt"):
        await process(satir)  # her satırı async işle

async def producer(queue: asyncio.Queue, urls: list[str]):
    for url in urls:
        await queue.put(url)
    # Tüketici'ye "bitti" sinyali
    await queue.put(None)
 
async def consumer(queue: asyncio.Queue, worker_id: int):
    async with httpx.AsyncClient() as client:
        while True:
            url = await queue.get()
            if url is None:
                break
            data = await fetch_json(client, url)
            print(f"Worker-{worker_id}: {url}")
            queue.task_done()
 
async def main():
    queue = asyncio.Queue(maxsize=50)
 
    # 3 tüketici çalışsın
    workers = [asyncio.create_task(consumer(queue, i)) for i in range(3)]
    await producer(queue, urls)
 
    await queue.join()
    for w in workers:
        w.cancel()

# YANLIŞ: blocking çağrı event loop'u durdurur
async def yanlis():
    time.sleep(1)        # tüm loop durur!
    requests.get(url)    # sync HTTP — loop bloklanır
 
# DOĞRU: async eşdeğerlerini kullan
async def dogru():
    await asyncio.sleep(1)
    async with httpx.AsyncClient() as c:
        await c.get(url)
 
# YANLIŞ: asyncio.run içinde asyncio.run çağırmak
# RuntimeError: This event loop is already running
 
# DOĞRU: nest_asyncio (Jupyter için) veya farklı yapı
import nest_asyncio
nest_asyncio.apply()  # Jupyter Notebook'ta gerekli

from fastapi import FastAPI
import asyncpg
 
app = FastAPI()
pool: asyncpg.Pool | None = None
 
@app.on_event("startup")
async def startup():
    global pool
    pool = await asyncpg.create_pool(DATABASE_URL, min_size=5, max_size=20)
 
@app.get("/users")
async def get_users():
    async with pool.acquire() as conn:
        rows = await conn.fetch("SELECT id, name, email FROM users")
        return [dict(r) for r in rows]