درس ۴ سطح پیشرفته: GraphQL، WebSocket پایتون پیشرفته، SSE و Real-time Dashboard در پایتون

# pip install strawberry-graphql fastapi
import strawberry
from strawberry.fastapi import GraphQLRouter
from fastapi import FastAPI

@strawberry.type
class User:
    id: int
    name: str
    email: str

@strawberry.type
class Query:
    @strawberry.field
    def user(self, id: int) -> User:
        return User(id=id, name="علی", email="ali@example.com")
    
    @strawberry.field
    def users(self) -> list[User]:
        return [User(id=1, name="علی", email="ali@example.com")]

schema = strawberry.Schema(query=Query)
graphql_app = GraphQLRouter(schema)

app = FastAPI()
app.include_router(graphql_app, prefix="/graphql")

import asyncio
from typing import AsyncGenerator

@strawberry.type
class Subscription:
    @strawberry.subscription
    async def count(self, target: int = 10) -> AsyncGenerator[int, None]:
        for i in range(target):
            yield i
            await asyncio.sleep(1)

from fastapi import WebSocket, WebSocketDisconnect
from typing import List
import json

class RealtimeManager:
    def __init__(self):
        self.connections: List[WebSocket] = []
    
    async def connect(self, ws: WebSocket):
        await ws.accept()
        self.connections.append(ws)
    
    async def broadcast(self, data: dict):
        message = json.dumps(data)
        for conn in self.connections[:]:
            try:
                await conn.send_text(message)
            except:
                self.connections.remove(conn)

manager = RealtimeManager()

@app.websocket("/ws/dashboard")
async def websocket_dashboard(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            # دریافت پیام از کلاینت
            data = await websocket.receive_json()
            await manager.broadcast({
                "type": "new_data",
                "payload": data,
                "timestamp": datetime.utcnow().isoformat()
            })
    except WebSocketDisconnect:
        await manager.broadcast({"type": "user_left", "count": len(manager.connections)-1})

from fastapi import Request
from fastapi.responses import StreamingResponse
import asyncio

async def event_generator():
    i = 0
    while True:
        await asyncio.sleep(2)
        yield f"data: {{'count': {i}, 'time': '{datetime.utcnow().isoformat()}'}}\n\n"
        i += 1

@app.get("/stream")
async def stream_events(request: Request):
    async def event_stream():
        for event in event_generator():
            if await request.is_disconnected():
                break
            yield event
    return StreamingResponse(event_stream(), media_type="text/event-stream")

import redis
import json
from fastapi import FastAPI

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

@app.post("/publish-sale")
async def publish_sale(sale_data: dict):
    r.publish("sales-channel", json.dumps(sale_data))
    return {"status": "published"}

# در داشبورد فرانت‌اند با Redis subscribe گوش می‌دیم

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
import strawberry
from strawberry.fastapi import GraphQLRouter
import redis
import json
from datetime import datetime

# GraphQL Schema
@strawberry.type
class Query:
    @strawberry.field
    def sales_today(self) -> int:
        return int(r.get("sales_today") or 0)

@strawberry.type
class Subscription:
    @strawberry.subscription
    async def sales_counter(self) -> int:
        pubsub = r.pubsub()
        pubsub.subscribe("sales-channel")
        while True:
            message = pubsub.get_message()
            if message and message['type'] == 'message':
                yield int(r.get("sales_today") or 0)
            await asyncio.sleep(1)

schema = strawberry.Schema(query=Query, subscription=Subscription)
graphql_app = GraphQLRouter(schema)

# FastAPI App
app = FastAPI()
app.include_router(graphql_app, prefix="/graphql")

# Redis + WebSocket برای داشبورد
r = redis.Redis()
manager = ConnectionManager()

@app.websocket("/ws/sales")
async def sales_dashboard(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            await asyncio.sleep(2)
            count = r.get("sales_today") or 0
            await websocket.send_json({
                "type": "sales_update",
                "count": int(count),
                "time": datetime.utcnow().isoformat()
            })
    except WebSocketDisconnect:
        manager.disconnect(websocket)