درس ۲ سطح پیشرفته: Async/Await پیشرفته، WebSocket، Background Tasks و Event-Driven Architecture در پایتون

# همگام (سطح مبتدی)
import requests
import time

start = time.time()
response1 = requests.get("https://httpbin.org/delay/2")
response2 = requests.get("https://httpbin.org/delay/2")
print(f"زمان کل: {time.time() - start:.2f} ثانیه")  # ~4 ثانیه

# ناهمزمان (سطح پیشرفته)
import asyncio
import aiohttp

async def fetch_url(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    async with aiohttp.ClientSession() as session:
        start = time.time()
        tasks = [
            fetch_url(session, "https://httpbin.org/delay/2"),
            fetch_url(session, "https://httpbin.org/delay/2")
        ]
        results = await asyncio.gather(*tasks)
        print(f"زمان کل: {time.time() - start:.2f} ثانیه")  # ~2 ثانیه

import asyncio
import aiohttp

class AsyncProcessor:
    async def process_multiple(self, urls):
        async with aiohttp.ClientSession() as session:
            tasks = [self.fetch_data(session, url) for url in urls]
            return await asyncio.gather(*tasks)
    
    async def fetch_data(self, session, url):
        async with session.get(url) as response:
            return {
                "url": url,
                "status": response.status,
                "length": len(await response.text())
            }

# استفاده
processor = AsyncProcessor()
urls = ["https://httpbin.org/json"] * 10
results = asyncio.run(processor.process_multiple(urls))

from fastapi import FastAPI, WebSocket
from typing import List

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"کلاینت #{client_id}: {data}")
    except:
        manager.disconnect(websocket)

from fastapi import BackgroundTasks
import time

async def send_email(user_id: int):
    await asyncio.sleep(2)  # شبیه‌سازی ارسال ایمیل
    print(f"ایمیل برای کاربر {user_id} ارسال شد!")

@app.post("/order/")
async def create_order(user_id: int, background_tasks: BackgroundTasks):
    # سفارش رو ذخیره کن
    background_tasks.add_task(send_email, user_id)
    return {"message": "سفارش ثبت شد", "status": "ایمیل در حال ارسال..."}

import redis
import json

redis_client = redis.Redis(host='localhost', port=6379, db=0)

# کش کردن نتایج
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    cache_key = f"user:{user_id}"
    cached = redis_client.get(cache_key)
    
    if cached:
        return json.loads(cached)
    
    # از دیتابیس بگیر
    user = await get_user_from_db(user_id)
    redis_client.setex(cache_key, 300, json.dumps(user))  # ۵ دقیقه کش
    return user

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# Queue برای پیام‌ها
channel.queue_declare(queue='order_processing')

def process_order(ch, method, properties, body):
    order_data = json.loads(body)
    print(f"در حال پردازش سفارش {order_data['id']}...")
    # پردازش سفارش
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_consume(queue='order_processing', on_message_callback=process_order)
channel.start_consuming()

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List, Dict
import redis
import json
import asyncio
from datetime import datetime

app = FastAPI(title="سیستم پیام‌رسانی پیشرفته")
redis_client = redis.Redis(host='localhost', port=6379, db=0)

class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []
        self.user_sessions: Dict[str, WebSocket] = {}

    async def connect(self, websocket: WebSocket, user_id: str):
        await websocket.accept()
        self.active_connections.append(websocket)
        self.user_sessions[user_id] = websocket
        redis_client.sadd("active_users", user_id)
        await self.broadcast_user_count()

    def disconnect(self, websocket: WebSocket, user_id: str):
        self.active_connections.remove(websocket)
        del self.user_sessions[user_id]
        redis_client.srem("active_users", user_id)
        asyncio.create_task(self.broadcast_user_count())

    async def send_personal_message(self, message: str, to_user_id: str):
        if to_user_id in self.user_sessions:
            await self.user_sessions[to_user_id].send_text(message)

    async def broadcast_user_count(self):
        count = len(self.user_sessions)
        message = json.dumps({
            "type": "user_count",
            "count": count,
            "timestamp": datetime.utcnow().isoformat()
        })
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/{user_id}")
async def websocket_endpoint(websocket: WebSocket, user_id: str):
    await manager.connect(websocket, user_id)
    try:
        while True:
            data = await websocket.receive_text()
            message = json.loads(data)
            
            if message["type"] == "private":
                await manager.send_personal_message(
                    json.dumps(message), message["to_user_id"]
                )
            else:
                # ذخیره در Redis
                redis_client.lpush(f"messages:{user_id}", json.dumps(message))
                # ارسال به همه
                for connection in manager.active_connections:
                    await connection.send_text(json.dumps(message))
    except WebSocketDisconnect:
        manager.disconnect(websocket, user_id)