البرمجة غير المتزامنة مع asyncio

البرمجة غير المتزامنة تتيح لك تشغيل استدعاءات API متعددة بشكل متزامن. بدلاً من انتظار اكتمال كل استدعاء، يمكنك بدء عدة في وقت واحد.

لماذا غير متزامن؟

متزامن (بطيء):
الاستدعاء 1: |-------|
الاستدعاء 2:         |-------|
الاستدعاء 3:                 |-------|
الإجمالي: ~3 ثوانٍ

غير متزامن (سريع):
الاستدعاء 1: |-------|
الاستدعاء 2: |-------|
الاستدعاء 3: |-------|
الإجمالي: ~1 ثانية

بناء جملة Async الأساسي

import asyncio

# تعريف دالة async مع 'async def'
async def fetch_data():
    print("جارٍ البدء...")
    await asyncio.sleep(1)  # انتظار غير معطّل
    print("تم!")
    return "data"

# تشغيل دالة async
result = asyncio.run(fetch_data())

HTTP غير متزامن مع aiohttp

pip install aiohttp

import aiohttp
import asyncio

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

# تشغيله
data = asyncio.run(fetch_url("https://api.example.com/data"))

إجراء استدعاءات API متزامنة

import aiohttp
import asyncio
import os

async def chat_completion(session, prompt):
    """إجراء طلب إكمال محادثة واحد."""
    url = "https://api.openai.com/v1/chat/completions"
    headers = {
        "Authorization": f"Bearer {os.getenv('OPENAI_API_KEY')}",
        "Content-Type": "application/json"
    }
    data = {
        "model": "gpt-4",
        "messages": [{"role": "user", "content": prompt}]
    }

    async with session.post(url, headers=headers, json=data) as response:
        result = await response.json()
        return result["choices"][0]["message"]["content"]

async def batch_completions(prompts):
    """معالجة مطالبات متعددة بشكل متزامن."""
    async with aiohttp.ClientSession() as session:
        tasks = [chat_completion(session, p) for p in prompts]
        results = await asyncio.gather(*tasks)
        return results

# معالجة 5 مطالبات بشكل متزامن
prompts = [
    "ما هو بايثون؟",
    "ما هو جافاسكريبت؟",
    "ما هو Rust؟",
    "ما هو Go؟",
    "ما هو Ruby؟"
]

results = asyncio.run(batch_completions(prompts))
for prompt, result in zip(prompts, results):
    print(f"س: {prompt}\nج: {result[:100]}...\n")

المفاهيم الرئيسية

المفهوم	الوصف
async def	تعرّف دالة coroutine
await	تتوقف مؤقتاً حتى اكتمال العملية غير المتزامنة
asyncio.run()	نقطة الدخول لتشغيل كود async
asyncio.gather()	تشغيل coroutines متعددة بشكل متزامن
aiohttp	مكتبة عميل HTTP غير متزامنة

معالجة الأخطاء في Async

async def safe_fetch(session, url):
    try:
        async with session.get(url, timeout=10) as response:
            response.raise_for_status()
            return await response.json()
    except asyncio.TimeoutError:
        print(f"انتهت المهلة: {url}")
        return None
    except aiohttp.ClientError as e:
        print(f"خطأ: {e}")
        return None

async def fetch_all(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [safe_fetch(session, url) for url in urls]
        return await asyncio.gather(*tasks)

متى تستخدم Async

استخدم Async	استخدم Sync
استدعاءات API متعددة	استدعاء API واحد
مهام مرتبطة بـ I/O	مهام مرتبطة بـ CPU
استخراج الويب	سكريبتات بسيطة
البث في الوقت الفعلي	المعالجة التسلسلية

لقد أكملت وحدة العمل مع APIs! بعد ذلك، سنتعلم عن بنية المشروع. :::