StarletteとPydantic: FastAPIなしで強力なAPIを構築する

Pythonのウェブ開発の分野では、FastAPIはその簡潔でユーザーフレンドリーな機能により、多くの注目を集めています。しかし、実際には、これは単にStarletteとPydanticの高レベルなカプセル化にすぎません。公式Starletteウェブサイトには豊富な機能が紹介されており、Pydanticはその強力なデータ検証機能で知られています。これらの2つのライブラリを直接使用することで、開発者はFastAPIのカプセル化に頼らずに、柔軟に高性能なAPIを構築できます。次に、両方のコア機能と特徴を組み合わせることで、詳細に説明します。

1. Starletteのコア機能と例

1.1 非同期リクエスト処理

StarletteはASGI標準に基づいており、非同期タスクを効率的に処理できます。FastAPIでの記述方法と比較すると、その基盤となるロジックがより明確に反映されます。

FastAPIの例

from fastapi import FastAPI
import asyncio

# FastAPIアプリケーションインスタンスを作成
app = FastAPI()

# デコレータを使用してGETリクエストルートを定義します。この関数は非同期関数であり、他のリクエストをブロックせずに時間のかかる操作を処理できます
@app.get("/async_items/")
async def async_read_items():
    await asyncio.sleep(1)  # I/O操作をシミュレートし、1秒間一時停止します
    return {"message": "FastAPI非同期処理の例"}

Starletteの例

from starlette.applications import Starlette
from starlette.responses import JSONResponse
import asyncio

# Starletteアプリケーションインスタンスを作成
app = Starlette()

# アプリケーションインスタンスに直接ルートを定義し、パスとリクエストメソッドを指定します。処理関数は非同期関数です
@app.route("/async_items/", methods=["GET"])
async def async_read_items(request):
    await asyncio.sleep(1)  # I/O操作をシミュレートし、1秒間一時停止します
    return JSONResponse({"message": "Starlette非同期処理の例"})

見られるように、FastAPIはデコレータを通じてルート定義を簡素化しますが、StarletteはネイティブのASGI操作に近いです。アプリケーションインスタンスに直接ルートを定義し、より高い柔軟性を提供します。

1.2 ミドルウェアの使用

Starletteは、豊富な種類のミドルウェアをサポートしています。たとえば、単純なログミドルウェアを追加すると、FastAPIでは特定の依存性注入を介して実装する必要があります。

Starletteミドルウェアの例

from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware
import logging

# ロガーを設定
logger = logging.getLogger(__name__)

# カスタムログミドルウェア。BaseHTTPMiddlewareから継承
class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        # リクエストメソッドとURLを含むリクエスト情報をログに記録します
        logger.info(f"Request: {request.method} {request.url}")
        # リクエストの処理を続行し、レスポンスを取得します
        response = await call_next(request)
        # レスポンスステータスコードをログに記録します
        logger.info(f"Response: {response.status_code}")
        return response

# Starletteアプリケーションインスタンスを作成し、ミドルウェアインスタンスを渡します
app = Starlette(middleware=[LoggingMiddleware(app)])

# ルート処理関数を定義します
@app.route("/middleware_example/", methods=["GET"])
async def middleware_example(request):
    return JSONResponse({"message": "ミドルウェアが有効になっています"})

FastAPIで同様の機能を実現するには、カスタム依存関数とグローバル依存構成に依存する必要があります。比較すると、Starletteでのミドルウェアの使用方法はより直感的で、基盤となるASGI仕様に近いです。

1.3 WebSocketのサポート

StarletteはネイティブでWebSocketをサポートしています。以下はWebSocketチャットの簡単な例です。

from starlette.applications import Starlette
from starlette.websockets import WebSocket, WebSocketDisconnect
import json

# Starletteアプリケーションインスタンスを作成
app = Starlette()
# 接続されたクライアントのWebSocketオブジェクトを保存
connected_clients = []

# WebSocketルート処理関数を定義します
@app.websocket_route("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()  # WebSocket接続を受け入れます
    connected_clients.append(websocket)  # 接続されたクライアントをリストに追加します
    try:
        while True:
            # クライアントから送信されたテキストデータを受信します
            data = await websocket.receive_text()
            message = json.loads(data)  # 受信したJSON文字列をPythonオブジェクトに解析します
            for client in connected_clients:
                if client != websocket:
                    # 送信者を除く他のクライアントにメッセージを転送します
                    await client.send_text(json.dumps(message))
    except WebSocketDisconnect:
        connected_clients.remove(websocket)  # 接続が切断されたときに、リストからクライアントを削除します

FastAPIもWebSocketをサポートしていますが、実装の詳細はStarletteと似ています。StarletteはWebSocket処理インターフェースを直接公開しており、開発者が詳細なカスタマイズを行うのに便利です。

2. StarletteでのPydanticの応用

2.1 データ検証とシリアライゼーション

StarletteでPydanticを使用してデータ検証を行う場合、FastAPIと比較します。

FastAPIの例

from fastapi import FastAPI
from pydantic import BaseModel

# FastAPIアプリケーションインスタンスを作成
app = FastAPI()

# Pydanticを使用して、データを検証およびシリアライズするためのデータモデルを定義します
class Item(BaseModel):
    name: str
    price: float

# ルート処理関数を定義します。FastAPIは、受信データを自動的に検証し、レスポンスをシリアライズします
@app.post("/fastapi_items/")
async def create_fastapi_item(item: Item):
    return item

Starletteの例

from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.requests import Request
from pydantic import BaseModel

# Starletteアプリケーションインスタンスを作成
app = Starlette()

# Pydanticを使用して、データを検証およびシリアライズするためのデータモデルを定義します
class Item(BaseModel):
    name: str
    price: float

# ルート処理関数を定義し、リクエストデータと検証ロジックを手動で処理します
@app.route("/starlette_items/", methods=["POST"])
async def create_starlette_item(request: Request):
    data = await request.json()  # リクエストからJSONデータを取得
    try:
        item = Item(**data)  # Pydanticを使用してデータを検証します。無効な場合、例外がスローされます
    except ValueError as e:
        return JSONResponse({"error": str(e)}, status_code=400)  # 検証が失敗した場合、エラーレスポンスを返します
    return JSONResponse(item.dict())  # 検証が成功した場合、シリアライズされたレスポンスを返します

FastAPIはデータ検証とエラーの戻り値を自動的に処理しますが、Starletteでは、開発者は例外を手動でキャッチして処理する必要があります。ただし、このアプローチにより、開発者はより多くの制御が可能になります。

2.2 複雑なデータモデルとネストされた検証

複雑なデータモデルを扱う場合、Pydanticの利点がより顕著になります。

from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.requests import Request
from pydantic import BaseModel

# Starletteアプリケーションインスタンスを作成
app = Starlette()

# アドレスデータモデルを定義します
class Address(BaseModel):
    street: str
    city: str
    zip_code: str

# ユーザーデータモデルを定義します。これにはネストされたアドレスモデルが含まれます
class User(BaseModel):
    username: str
    email: str
    address: Address

# ユーザーデータの検証と保存を処理するためのルート処理関数を定義します
@app.route("/users/", methods=["POST"])
async def create_user(request: Request):
    data = await request.json()  # リクエストからJSONデータを取得します
    try:
        user = User(**data)  # Pydanticを使用してネストされたデータを検証します。無効な場合、例外がスローされます
    except ValueError as e:
        return JSONResponse({"error": str(e)}, status_code=400)  # 検証が失敗した場合、エラーレスポンスを返します
    return JSONResponse(user.dict())  # 検証が成功した場合、シリアライズされたレスポンスを返します

StarletteであろうとFastAPIであろうと、Pydanticはネストされたデータ構造の検証を効率的に処理して、データの整合性と正確性を確保できます。

3. StarletteとPydanticのディープインテグレーション

StarletteのルーティングとミドルウェアをPydanticのデータ検証と組み合わせることで、完全に機能するAPIを構築できます。

from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.requests import Request
from starlette.exceptions import HTTPException
from starlette.middleware.cors import CORSMiddleware
from pydantic import BaseModel

# Starletteアプリケーションインスタンスを作成
app = Starlette()
# すべてのオリジンからのリクエストを許可するために、CORSミドルウェアを追加します（本番環境では、特定のドメイン名を制限する必要があります）
app.add_middleware(CORSMiddleware, allow_origins=["*"])

# Pydanticを使用して、プロダクトデータモデルを定義します
class Product(BaseModel):
    name: str
    price: float
    quantity: int

# プロダクトデータを格納するリスト
products = []

# プロダクトを作成するためのルート処理関数を定義します
@app.route("/products/", methods=["POST"])
async def create_product(request: Request):
    data = await request.json()  # リクエストからJSONデータを取得します
    try:
        product = Product(**data)  # Pydanticを使用してデータを検証します。無効な場合、例外がスローされます
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))  # 検証が失敗した場合、HTTP例外を返します
    products.append(product.dict())  # 検証が成功した場合、プロダクトデータをリストに追加します
    return JSONResponse(product.dict())  # 作成されたプロダクトデータを返します

# すべてのプロダクトを取得するためのルート処理関数を定義します
@app.route("/products/", methods=["GET"])
async def get_products(request):
    return JSONResponse(products)  # すべてのプロダクトデータを返します

この例では、Starletteがルーティング、クロスオリジン問題（ミドルウェアを介して）、およびPydanticがデータ検証とシリアライゼーションを実行する完全なプロセスを示しています。FastAPIと比較して、ドキュメントを自動的に生成するなどの機能はありませんが、開発者は実際のニーズに応じて、drf-spectacularやapispecを使用してAPIドキュメントを生成するなど、サードパーティライブラリを柔軟に選択して拡張できます。

結論

StarletteとPydanticの組み合わせにより、FastAPIのカプセル化に頼らずに、高性能で機能豊富なAPIを構築できます。Starletteは柔軟なASGIアプリケーション基盤を提供し、非同期処理、ミドルウェア、WebSocketなどのコア機能をサポートします。Pydanticはデータ検証とシリアライゼーションに焦点を当てています。FastAPIは開発プロセスを簡素化しますが、StarletteとPydanticを直接使用すると、開発者は基盤となる原則をより深く理解し、プロジェクトの要件に応じて高度にカスタマイズされた調整を行い、複雑なシナリオでより強力な適応性を示すことができます。