純PythonでFastAPI風フレームワークを自作する


[![Group259.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/4075813/cd9aa3b8-fbbb-45a9-8c56-ee54e75d3970.png)](https://leapcell.io/)


**[Leapcell: The Best of Serverless Web Hosting](https://leapcell.io/)**

# WSGIからFastAPI風のルーティング仕組みを実装する  

PythonのWeb開発の世界で、FastAPIは効率的で簡潔なルーティングデザインと強力な機能で開発者に広く愛用されています。FastAPIはASGI（Asynchronous Server Gateway Interface）プロトコルをベースに構築されており、従来のWSGI（Web Server Gateway Interface）とは異なります。本稿では、WSGIから始まり、FastAPIに似たルーティング仕組みを実装する方法を探りながら、WSGIやUvicornなどのキーコンセプトとそれらの関係を深く分析していきます。  


## I. WSGI、ASGI、Uvicornの概念分析  

### 1.1 WSGI: Web Server Gateway Interface  
WSGIはPythonのWeb開発における標準インターフェースで、WebサーバとPythonのWebアプリケーション間の通信仕様を定めています。その登場により、異なるWebサーバとWebフレームワーク間の互換性の問題が解決され、開発者はWebサーバやフレームワークを自由に選択して動作させることができるようになりました。  

WSGIでは、Webアプリケーションを呼び出し可能なオブジェクトとして抽象化しており、このオブジェクトは2つのパラメータを受け取ります。1つはリクエスト情報を含む辞書（`environ`）、もう1つはレスポンスのステータスコードとヘッダー情報を送信するためのコールバック関数（`start_response`）です。例えば、以下のようなシンプルなWSGIアプリケーションを作成することができます：  
```python  
def simple_app(environ, start_response):  
    status = '200 OK'  
    response_headers = [('Content-type', 'text/plain')]  
    start_response(status, response_headers)  
    return [b'Hello, World!']  
```  
この例では、`environ`にはリクエストメソッド、URL、HTTPヘッダーなどのリクエスト関連の情報が含まれています。`start_response`関数はレスポンスのステータスコードとヘッダー情報を設定するために使用され、関数の返り値がレスポンスのボディとなります。  

### 1.2 ASGI: Asynchronous Server Gateway Interface  
Pythonにおける非同期プログラミングの台頭、特に`asyncio`ライブラリの普及に伴い、従来のWSGIでは非同期I/O操作に対応することができませんでした。そこでASGIが登場しました。ASGIはWSGIの拡張であり、同期アプリケーションをサポートするだけでなく、非同期リクエストを効率的に処理することができ、現代のWebアプリケーションが求める高性能・高並行処理のニーズに応えるものとなっています。  

ASGIもまた、サーバとアプリケーション間の通信プロトコルを定めていますが、そのアプリケーションの呼び出し可能オブジェクトは非同期メソッドをサポートしています。以下に簡単なASGIアプリケーションの例を示します：  
```python  
async def simple_asgi_app(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!'  
    })  
```  
ここで、`scope`にはリクエストの種類（例えば`http`）などの情報が含まれています。`receive`はクライアントから送信されたデータを受け取るために使用され、`send`はクライアントにレスポンスデータを送信するためのもので、全体の処理プロセスは非同期操作をサポートしています。  

### 1.3 Uvicorn: ASGIサーバ  
UvicornはPythonベースの高性能ASGIサーバであり、ASGIアプリケーションを実行することができるほか、WSGIアプリケーションにも互換性を持っています。Uvicornは`uvloop`や`httptools`などのライブラリを使用して効率的なイベントループとHTTPパースを実現しており、ASGIの非同期メリットを最大限に活かし、優れたパフォーマンスを提供します。  

実際の開発では、Uvicornを用いてASGIまたはWSGIアプリケーションを起動することができます。例えば、上記のASGIアプリケーションを起動するには、以下のコマンドを実行します：  
```bash  
uvicorn main:app --reload  
```  
ここで、`main`はアプリケーションコードを含むPythonモジュールの名前で、`app`は当該モジュール内で定義されたASGIアプリケーションオブジェクトです。`--reload`パラメータは、コードが変更された際に自動的にアプリケーションを再読み込みするためのものです。  


## II. WSGIからFastAPI風のルーティング仕組みを実装する  

### 2.1 ルーティングシステムの基本原理  
FastAPIのルーティングシステムでは、デコレータを用いて関数を特定のURLパスにバインドしています。対応するパスにマッチするリクエストが届くと、対応する処理関数が呼び出され、レスポンスが返されます。WSGIで同様のルーティングシステムを実装する核心的なアイデアは、リクエストのURLパスに基づいて対応する処理関数を見つけ、それを呼び出してレスポンスを生成することです。  

まず、URLパスと処理関数のマッピングを格納するためのルーティングテーブルを定義します。リクエストが届くと、ルーティングテーブルで一致するパスを検索し、対応する処理関数を呼び出してレスポンスを生成します。  

### 2.2 簡易WSGIルーティングシステムの実装  
以下では、段階的に簡易なWSGIルーティングシステムを実装し、FastAPIのルーティング機能を模倣していきます。  

まず、空のルーティングテーブルを定義します：  
```python  
route_table = {}  
```  

次に、関数をURLパスにバインドし、ルーティングテーブルに追加するためのデコレータを作成します：  
```python  
def route(path):  
    def decorator(func):  
        route_table[path] = func  
        return func  
    return decorator  
```  

次に、リクエストパスに基づいて対応する処理関数を検索し、呼び出すWSGIアプリケーションを実装します：  
```python  
def wsgi_app(environ, start_response):  
    path = environ.get('PATH_INFO', '/')  
    if path in route_table:  
        response_body = route_table[path]()  
        status = '200 OK'  
    else:  
        response_body = [b'404 Not Found']  
        status = '404 Not Found'  

    response_headers = [('Content-type', 'text/plain')]  
    start_response(status, response_headers)  
    return response_body  
```  

これで、定義したルーティングデコレータを用いて処理関数を定義することができます：  
```python  
@route('/')  
def index():  
    return [b'Welcome to the index page!']  

@route('/about')  
def about():  
    return [b'This is the about page.']  
```  

最後に、Uvicornを用いてこのWSGIアプリケーションを実行します：  
```bash  
uvicorn main:wsgi_app --reload  
```  

上記のステップにより、異なるURLパスに応じて対応するコンテンツを返す簡易WSGIルーティングシステムが実装され、FastAPIのルーティング機能を初步的に模倣することができました。  

### 2.3 ルーティングシステムの改良  
上記のルーティングシステムは基本バージョンに過ぎず、多くの課題が残されています。例えば、動的ルーティング（パラメータ化されたパスなど）やリクエストメソッドの処理に対応していません。以下では、これらの点を改良していきます。  

#### 動的ルーティングのサポート  
動的ルーティングをサポートするために、正規表現を用いてパスをマッチングし、パスからパラメータを抽出することができます。`re`モジュールをインポートし、ルーティングテーブルとルーティングデコレータを修正します：  
```python  
import re  

route_table = {}  

def route(path):  
    def decorator(func):  
        route_table[path] = func  
        return func  
    return decorator  

def dynamic_route(path_pattern):  
    def decorator(func):  
        route_table[path_pattern] = func  
        return func  
    return decorator  
```  

同時に、WSGIアプリケーションを修正して動的ルーティングを処理できるようにします：  
```python  
def wsgi_app(environ, start_response):  
    path = environ.get('PATH_INFO', '/')  
    for pattern, handler in route_table.items():  
        if isinstance(pattern, str):  
            if path == pattern:  
                response_body = handler()  
                status = '200 OK'  
                break  
        else:  
            match = re.match(pattern, path)  
            if match:  
                args = match.groups()  
                response_body = handler(*args)  
                status = '200 OK'  
                break  
    else:  
        response_body = [b'404 Not Found']  
        status = '404 Not Found'  

    response_headers = [('Content-type', 'text/plain')]  
    start_response(status, response_headers)  
    return response_body  
```  

これで、動的ルーティングの処理関数を定義することができます：  
```python  
@dynamic_route(r'/user/(\d+)')  
def user_detail(user_id):  
    return [f'User {user_id} detail page.'.encode()]  
```  

#### リクエストメソッドのサポート  
異なるリクエストメソッド（GETやPOSTなど）をサポートするために、各パスに対応する異なるリクエストメソッドの処理関数をルーティングテーブルに格納する必要があります。ルーティングテーブルとルーティングデコレータを以下のように修正します：  
```python  
route_table = {}  

def route(path, methods=['GET']):  
    def decorator(func):  
        if path not in route_table:  
            route_table[path] = {}  
        for method in methods:  
            route_table[path][method] = func  
        return func  
    return decorator  
```  

同時に、WSGIアプリケーションを修正して、リクエストメソッドに基づいて対応する処理関数を呼び出せるようにします：  
```python  
def wsgi_app(environ, start_response):  
    path = environ.get('PATH_INFO', '/')  
    method = environ.get('REQUEST_METHOD', 'GET')  
    if path in route_table and method in route_table[path]:  
        response_body = route_table[path][method]()  
        status = '200 OK'  
    else:  
        response_body = [b'404 Not Found']  
        status = '404 Not Found'  

    response_headers = [('Content-type', 'text/plain')]  
    start_response(status, response_headers)  
    return response_body  
```  

これで、異なるリクエストメソッドをサポートする処理関数を定義することができます：  
```python  
@route('/login', methods=['GET', 'POST'])  
def login():  
    return [b'Login page.']  
```  

以上の改良により、WSGIルーティングシステムはより完成度の高いものとなり、動的ルーティングや異なるリクエストメソッドをサポートするようになり、FastAPIのルーティング機能に一層近づいたものとなりました。  



## III. まとめと展望
本稿では、WSGIの基本概念から始まり、ASGIやUvicornに関する知識を紹介しながら、段階的に簡易WSGIルーティングシステムを実装することでFastAPIのルーティング機能を模倣するプロセスを示しました。この過程を通じて、Webサーバとアプリケーション間の通信プロトコルや、ルーティングシステムの核心原理を深く理解することができました。  

実装したWSGIルーティングシステムはまだFastAPIと比較すると機能が乏しいものの、ルーティングシステムの実装の考え方を明確に示すものとなりました。今後は、このルーティングシステムをさらに改良していくことが考えられます。例えば、リクエストボディの解析機能やレスポンスボディのシリアライズ機能を追加することで、実際の本番環境で使用されるWebフレームワークにより近づけることが可能です。また、このルーティングシステムをASGI環境に拡張し、非同期プログラミングのメリットを最大限に活かすことで、アプリケーションのパフォーマンスと並行処理能力を向上させることも検討できます。  

以上の内容で、WSGIを用いたルーティング仕組みの実装プロセスを詳細に説明しました。もし特定の部分をさらに詳しく展開したい場合や、新しい機能を追加したいと考えられましたら、お気軽にお知らせください。  


# [Leapcell: The Best of Serverless Web Hosting](https://leapcell.io/)  

**Pythonサービスのデプロイに最適なプラットフォーム：[Leapcell](https://leapcell.io/)**  

[![brandpic7.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/4075813/f6efbbc8-d2a3-4294-b10a-706c3c0aa445.png)](https://leapcell.io/)


### 🚀 好きな言語で開発  
JavaScript、Python、Go、Rustで自由自在に開発できます。  

### 🌍 無料で無制限のプロジェクトをデプロイ  
使用分のみ課金—リクエストがなければ一切請求されません。  

### ⚡ 従量課金制、隠れた料金はありません  
アイドル料金など一切なく、シームレスにスケーリング可能です。  

[![Frame3-withpadding2x.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/4075813/fc58bcbf-8e23-45dd-a69e-95e546d9ddc8.png)](https://leapcell.io/)


[📖 ドキュメントを確認する](https://docs.leapcell.io/)  

🔹 Twitterでフォロー：[@LeapcellHQ](https://x.com/LeapcellHQ)