Django 面試精華：ASGI vs WSGI

# 傳統方式：WSGI
gunicorn myproject.wsgi:application

# 新方式：ASGI
uvicorn myproject.asgi:application

# 傳統同步視圖（WSGI）
def get_user_data(request):
    user = User.objects.get(id=request.user.id)  # 數據庫查詢：100ms
    posts = Post.objects.filter(user=user)       # 數據庫查詢：150ms
    comments = Comment.objects.filter(user=user) # 數據庫查詢：120ms
    # 總耗時：370ms（串行執行）

    return JsonResponse({
        'user': user,
        'posts': posts,
        'comments': comments,
    })


# 異步視圖（ASGI）
async def get_user_data(request):
    # 並行執行三個查詢！
    user, posts, comments = await asyncio.gather(
        User.objects.aget(id=request.user.id),     # 100ms
        Post.objects.filter(user=user).all(),      # 150ms
        Comment.objects.filter(user=user).all(),   # 120ms
    )
    # 總耗時：150ms（並行執行，取最慢的那個）

    return JsonResponse({
        'user': user,
        'posts': posts,
        'comments': comments,
    })

# wsgi.py

def application(environ, start_response):
    """
    WSGI 應用的標準簽名

    Args:
        environ: 包含請求資訊的字典（HTTP headers, method, path 等）
        start_response: 用於返回狀態碼和響應頭的回調函數

    Returns:
        可迭代對象（通常是列表），包含響應體
    """
    # 解析請求
    path = environ['PATH_INFO']
    method = environ['REQUEST_METHOD']

    # 設置響應
    status = '200 OK'
    headers = [('Content-Type', 'text/plain')]
    start_response(status, headers)

    # 返回響應體
    return [b'Hello World!']


# Django 的 WSGI 應用
from django.core.wsgi import get_wsgi_application
import os

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
application = get_wsgi_application()

def sync_view(request):
    # 1. 查詢數據庫（阻塞 100ms）
    user = User.objects.get(id=1)  # 線程等待...

    # 2. 調用外部 API（阻塞 500ms）
    response = requests.get('https://api.example.com/data')  # 線程等待...

    # 3. 讀取文件（阻塞 50ms）
    with open('data.txt') as f:  # 線程等待...
        data = f.read()

    # 總耗時：650ms
    # 在這 650ms 內，Worker 線程被占用，無法處理其他請求
    return HttpResponse('OK')

# asgi.py

async def application(scope, receive, send):
    """
    ASGI 應用的標準簽名

    Args:
        scope: 包含連接資訊的字典（類似 WSGI 的 environ）
        receive: 異步可調用對象，用於接收事件（請求體）
        send: 異步可調用對象，用於發送事件（響應）
    """
    # 檢查連接類型
    if scope['type'] == 'http':
        # HTTP 請求
        await send({
            'type': 'http.response.start',
            'status': 200,
            'headers': [[b'content-type', b'text/plain']],
        })
        await send({
            'type': 'http.response.body',
            'body': b'Hello World!',
        })

    elif scope['type'] == 'websocket':
        # WebSocket 連接
        await send({'type': 'websocket.accept'})
        message = await receive()
        await send({
            'type': 'websocket.send',
            'text': 'Hello WebSocket!',
        })


# Django 的 ASGI 應用
from django.core.asgi import get_asgi_application
import os

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
application = get_asgi_application()

async def async_view(request):
    # 1. 異步查詢數據庫（不阻塞，100ms）
    user = await User.objects.aget(id=1)  # 切換到其他協程

    # 2. 異步調用外部 API（不阻塞，500ms）
    async with aiohttp.ClientSession() as session:
        async with session.get('https://api.example.com/data') as response:
            data = await response.json()  # 切換到其他協程

    # 3. 異步讀取文件（不阻塞，50ms）
    async with aiofiles.open('data.txt') as f:
        content = await f.read()  # 切換到其他協程

    # 總耗時：650ms
    # 但在這 650ms 內，Worker 可以處理其他請求！
    return HttpResponse('OK')

# WSGI 同步視圖
def wsgi_view(request):
    user = User.objects.get(id=1)        # 100ms
    posts = Post.objects.filter(user=user).all()  # 200ms
    return JsonResponse({'user': user, 'posts': posts})
    # 總耗時：300ms
    # Worker 被占用：300ms


# ASGI 異步視圖
async def asgi_view(request):
    user, posts = await asyncio.gather(
        User.objects.aget(id=1),          # 100ms
        Post.objects.filter(user=user).all(),  # 200ms
    )
    return JsonResponse({'user': user, 'posts': posts})
    # 總耗時：200ms（並行執行）
    # Worker 實際占用時間：< 1ms（其餘時間可處理其他請求）

# WSGI 同步視圖
def wsgi_cpu_intensive(request):
    result = complex_calculation()  # 純 CPU 計算，1 秒
    return JsonResponse({'result': result})
    # 總耗時：1 秒


# ASGI 異步視圖（並無優勢）
async def asgi_cpu_intensive(request):
    result = complex_calculation()  # 仍然阻塞事件循環，1 秒
    return JsonResponse({'result': result})
    # 總耗時：1 秒
    # ⚠️ 而且阻塞了事件循環，影響其他協程！

# 只能在 ASGI 下運行
from channels.generic.websocket import AsyncWebsocketConsumer
import json

class ChatConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        await self.accept()

    async def receive(self, text_data):
        data = json.loads(text_data)
        message = data['message']

        # 發送消息給客戶端
        await self.send(text_data=json.dumps({
            'message': message
        }))

    async def disconnect(self, close_code):
        pass

# 開發環境
python manage.py runserver  # Django 開發服務器（WSGI）

# 生產環境
gunicorn myproject.wsgi:application --workers 4

# 開發環境
python manage.py runserver  # Django 3.0+ 自動檢測並使用 ASGI

# 生產環境
uvicorn myproject.asgi:application --workers 2
# 或
daphne myproject.asgi:application
# 或（推薦）
gunicorn myproject.asgi:application -k uvicorn.workers.UvicornWorker --workers 2

# 生產環境最佳實踐
gunicorn myproject.asgi:application \
    --bind 0.0.0.0:8000 \
    --workers 2 \
    --worker-class uvicorn.workers.UvicornWorker \
    --access-logfile - \
    --error-logfile -

# 傳統 CMS 系統
def article_list(request):
    articles = Article.objects.filter(published=True)
    return render(request, 'articles.html', {'articles': articles})

# CPU 密集型任務
def generate_report(request):
    data = Report.objects.all()
    pdf = generate_pdf(data)  # CPU 密集型，500ms
    return FileResponse(pdf)

# 聚合多個微服務數據
async def dashboard(request):
    user, orders, notifications, analytics = await asyncio.gather(
        fetch_user_service(request.user.id),
        fetch_order_service(request.user.id),
        fetch_notification_service(request.user.id),
        fetch_analytics_service(request.user.id),
    )
    # 4 個服務調用並行執行，總耗時取最慢的那個

    return JsonResponse({
        'user': user,
        'orders': orders,
        'notifications': notifications,
        'analytics': analytics,
    })


# WebSocket 聊天
from channels.generic.websocket import AsyncWebsocketConsumer

class ChatConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        await self.channel_layer.group_add("chat", self.channel_name)
        await self.accept()

    async def receive(self, text_data):
        await self.channel_layer.group_send("chat", {
            'type': 'chat_message',
            'message': text_data
        })

upstream asgi_backend {
    server 127.0.0.1:8000;  # Uvicorn
}

upstream wsgi_backend {
    server 127.0.0.1:8001;  # Gunicorn
}

server {
    listen 80;

    # ASGI 路由
    location /api/ {
        proxy_pass http://asgi_backend;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
    }

    location /ws/ {
        proxy_pass http://asgi_backend;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
    }

    # WSGI 路由
    location / {
        proxy_pass http://wsgi_backend;
    }
}

# 同一個 Django 項目
myproject/
├── wsgi.py   # WSGI 入口
└── asgi.py   # ASGI 入口

# 可以選擇不同的啟動方式
gunicorn myproject.wsgi:application  # WSGI
uvicorn myproject.asgi:application   # ASGI

# 同步視圖在 WSGI 和 ASGI 下都能運行
def sync_view(request):
    return HttpResponse("OK")

# 異步視圖只能在 ASGI 下運行
async def async_view(request):
    return HttpResponse("OK")

async def async_view(request):
    # 大量 I/O 操作
    data1 = await fetch_api1()  # 100ms
    data2 = await fetch_api2()  # 100ms
    data3 = await fetch_api3()  # 100ms
    # 並行執行，總耗時 100ms

def sync_view(request):
    # CPU 密集型計算
    result = complex_calculation()  # 1 秒
    # WSGI: 1 秒
    # ASGI: 1 秒 + 事件循環開銷，而且阻塞其他協程