FastAPIの裏側：ASGIとルーティングの仕組みを解説

スクラッチからシンプルなFastAPIを構築する：ASGIとコアルーティングの理解

はじめに：なぜ車輪の再発明をするのか？

Pythonの非同期Webフレームワークについて語るとき、FastAPIは間違いなく近年で最も輝かしい星です。その優れたパフォーマンス、自動APIドキュメント生成、および型ヒントのサポートで広く称賛されています。しかし、この強力なフレームワークの背後にはどのような魔法が隠されているのか疑問に思ったことはありませんか？

今日は、ASGIプロトコルとルーティングシステムの2つのコアコンセプトの理解に焦点を当てて、FastAPIの簡略化されたバージョンをスクラッチから構築します。自分の手で構築することで、最新の非同期Webフレームワークの動作原理を把握できます。これはFastAPIをより良く使用するのに役立つだけでなく、問題が発生したときに根本原因をすばやく特定できるようになります。

ASGIとは？ WSGIより優れている理由

コーディングを開始する前に、FastAPIが高パフォーマンスの非同期処理を実現するための基盤であるASGI（Asynchronous Server Gateway Interface）を理解する必要があります。

WSGIの制限

DjangoまたはFlaskを使用したことがある場合は、WSGI（Web Server Gateway Interface）について聞いたことがあるかもしれません。 WSGIは、Python Webアプリケーションとサーバー間の同期インターフェース仕様ですが、明らかな欠点があります。

• 一度に1つのリクエストしか処理できず、同時実行性がない
• 長期間の接続（WebSocketなど）をサポートしない
• 非同期I/Oの利点を十分に活用できない

ASGIの利点

ASGIは、これらの問題を解決するために作成されました。

• 完全に非同期で、複数のリクエストの同時処理をサポート
• WebSocketおよびHTTP/2と互換性がある
• ミドルウェアが非同期環境で動作することを許可
• リクエストライフサイクル全体で非同期イベントをサポート

簡単に言うと、ASGIは、非同期Webアプリケーションがサーバー（Uvicornなど）と通信できるようにする標準インターフェースを定義します。次に、最小限のASGIサーバーを実装します。

ステップ1：基本的なASGIサーバーを実装する

ASGIアプリケーションは、基本的に、スコープ、受信、送信の3つのパラメーターを受け取る呼び出し可能なオブジェクト（関数またはクラス）です。

# 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 = {
            "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()))
        
        # 受信および送信メソッドの定義
        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の最も直感的な機能の1つは、次のようなエレガントなルート定義です。

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

同様のルーティング機能を実装しましょう。

ルーティングコアコンポーネントの設計

3つのコアコンポーネントが必要です。

• ルーター：すべてのルーティングルールを管理
• デコレーター：@get、@postなど、ルートを登録するため
• パスのマッチング：動的なパスパラメーター（/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
    
    # ショートカットメソッド
    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のハイライトの1つは、自動パラメーター解析と型変換です。この機能を実装しましょう。

# ルーターの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サポート）の実装

次に、POSTリクエストボディのサポートを追加し、JSONデータの解析を有効にします。

# ルーターにリクエストボディの解析を追加
async def handle(self, scope: Dict[str, Any], receive: Callable) -> Dict[str, Any]:
    # ... 以前のコード ...
    
    # POSTリクエストの場合、リクエストボディを解析
    request_body = {}
    if method == "POST":
        # 受信からリクエストボディを取得
        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にアクセス → パラメーター付きの応答を取得
• {