[FastAPI] FastAPI 개념 뿌시기: ASGI

ASGI

ASGI(Asynchronous Server Gateway Interface)는 비동기 Python 웹 서버, 프레임워크, 애플리케이션 간의 표준 인터페이스이다. WSGI의 후속 버전으로, 비동기 처리와 WebSocket 같은 장시간 연결을 지원한다.

 

1. 주요 기능 및 특징

• 비동기 처리: async/await를 활용한 고성능 웹 애플리케이션
• 프로토콜 지원: HTTP, WebSocket, HTTP/2 등 다양한 프로토콜
• 장시간 연결: 실시간 통신, 스트리밍 지원
• 표준화: 서버와 프레임워크 간 호환성 보장
• 동시성: 많은 수의 동시 연결 효율적 처리
• WSGI 호환: WSGI 앱을 ASGI로 래핑 가능

 

2. 사용 목적

• FastAPI, Starlette 같은 비동기 프레임워크 사용
• WebSocket 실시간 통신 지원
• 높은 동시성이 요구되는 애플리케이션
• SSE(Server-Sent Events), 스트리밍 지원
• 마이크로서비스, API 게이트웨이 구축

 

3. 파이썬 예시 코드

① 기본 ASGI 애플리케이션

# 가장 기본적인 ASGI 애플리케이션
async def app(scope, receive, send):
    """
    scope: 연결 정보 (타입, 경로, 헤더 등)
    receive: 클라이언트로부터 메시지 수신
    send: 클라이언트에게 메시지 전송
    """
    assert scope['type'] == '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, ASGI!',
    })

# Uvicorn으로 실행: uvicorn main:app

 

② ASGI 구조 이해하기

async def detailed_app(scope, receive, send):
    """
    ASGI의 3가지 핵심 파라미터 설명
    """
    
    # 1. scope: 연결에 대한 정보
    print(f"Type: {scope['type']}")  # 'http' or 'websocket' or 'lifespan'
    print(f"Method: {scope.get('method')}")  # GET, POST, etc
    print(f"Path: {scope.get('path')}")  # /users/123
    print(f"Query String: {scope.get('query_string')}")  # b'name=john'
    print(f"Headers: {scope.get('headers')}")  # [(b'host', b'localhost')]
    
    # 2. receive: 클라이언트 메시지 수신
    message = await receive()
    print(f"Received: {message}")
    # HTTP: {'type': 'http.request', 'body': b'...'}
    # WebSocket: {'type': 'websocket.receive', 'text': '...'}
    
    # 3. send: 클라이언트에게 메시지 전송
    await send({
        'type': 'http.response.start',
        'status': 200,
        'headers': [[b'content-type', b'application/json']],
    })
    
    await send({
        'type': 'http.response.body',
        'body': b'{"message": "Hello"}',
    })

③ HTTP 요청 처리

import json

async def http_app(scope, receive, send):
    """HTTP 요청 처리 예시"""
    
    if scope['type'] != 'http':
        return
    
    # 요청 경로에 따른 라우팅
    path = scope['path']
    method = scope['method']
    
    if path == '/' and method == 'GET':
        await handle_home(scope, receive, send)
    elif path == '/api/users' and method == 'GET':
        await handle_users(scope, receive, send)
    elif path == '/api/users' and method == 'POST':
        await handle_create_user(scope, receive, send)
    else:
        await handle_404(scope, receive, send)

async def handle_home(scope, receive, send):
    await send({
        'type': 'http.response.start',
        'status': 200,
        'headers': [[b'content-type', b'text/html']],
    })
    
    await send({
        'type': 'http.response.body',
        'body': b'<h1>Welcome to ASGI App</h1>',
    })

async def handle_users(scope, receive, send):
    users = [
        {"id": 1, "name": "Alice"},
        {"id": 2, "name": "Bob"}
    ]
    
    await send({
        'type': 'http.response.start',
        'status': 200,
        'headers': [[b'content-type', b'application/json']],
    })
    
    await send({
        'type': 'http.response.body',
        'body': json.dumps(users).encode('utf-8'),
    })

async def handle_create_user(scope, receive, send):
    # 요청 바디 읽기
    body = b''
    while True:
        message = await receive()
        body += message.get('body', b'')
        if not message.get('more_body'):
            break
    
    # JSON 파싱
    try:
        data = json.loads(body.decode('utf-8'))
        user = {"id": 3, "name": data.get('name')}
        
        await send({
            'type': 'http.response.start',
            'status': 201,
            'headers': [[b'content-type', b'application/json']],
        })
        
        await send({
            'type': 'http.response.body',
            'body': json.dumps(user).encode('utf-8'),
        })
    except Exception as e:
        await handle_400(scope, receive, send, str(e))

async def handle_404(scope, receive, send):
    await send({
        'type': 'http.response.start',
        'status': 404,
        'headers': [[b'content-type', b'text/plain']],
    })
    
    await send({
        'type': 'http.response.body',
        'body': b'Not Found',
    })

async def handle_400(scope, receive, send, error):
    await send({
        'type': 'http.response.start',
        'status': 400,
        'headers': [[b'content-type', b'application/json']],
    })
    
    await send({
        'type': 'http.response.body',
        'body': json.dumps({"error": error}).encode('utf-8'),
    })

④ WebSocket 처리

async def websocket_app(scope, receive, send):
    """WebSocket 연결 처리"""
    
    if scope['type'] == 'websocket':
        # WebSocket 연결 수락
        await send({
            'type': 'websocket.accept'
        })
        
        # 메시지 송수신 루프
        while True:
            message = await receive()
            
            if message['type'] == 'websocket.receive':
                # 클라이언트로부터 메시지 수신
                text = message.get('text', '')
                
                # 에코 응답
                await send({
                    'type': 'websocket.send',
                    'text': f'Echo: {text}'
                })
            
            elif message['type'] == 'websocket.disconnect':
                # 연결 종료
                print("WebSocket disconnected")
                break
    
    elif 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'Use WebSocket connection',
        })

⑤ 채팅 서버 예시

import json
from typing import Set

# 연결된 클라이언트 관리
connections: Set = set()

async def chat_server(scope, receive, send):
    """간단한 채팅 서버"""
    
    if scope['type'] == 'websocket':
        await send({'type': 'websocket.accept'})
        
        # 현재 연결 추가
        connections.add(send)
        
        try:
            while True:
                message = await receive()
                
                if message['type'] == 'websocket.receive':
                    text = message.get('text', '')
                    data = json.loads(text)
                    
                    # 모든 클라이언트에게 브로드캐스트
                    broadcast_message = json.dumps({
                        'user': data.get('user', 'Anonymous'),
                        'message': data.get('message', '')
                    })
                    
                    for connection in connections:
                        try:
                            await connection({
                                'type': 'websocket.send',
                                'text': broadcast_message
                            })
                        except:
                            pass
                
                elif message['type'] == 'websocket.disconnect':
                    break
        finally:
            # 연결 제거
            connections.discard(send)
    
    elif scope['type'] == 'http':
        # 채팅 클라이언트 HTML 제공
        html = """
        <!DOCTYPE html>
        <html>
        <head><title>Chat</title></head>
        <body>
            <div id="messages"></div>
            <input id="username" placeholder="Username" />
            <input id="message" placeholder="Message" />
            <button onclick="send()">Send</button>
            
            <script>
                const ws = new WebSocket('ws://localhost:8000/ws');
                
                ws.onmessage = (event) => {
                    const data = JSON.parse(event.data);
                    const msg = document.createElement('div');
                    msg.textContent = `${data.user}: ${data.message}`;
                    document.getElementById('messages').appendChild(msg);
                };
                
                function send() {
                    const user = document.getElementById('username').value;
                    const message = document.getElementById('message').value;
                    ws.send(JSON.stringify({user, message}));
                    document.getElementById('message').value = '';
                }
            </script>
        </body>
        </html>
        """
        
        await send({
            'type': 'http.response.start',
            'status': 200,
            'headers': [[b'content-type', b'text/html']],
        })
        await send({
            'type': 'http.response.body',
            'body': html.encode('utf-8'),
        })

⑥ ASGI 미들웨어

class LoggingMiddleware:
    """요청 로깅 미들웨어"""
    
    def __init__(self, app):
        self.app = app
    
    async def __call__(self, scope, receive, send):
        if scope['type'] == 'http':
            print(f"[REQUEST] {scope['method']} {scope['path']}")
        
        await self.app(scope, receive, send)
        
        if scope['type'] == 'http':
            print(f"[RESPONSE] Completed")

class CORSMiddleware:
    """CORS 처리 미들웨어"""
    
    def __init__(self, app):
        self.app = app
    
    async def __call__(self, scope, receive, send):
        if scope['type'] == 'http':
            async def send_with_cors(message):
                if message['type'] == 'http.response.start':
                    headers = list(message.get('headers', []))
                    headers.append([b'access-control-allow-origin', b'*'])
                    message['headers'] = headers
                await send(message)
            
            await self.app(scope, receive, send_with_cors)
        else:
            await self.app(scope, receive, send)

# 미들웨어 적용
async def my_app(scope, receive, send):
    await send({
        'type': 'http.response.start',
        'status': 200,
        'headers': [[b'content-type', b'text/plain']],
    })
    await send({
        'type': 'http.response.body',
        'body': b'Hello with middleware',
    })

# 여러 미들웨어 체이닝
app = LoggingMiddleware(CORSMiddleware(my_app))

⑦ Lifespan 이벤트 처리

import asyncio

async def lifespan_app(scope, receive, send):
    """애플리케이션 생명주기 관리"""
    
    if scope['type'] == 'lifespan':
        while True:
            message = await receive()
            
            if message['type'] == 'lifespan.startup':
                # 시작 시 실행
                print("🚀 애플리케이션 시작")
                # DB 연결, 캐시 초기화 등
                await send({'type': 'lifespan.startup.complete'})
            
            elif message['type'] == 'lifespan.shutdown':
                # 종료 시 실행
                print("🛑 애플리케이션 종료")
                # DB 연결 해제, 리소스 정리 등
                await send({'type': 'lifespan.shutdown.complete'})
                return
    
    elif scope['type'] == 'http':
        await send({
            'type': 'http.response.start',
            'status': 200,
            'headers': [[b'content-type', b'text/plain']],
        })
        await send({
            'type': 'http.response.body',
            'body': b'App is running',
        })

⑧ Server-Sent Events (SSE)

• Server-Sent Events(서버 전송 이벤트)는 서버에서 클라이언트로의 단방향 실시간 데이터 전송 기술이다. 실시간 알림, 주식 정보, 뉴스 피드 등에 활용된다.

import asyncio

async def sse_app(scope, receive, send):
    """Server-Sent Events 스트리밍"""
    
    if scope['type'] == 'http' and scope['path'] == '/events':
        await send({
            'type': 'http.response.start',
            'status': 200,
            'headers': [
                [b'content-type', b'text/event-stream'],
                [b'cache-control', b'no-cache'],
                [b'connection', b'keep-alive'],
            ],
        })
        
        # 주기적으로 이벤트 전송
        for i in range(10):
            data = f"data: Message {i}\n\n"
            await send({
                'type': 'http.response.body',
                'body': data.encode('utf-8'),
                'more_body': True,
            })
            await asyncio.sleep(1)
        
        # 스트림 종료
        await send({
            'type': 'http.response.body',
            'body': b'',
        })
    
    else:
        # SSE 클라이언트 HTML
        html = """
        <!DOCTYPE html>
        <html>
        <head><title>SSE</title></head>
        <body>
            <h1>Server-Sent Events</h1>
            <div id="events"></div>
            
            <script>
                const eventSource = new EventSource('/events');
                eventSource.onmessage = (event) => {
                    const div = document.createElement('div');
                    div.textContent = event.data;
                    document.getElementById('events').appendChild(div);
                };
            </script>
        </body>
        </html>
        """
        
        await send({
            'type': 'http.response.start',
            'status': 200,
            'headers': [[b'content-type', b'text/html']],
        })
        await send({
            'type': 'http.response.body',
            'body': html.encode('utf-8'),
        })

⑨ ASGI vs WSGI 비교

# WSGI (동기)
def wsgi_app(environ, start_response):
    """전통적인 WSGI 애플리케이션"""
    status = '200 OK'
    headers = [('Content-Type', 'text/plain')]
    start_response(status, headers)
    return [b'Hello, WSGI!']

# ASGI (비동기)
async def asgi_app(scope, receive, send):
    """비동기 ASGI 애플리케이션"""
    await send({
        'type': 'http.response.start',
        'status': 200,
        'headers': [[b'content-type', b'text/plain']],
    })
    await send({
        'type': 'http.response.body',
        'body': b'Hello, ASGI!',
    })

# ASGI의 장점:
# 1. 비동기 처리 가능
# 2. WebSocket, HTTP/2 지원
# 3. 장시간 연결 지원
# 4. 더 높은 동시성

⑩ FastAPI가 ASGI를 사용하는 방식

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello"}

# FastAPI는 내부적으로 ASGI 호출 가능 객체로 작동
# app(scope, receive, send) 형태로 호출됨
# Uvicorn이 이 ASGI 인터페이스를 통해 FastAPI와 통신

 

4. ASGI 서버 예시

• Uvicorn: 가장 빠른 ASGI 서버
• Daphne: Django Channels 개발팀 제작
• Hypercorn: HTTP/2 지원