フレームワーク

Starlette

概要

Starletteは、Pythonで高性能な非同期Webアプリケーション開発のための軽量ASGIフレームワークです。FastAPIの基盤技術として使用されています。

詳細

StarletteはTom Christie（Django REST frameworkの作者）によって開発された軽量なASGI（Asynchronous Server Gateway Interface）フレームワークです。「The little ASGI framework that shines」をキャッチフレーズに、非同期Webアプリケーション開発に必要な最小限の機能を提供します。FastAPIやResponderなど多くの上位フレームワークの基盤として採用されており、高いパフォーマンスと柔軟性を持つミニマリストなフレームワークです。ルーティング、ミドルウェアシステム、WebSocket対応、静的ファイル配信、テンプレートエンジン統合、認証システム、テストクライアント、背景タスクなど、基本的なWeb開発機能を備えながら、フレームワーク自体は極めて軽量に設計されています。ASGIサーバー（Uvicorn、Hypercorn等）上で動作し、真の非同期処理によりNode.jsやGoと同等のパフォーマンスを実現します。依存関係を最小限に抑えたピュアなPython実装により、カスタマイズ性と拡張性を重視するプロジェクト、マイクロサービス、API開発、リアルタイムアプリケーション開発に最適です。

メリット・デメリット

メリット

• 極めて軽量: 最小限の機能セットで高いパフォーマンスを実現
• 非同期ネイティブ: ASGI準拠で真の非同期処理をサポート
• 高いパフォーマンス: 軽量設計により優れた実行速度とメモリ効率
• 柔軟なアーキテクチャ: 必要な機能のみを組み合わせて使用可能
• WebSocketサポート: リアルタイム通信機能を標準搭載
• 強力なミドルウェア: 拡張性の高いミドルウェアシステム
• FastAPI基盤: FastAPIの土台として実績のある信頼性

デメリット

• 機能の制限: データベースORM、認証、フォーム処理が非搭載
• 学習リソース: 比較的新しいフレームワークのため日本語情報が限定的
• 開発効率: 高レベル機能は自力実装またはライブラリ追加が必要
• エコシステム: サードパーティライブラリが比較的少ない
• 非同期の複雑さ: async/awaitプログラミングの理解が必要

主要リンク

• Starlette公式サイト
• Starlette公式ドキュメント
• Starlette GitHub リポジトリ
• Uvicorn ASGI サーバー
• FastAPI公式サイト
• Awesome Starlette

書き方の例

Hello World

from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route

async def homepage(request):
    return JSONResponse({'hello': 'world'})

app = Starlette(debug=True, routes=[
    Route('/', homepage),
])

# サーバー起動: uvicorn main:app --reload

ルーティングとレスポンス

from starlette.applications import Starlette
from starlette.responses import PlainTextResponse, JSONResponse
from starlette.routing import Route

async def homepage(request):
    return PlainTextResponse("Homepage")

async def about(request):
    return PlainTextResponse("About")

async def user_profile(request):
    username = request.path_params['username']
    return JSONResponse({
        'username': username,
        'message': f'Hello, {username}!'
    })

async def search(request):
    query = request.query_params.get('q', '')
    return JSONResponse({
        'query': query,
        'results': f'Search results for: {query}'
    })

routes = [
    Route("/", endpoint=homepage),
    Route("/about", endpoint=about),
    Route("/user/{username}", endpoint=user_profile),
    Route("/search", endpoint=search),
]

app = Starlette(routes=routes)

ミドルウェアとCORS

from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.cors import CORSMiddleware
from starlette.middleware.sessions import SessionMiddleware
from starlette.middleware.trustedhost import TrustedHostMiddleware
from starlette.responses import JSONResponse
from starlette.routing import Route

async def homepage(request):
    # セッション使用例
    request.session['visits'] = request.session.get('visits', 0) + 1
    return JSONResponse({
        'message': 'Hello, world!',
        'visits': request.session['visits']
    })

# ミドルウェア設定
middleware = [
    Middleware(
        CORSMiddleware,
        allow_origins=['*'],
        allow_credentials=True,
        allow_methods=['*'],
        allow_headers=['*']
    ),
    Middleware(SessionMiddleware, secret_key='your-secret-key'),
    Middleware(
        TrustedHostMiddleware,
        allowed_hosts=['example.com', '*.example.com']
    )
]

routes = [
    Route("/", endpoint=homepage)
]

app = Starlette(debug=True, routes=routes, middleware=middleware)

カスタムミドルウェア

from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.responses import JSONResponse
from starlette.routing import Route
import time
import uuid

# カスタムミドルウェア: リクエスト時間測定
class TimingMiddleware:
    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        start_time = time.time()
        
        async def send_wrapper(message):
            if message["type"] == "http.response.start":
                process_time = time.time() - start_time
                headers = dict(message["headers"])
                headers[b"x-process-time"] = str(process_time).encode()
                message["headers"] = list(headers.items())
            await send(message)

        await self.app(scope, receive, send_wrapper)

# リクエストIDミドルウェア
class RequestIDMiddleware:
    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        scope["request_id"] = str(uuid.uuid4())
        await self.app(scope, receive, send)

async def homepage(request):
    request_id = request.scope.get("request_id", "unknown")
    return JSONResponse({
        'message': 'Hello, world!',
        'request_id': request_id
    })

middleware = [
    Middleware(TimingMiddleware),
    Middleware(RequestIDMiddleware)
]

app = Starlette(
    routes=[Route("/", endpoint=homepage)],
    middleware=middleware
)

WebSocketサポート

from starlette.applications import Starlette
from starlette.routing import Route, WebSocketRoute
from starlette.responses import HTMLResponse
from starlette.websockets import WebSocket, WebSocketDisconnect
from typing import List

# 接続管理クラス
class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def send_personal_message(self, message: str, websocket: WebSocket):
        await websocket.send_text(message)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

async def websocket_endpoint(websocket):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.send_personal_message(f"You wrote: {data}", websocket)
            await manager.broadcast(f"Someone says: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast("Someone left the chat")

async def websocket_user(websocket):
    client_id = websocket.path_params["client_id"]
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Client {client_id}: {data}")
    except WebSocketDisconnect:
        print(f"Client {client_id} disconnected")

async def homepage(request):
    return HTMLResponse("""
    <!DOCTYPE html>
    <html>
        <head>
            <title>WebSocket Test</title>
        </head>
        <body>
            <h1>WebSocket Test</h1>
            <div id="messages"></div>
            <input type="text" id="messageText" placeholder="Type a message...">
            <button onclick="sendMessage()">Send</button>
            <script>
                const ws = new WebSocket("ws://localhost:8000/ws");
                ws.onmessage = function(event) {
                    const messages = document.getElementById('messages');
                    messages.innerHTML += '<div>' + event.data + '</div>';
                };
                function sendMessage() {
                    const messageText = document.getElementById('messageText');
                    ws.send(messageText.value);
                    messageText.value = '';
                }
            </script>
        </body>
    </html>
    """)

routes = [
    Route("/", endpoint=homepage),
    WebSocketRoute("/ws", endpoint=websocket_endpoint),
    WebSocketRoute("/ws/{client_id}", endpoint=websocket_user),
]

app = Starlette(debug=True, routes=routes)

静的ファイルとテンプレート

from starlette.applications import Starlette
from starlette.routing import Route, Mount
from starlette.staticfiles import StaticFiles
from starlette.templating import Jinja2Templates
from starlette.responses import HTMLResponse

# テンプレート設定
templates = Jinja2Templates(directory='templates')

async def homepage(request):
    return templates.TemplateResponse(
        'index.html', 
        {'request': request, 'title': 'Starlette App'}
    )

async def user_profile(request):
    username = request.path_params['username']
    user_data = {
        'username': username,
        'email': f'{username}@example.com',
        'created_at': '2025-01-01'
    }
    return templates.TemplateResponse(
        'profile.html', 
        {'request': request, 'user': user_data}
    )

routes = [
    Route('/', endpoint=homepage),
    Route('/user/{username}', endpoint=user_profile),
    Mount('/static', StaticFiles(directory='static'), name='static')
]

app = Starlette(debug=True, routes=routes)

認証システム

from starlette.applications import Starlette
from starlette.authentication import (
    AuthCredentials, AuthenticationBackend, AuthenticationError, SimpleUser
)
from starlette.middleware import Middleware
from starlette.middleware.authentication import AuthenticationMiddleware
from starlette.responses import PlainTextResponse, JSONResponse
from starlette.routing import Route
from starlette.authentication import requires
import base64
import binascii

class BasicAuthBackend(AuthenticationBackend):
    async def authenticate(self, conn):
        if "Authorization" not in conn.headers:
            return

        auth = conn.headers["Authorization"]
        try:
            scheme, credentials = auth.split()
            if scheme.lower() != 'basic':
                return
            decoded = base64.b64decode(credentials).decode("ascii")
        except (ValueError, UnicodeDecodeError, binascii.Error):
            raise AuthenticationError('Invalid basic auth credentials')

        username, _, password = decoded.partition(":")
        # 実際の実装では、データベースでユーザー検証
        if username == "admin" and password == "secret":
            return AuthCredentials(["authenticated"]), SimpleUser(username)
        
        raise AuthenticationError('Invalid credentials')

async def homepage(request):
    if request.user.is_authenticated:
        return PlainTextResponse(f'Hello, {request.user.display_name}')
    return PlainTextResponse('Hello, anonymous user')

@requires("authenticated")
async def protected(request):
    return JSONResponse({
        "message": f"Hello {request.user.display_name}, this is protected!",
        "user": request.user.display_name
    })

def on_auth_error(request, exc):
    return JSONResponse({"error": str(exc)}, status_code=401)

middleware = [
    Middleware(
        AuthenticationMiddleware, 
        backend=BasicAuthBackend(), 
        on_error=on_auth_error
    ),
]

routes = [
    Route("/", endpoint=homepage),
    Route("/protected", endpoint=protected),
]

app = Starlette(routes=routes, middleware=middleware)

高度なミドルウェア

from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse
from starlette.routing import Route
import time
import logging

# ロギング設定
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        start_time = time.time()
        
        # リクエストログ
        logger.info(f"Request: {request.method} {request.url}")
        
        response = await call_next(request)
        
        # レスポンスログ
        process_time = time.time() - start_time
        logger.info(f"Response: {response.status_code} ({process_time:.3f}s)")
        
        return response

class ErrorHandlingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        try:
            response = await call_next(request)
            return response
        except Exception as exc:
            logger.error(f"Unhandled error: {exc}")
            return JSONResponse(
                {"error": "Internal server error"}, 
                status_code=500
            )

# レート制限ミドルウェア（簡易版）
class RateLimitMiddleware:
    def __init__(self, app, calls: int = 10, period: int = 60):
        self.app = app
        self.calls = calls
        self.period = period
        self.clients = {}

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        client_ip = scope["client"][0]
        current_time = time.time()
        
        # クライアントの履歴をクリーンアップ
        if client_ip in self.clients:
            self.clients[client_ip] = [
                t for t in self.clients[client_ip] 
                if current_time - t < self.period
            ]
        else:
            self.clients[client_ip] = []

        # レート制限チェック
        if len(self.clients[client_ip]) >= self.calls:
            response = JSONResponse(
                {"error": "Rate limit exceeded"}, 
                status_code=429
            )
            await response(scope, receive, send)
            return

        # リクエストを記録
        self.clients[client_ip].append(current_time)
        await self.app(scope, receive, send)

async def api_endpoint(request):
    return JSONResponse({"message": "Hello from API!"})

async def slow_endpoint(request):
    # 重い処理をシミュレート
    import asyncio
    await asyncio.sleep(2)
    return JSONResponse({"message": "This was slow"})

middleware = [
    Middleware(LoggingMiddleware),
    Middleware(ErrorHandlingMiddleware),
    Middleware(RateLimitMiddleware, calls=5, period=60),
]

routes = [
    Route("/api", endpoint=api_endpoint),
    Route("/slow", endpoint=slow_endpoint),
]

app = Starlette(routes=routes, middleware=middleware)

テスト

# test_app.py
import pytest
from starlette.testclient import TestClient
from main import app

@pytest.fixture
def client():
    return TestClient(app)

def test_homepage(client):
    response = client.get("/")
    assert response.status_code == 200
    assert "Hello" in response.text

def test_json_endpoint(client):
    response = client.get("/api")
    assert response.status_code == 200
    data = response.json()
    assert "message" in data

def test_websocket(client):
    with client.websocket_connect("/ws") as websocket:
        websocket.send_text("Hello WebSocket")
        data = websocket.receive_text()
        assert "Hello WebSocket" in data

def test_authenticated_endpoint(client):
    # 認証なしでアクセス
    response = client.get("/protected")
    assert response.status_code == 401
    
    # Basic認証でアクセス
    response = client.get(
        "/protected",
        auth=("admin", "secret")
    )
    assert response.status_code == 200

def test_rate_limiting(client):
    # 制限内のリクエスト
    for i in range(5):
        response = client.get("/api")
        assert response.status_code == 200
    
    # 制限を超えるリクエスト
    response = client.get("/api")
    assert response.status_code == 429

def test_static_files(client):
    # 静的ファイルのテスト（ファイルが存在する場合）
    response = client.get("/static/test.txt")
    # ファイルの存在に応じて 200 または 404

def test_template_response(client):
    response = client.get("/user/testuser")
    assert response.status_code == 200
    assert "testuser" in response.text

# 非同期テストの例
import pytest_asyncio

@pytest_asyncio.async_test
async def test_async_endpoint():
    from httpx import AsyncClient
    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.get("/slow")
        assert response.status_code == 200

# テスト実行コマンド
# pip install pytest httpx pytest-asyncio
# pytest
# pytest -v --cov=main