FastAPI가 내부적으로 어떻게 작동하는가: ASGI와 라우팅 설명

---

스크래치부터 간단한 FastAPI 구축: ASGI와 핵심 라우팅 이해

소개: 왜 이 바퀴를 재발명하는가?

파이썬 비동기 웹 프레임워크에 대해 이야기할 때, FastAPI는 의심할 여지 없이 최근 몇 년간 가장 밝게 빛나는 별입니다. 이 프레임워크는 뛰어난 성능, 자동 API 문서 생성, 타입 힌트 지원으로 널리 인정받고 있습니다. 하지만 이 강력한 프레임워크 뒤에 어떤 마법이 숨어 있는지 궁금해 본 적이 있나요?

오늘 우리는 ASGI 프로토콜과 라우팅 시스템이라는 두 가지 핵심 개념을 이해하는 데 초점을 맞춰 FastAPI의 단순화된 버전을 스크래치부터 구축할 것입니다. 우리 손으로 직접 구축함으로써 현대 비동기 웹 프레임워크의 작동 원리를 파악하게 될 것입니다. 이는 FastAPI를 더 잘 사용하는 데 도움이 될 뿐만 아니라 문제가 발생했을 때 근본 원인을 신속하게 식별할 수 있게 해줍니다.

ASGI란 무엇인가? WSGI보다 왜 더 발전했는가?

코딩을 시작하기 전에 ASGI(Asynchronous Server Gateway Interface)를 이해해야 합니다. ASGI는 FastAPI가 고성능 비동기 처리를 달성할 수 있도록 하는 기반입니다.

WSGI의 제한 사항

Django나 Flask를 사용해 본 적이 있다면 WSGI(Web Server Gateway Interface)에 대해 들어본 적이 있을 것입니다. WSGI는 파이썬 웹 애플리케이션과 서버 간의 동기 인터페이스 사양이지만 다음과 같은 명백한 결함이 있습니다.

• 한 번에 하나의 요청만 처리할 수 있으며 동시성이 없습니다.
• 장기 연결(예: WebSocket)을 지원하지 않습니다.
• 비동기 I/O의 장점을 최대한 활용할 수 없습니다.

ASGI의 장점

ASGI는 이러한 문제를 해결하기 위해 만들어졌습니다.

• 완전히 비동기적이며 여러 요청의 동시 처리를 지원합니다.
• WebSocket 및 HTTP/2와 호환됩니다.
• 미들웨어가 비동기 환경에서 작동할 수 있도록 합니다.
• 요청 수명 주기 전반에 걸쳐 비동기 이벤트를 지원합니다.

간단히 말해서 ASGI는 비동기 웹 애플리케이션이 서버(예: Uvicorn)와 통신할 수 있도록 하는 표준 인터페이스를 정의합니다. 다음으로 최소한의 ASGI 서버를 구현할 것입니다.

1단계: 기본 ASGI 서버 구현

ASGI 애플리케이션은 기본적으로 scope, receive, send의 세 가지 매개변수를 받는 호출 가능한 객체(함수 또는 클래스)입니다.

# asgi_server.py
import socket
import asyncio
import json
from typing import Callable, Awaitable, Dict, Any

# ASGI 애플리케이션 타입 정의
ASGIApp = Callable[[Dict[str, Any], Callable[[], Awaitable[Dict]]], Awaitable[None]]

class ASGIServer:
    def __init__(self, host: str = "127.0.0.1", port: int = 8000):
        self.host = host
        self.port = port
        self.app: ASGIApp = self.default_app  # 기본 어플리케이션
    
    async def default_app(self, scope: Dict[str, Any], receive: Callable, send: Callable):
        """기본 어플리케이션: 404 응답 반환"""
        if scope["type"] == "http":
            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_connection(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
        """새로운 연결을 처리하고, HTTP 요청을 파싱하여 ASGI 어플리케이션에 전달"""
        data = await reader.read(1024)
        request = data.decode().split("\r\n")
        method, path, _ = request[0].split()
        
        # ASGI scope 생성
        scope = {
            "type": "http",
            "method": method,
            "path": path,
            "headers": []
        }
        
        # 요청 헤더 파싱
        for line in request[1:]:
            if line == "":
                break
            key, value = line.split(":", 1)
            scope["headers"].append((key.strip().lower().encode(), value.strip().encode()))
        
        # receive 및 send 메서드 정의
        async def receive() -> Dict:
            """메시지 수신 시뮬레이션 (간소화 버전)"""
            return {"type": "http.request", "body": b""}
        
        async def send(message: Dict):
            """클라이언트에 응답 전송"""
            if message["type"] == "http.response.start":
                status = message["status"]
                status_line = f"HTTP/1.1 {status} OK\r\n"
                headers = "".join([f"{k.decode()}: {v.decode()}\r\n" for k, v in message["headers"]])
                writer.write(f"{status_line}{headers}\r\n".encode())
            
            if message["type"] == "http.response.body":
                writer.write(message["body"])
                await writer.drain()
                writer.close()
        
        # ASGI 애플리케이션 호출
        await self.app(scope, receive, send)
    
    async def run(self):
        """서버 시작"""
        server = await asyncio.start_server(
            self.handle_connection, self.host, self.port
        )
        print(f"Server running on http://{self.host}:{self.port}")
        async with server:
            await server.serve_forever()

# 서버 실행
if __name__ == "__main__":
    server = ASGIServer()
    asyncio.run(server.run())

이 단순화된 ASGI 서버는 기본적인 HTTP 요청을 처리하고 응답을 반환할 수 있습니다. 스크립트를 실행한 후 http://127.0.0.1:8000을 방문하여 테스트해 보세요. 아직 라우트를 정의하지 않았기 때문에 "Not Found"가 표시됩니다.

2단계: 라우팅 시스템 구현

FastAPI의 가장 직관적인 기능 중 하나는 다음과 같은 우아한 라우트 정의입니다.

@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

유사한 라우팅 기능을 구현해 보겠습니다.

라우팅 핵심 구성 요소 설계

세 가지 핵심 구성 요소가 필요합니다.

• Router: 모든 라우팅 규칙 관리
• Decorators: @get, @post 등 라우트 등록용
• Path matching: 동적 경로 매개변수 처리(예: /items/{item_id})

# router.py
from typing import Callable, Awaitable, Dict, Any, List, Tuple, Pattern
import re
from functools import wraps

# 라우트 타입 정의
RouteHandler = Callable[[Dict[str, Any]], Awaitable[Dict[str, Any]]]

class Route:
    def __init__(self, path: str, methods: List[str], handler: RouteHandler):
        self.path = path
        self.methods = [m.upper() for m in methods]
        self.handler = handler
        self.path_pattern, self.param_names = self.compile_path(path)
    
    def compile_path(self, path: str) -> Tuple[Pattern, List[str]]:
        """경로를 정규 표현식으로 변환하고 매개변수 이름 추출"""
        param_names = []
        pattern = re.sub(r"{([\w]+)}", lambda m: (param_names.append(m.group(1)), r"(\\w+)")[1], path)
        return re.compile(f"^{pattern}$"), param_names
    
    def match(self, path: str, method: str) -> Tuple[bool, Dict[str, Any]]:
        """경로 및 메서드 일치, 매개변수 반환"""
        if method not in self.methods:
            return False, {}
        
        match = self.path_pattern.match(path)
        if not match:
            return False, {}
        
        params = dict(zip(self.param_names, match.groups()))
        return True, params

class Router:
    def __init__(self):
        self.routes: List[Route] = []
    
    def add_route(self, path: str, methods: List[str], handler: RouteHandler):
        """라우트 추가"""
        self.routes.append(Route(path, methods, handler))
    
    def route(self, path: str, methods: List[str]):
        """라우트 데코레이터"""
        def decorator(handler: RouteHandler):
            self.add_route(path, methods, handler)
            @wraps(handler)
            async def wrapper(*args, **kwargs):
                return await handler(*args, **kwargs)
            return wrapper
        return decorator
    
    # Shortcut 메서드
    def get(self, path: str):
        return self.route(path, ["GET"])
    
    def post(self, path: str):
        return self.route(path, ["POST"])
    
    async def handle(self, scope: Dict[str, Any], receive: Callable) -> Dict[str, Any]:
        """요청 처리, 일치하는 라우트 찾기 및 실행"""
        path = scope["path"]
        method = scope["method"]
        
        for route in self.routes:
            matched, params = route.match(path, method)
            if matched:
                # 쿼리 매개변수 파싱
                query_params = self.parse_query_params(scope)
                # 경로 매개변수와 쿼리 매개변수 병합
                request_data = {** params, **query_params}
                # 핸들러 함수 호출
                return await route.handler(request_data)
        
        # 라우트를 찾을 수 없음
        return {"status": 404, "body": {"detail": "Not Found"}}
    
    def parse_query_params(self, scope: Dict[str, Any]) -> Dict[str, Any]:
        """쿼리 매개변수 파싱 (단순화 버전)"""
        # 실제 ASGI에서 쿼리 매개변수는 scope["query_string"]에 있습니다.
        query_string = scope.get("query_string", b"").decode()
        params = {}
        if query_string:
            for pair in query_string.split("&"):
                if "=" in pair:
                    key, value = pair.split("=", 1)
                    params[key] = value
        return params

라우팅을 ASGI 서버와 통합

이제 라우팅 시스템을 사용하도록 ASGI 서버를 수정해야 합니다.

# ASGIServer 클래스에 라우팅 지원 추가
class ASGIServer:
    def __init__(self, host: str = "127.0.0.1", port: int = 8000):
        self.host = host
        self.port = port
        self.router = Router()  # 라우터 인스턴스화
        self.app = self.asgi_app  # 라우팅 활성화 ASGI 애플리케이션 사용
    
    async def asgi_app(self, scope: Dict[str, Any], receive: Callable, send: Callable):
        """라우팅 기능이 있는 ASGI 어플리케이션"""
        if scope["type"] == "http":
            # 요청 처리
            response = await self.router.handle(scope, receive)
            status = response.get("status", 200)
            body = json.dumps(response.get("body", {})).encode()
            
            # 응답 전송
            await send({
                "type": "http.response.start",
                "status": status,
                "headers": [(b"content-type", b"application/json")]
            })
            await send({
                "type": "http.response.body",
                "body": body
            })

3단계: 매개변수 파싱 및 타입 변환 구현

FastAPI의 하이라이트 중 하나는 자동 매개변수 파싱 및 타입 변환입니다. 이 기능을 구현해 보겠습니다.

# Router의 handle 메서드에 타입 변환 추가
async def handle(self, scope: Dict[str, Any], receive: Callable) -> Dict[str, Any]:
    # ... 이전 코드 ...
    
    if matched:
        # 쿼리 매개변수 파싱
        query_params = self.parse_query_params(scope)
        # 경로 매개변수와 쿼리 매개변수 병합
        raw_data = {** params, **query_params}
        
        # 핸들러 함수에서 매개변수 타입 어노테이션 가져오기
        handler_params = route.handler.__annotations__
        
        # 타입 변환
        request_data = {}
        for key, value in raw_data.items():
            if key in handler_params:
                target_type = handler_params[key]
                try:
                    # 타입 변환 시도
                    request_data[key] = target_type(value)
                except (ValueError, TypeError):
                    return {
                        "status": 400,
                        "body": {"detail": f"Invalid type for {key}, expected {target_type}"}
                    }
            else:
                request_data[key] = value
        
        # 핸들러 함수 호출
        return await route.handler(request_data)

이제 프레임워크는 함수 어노테이션에 지정된 타입으로 매개변수를 자동으로 변환할 수 있습니다!

4단계: 요청 바디 파싱 구현 (POST 지원)

다음으로 JSON 데이터 파싱을 가능하게 하여 POST 요청 바디에 대한 지원을 추가합니다.

# Router에 요청 바디 파싱 추가
async def handle(self, scope: Dict[str, Any], receive: Callable) -> Dict[str, Any]:
    # ... 이전 코드 ...
    
    # POST 요청인 경우 요청 바디 파싱
    request_body = {}
    if method == "POST":
        # receive에서 요청 바디 가져오기
        message = await receive()
        if message["type"] == "http.request" and "body" in message:
            try:
                request_body = json.loads(message["body"].decode())
            except json.JSONDecodeError:
                return {
                    "status": 400,
                    "body": {"detail": "Invalid JSON"}
                }
    
    # 모든 매개변수 병합
    raw_data = {** params, **query_params,** request_body}
    # ... 타입 변환 및 핸들러 함수 호출 ...

5단계: 완전한 예제 애플리케이션 구축

이제 FastAPI와 마찬가지로 프레임워크를 사용할 수 있습니다.

# main.py
from asgi_server import ASGIServer
import asyncio

# 서버 인스턴스르 생성합니다 (라우터 포함)
app = ASGIServer()
router = app.router

# 라우트 정의
@router.get("/")
async def root():
    return {"message": "Hello, World!"}

@router.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

@router.post("/items/")
async def create_item(name: str, price: float):
    return {"item": {"name": name, "price": price, "id": 42}}

# 어플리케이션 실행
if __name__ == "__main__":
    asyncio.run(app.run())

이 애플리케이션을 테스트합니다.

• http://127.0.0.1:8000 방문 → 환영 메시지 받기
• http://127.0.0.1:8000/items/42?q=test 방문 → 매개변수가 있는 응답 받기
• {