LoginSignup
0
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@automation2025(frencia)

Pythonの高速Webフレームワーク「Sanic」入門

Posted at

はじめに

PythonでWebアプリケーションを開発する際、多くの開発者がFlaskやDjangoを選択します。しかし、より高速で非同期処理に特化したフレームワークをお探しの方には、Sanicがおすすめです。Sanicは、asyncioを活用した高速なWebフレームワークで、特に大規模なアプリケーションや高負荷な環境で力を発揮します。この記事では、Sanicの基本から応用まで、15の章に分けて詳しく解説していきます。

第1章：Sanicのインストールと基本設定

Sanicを使い始めるには、まずインストールが必要です。Pythonの仮想環境を作成し、pipを使ってSanicをインストールしましょう。

# 仮想環境の作成
python -m venv sanic_env

# 仮想環境の有効化
source sanic_env/bin/activate  # Linuxの場合
sanic_env\Scripts\activate  # Windowsの場合

# Sanicのインストール
pip install sanic

インストールが完了したら、最小限のSanicアプリケーションを作成してみましょう。以下のコードをapp.pyとして保存します。

from sanic import Sanic
from sanic.response import json

app = Sanic("MyFirstSanicApp")

@app.route("/")
async def hello_world(request):
    return json({"message": "Hello, World!"})

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000)

このコードでは、Sanicアプリケーションを作成し、ルートパス（"/"）にアクセスすると、JSONレスポンスを返すシンプルなエンドポイントを定義しています。

第2章：ルーティングの基本

Sanicでは、デコレータを使用してルーティングを定義します。様々なHTTPメソッドに対応したルートを簡単に作成できます。

from sanic import Sanic
from sanic.response import json, text

app = Sanic("RoutingExample")

@app.route("/")
async def index(request):
    return text("Welcome to the home page")

@app.route("/json", methods=["POST"])
async def post_json(request):
    return json({"received": True, "message": request.json})

@app.route("/user/<user_id:int>")
async def get_user(request, user_id):
    return json({"user_id": user_id})

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000)

このコードでは、GETリクエスト、POSTリクエスト、そしてパラメータを含むルートを定義しています。<user_id:int>のような記法を使うことで、URLパラメータの型を指定することもできます。

第3章：リクエストハンドリング

Sanicでは、リクエストオブジェクトを通じてクライアントからのデータを簡単に取得できます。フォームデータ、JSONデータ、クエリパラメータなど、様々な形式のデータを処理できます。

from sanic import Sanic
from sanic.response import json

app = Sanic("RequestHandling")

@app.route("/form", methods=["POST"])
async def handle_form(request):
    name = request.form.get("name")
    email = request.form.get("email")
    return json({"name": name, "email": email})

@app.route("/query")
async def handle_query(request):
    param = request.args.get("param")
    return json({"param": param})

@app.route("/headers")
async def handle_headers(request):
    user_agent = request.headers.get("User-Agent")
    return json({"user_agent": user_agent})

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000)

このコードでは、フォームデータ、クエリパラメータ、そしてヘッダー情報を取得する方法を示しています。requestオブジェクトの様々なプロパティを使用することで、クライアントからの情報を簡単に取得できます。

第4章：レスポンス生成

Sanicでは、様々な形式のレスポンスを簡単に生成できます。テキスト、JSON、HTMLなど、アプリケーションのニーズに合わせて適切なレスポンスを返すことができます。

from sanic import Sanic
from sanic.response import json, text, html, file

app = Sanic("ResponseGeneration")

@app.route("/text")
async def text_response(request):
    return text("This is a plain text response")

@app.route("/json")
async def json_response(request):
    return json({"message": "This is a JSON response"})

@app.route("/html")
async def html_response(request):
    return html("<h1>This is an HTML response</h1>")

@app.route("/file")
async def file_response(request):
    return await file("path/to/your/file.pdf")

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000)

このコードでは、テキスト、JSON、HTML、そしてファイルレスポンスの生成方法を示しています。fileレスポンスを使用する際は、awaitキーワードを使用して非同期でファイルを送信することに注意してください。

第5章：ミドルウェアの使用

ミドルウェアは、リクエストの処理前後に実行される関数で、認証、ログ記録、エラーハンドリングなどに使用されます。Sanicでは、リクエスト前、レスポンス後、そしてHTTPエクセプション時のミドルウェアを定義できます。

from sanic import Sanic
from sanic.response import json

app = Sanic("MiddlewareExample")

@app.middleware("request")
async def add_key(request):
    request.ctx.foo = "bar"

@app.middleware("response")
async def custom_banner(request, response):
    response.headers["X-Served-By"] = "Sanic"

@app.route("/")
async def index(request):
    return json({"test": request.ctx.foo})

@app.exception(Exception)
async def generic_exception(request, exception):
    return json({"error": str(exception)}, status=500)

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000)

このコードでは、リクエスト前にコンテキストにデータを追加し、レスポンス後にカスタムヘッダーを追加しています。また、一般的な例外をキャッチして適切なJSONレスポンスを返すエクセプションハンドラも定義しています。

第6章：データベース接続

Sanicは非同期処理に特化しているため、データベース操作も非同期で行うことが推奨されます。ここでは、aiomysqlを使用してMySQLデータベースに非同期で接続する例を示します。

import aiomysql
from sanic import Sanic
from sanic.response import json

app = Sanic("DatabaseExample")

@app.listener('before_server_start')
async def setup_db(app, loop):
    app.ctx.pool = await aiomysql.create_pool(
        host='127.0.0.1', port=3306,
        user='your_username', password='your_password',
        db='your_database', loop=loop
    )

@app.listener('after_server_stop')
async def close_db(app, loop):
    app.ctx.pool.close()
    await app.ctx.pool.wait_closed()

@app.route("/users")
async def get_users(request):
    async with app.ctx.pool.acquire() as conn:
        async with conn.cursor() as cur:
            await cur.execute("SELECT * FROM users")
            result = await cur.fetchall()
            return json({"users": [dict(zip([column[0] for column in cur.description], row)) for row in result]})

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000)

このコードでは、サーバー起動前にデータベース接続プールを作成し、サーバー停止後にプールを閉じています。/usersエンドポイントでは、データベースからユーザー情報を非同期で取得し、JSONレスポンスとして返しています。

第7章：認証とセッション管理

Sanicでの認証とセッション管理は、ミドルウェアとデコレータを組み合わせて実装できます。ここでは、シンプルなトークンベースの認証システムを実装します。

from sanic import Sanic
from sanic.response import json
from functools import wraps
import jwt
import datetime

app = Sanic("AuthExample")
app.config.SECRET_KEY = "your-secret-key"

def token_required(f):
    @wraps(f)
    async def decorated(request, *args, **kwargs):
        token = request.token
        if not token:
            return json({"message": "Token is missing"}, status=401)
        try:
            data = jwt.decode(token, app.config.SECRET_KEY, algorithms=["HS256"])
        except:
            return json({"message": "Token is invalid"}, status=401)
        return await f(request, *args, **kwargs)
    return decorated

@app.route("/login")
async def login(request):
    auth = request.json
    if auth and auth["username"] == "admin" and auth["password"] == "password":
        token = jwt.encode({
            "user": auth["username"],
            "exp": datetime.datetime.utcnow() + datetime.timedelta(minutes=30)
        }, app.config.SECRET_KEY)
        return json({"token": token})
    return json({"message": "Could not verify"}, status=401)

@app.route("/protected")
@token_required
async def protected(request):
    return json({"message": "This is a protected route"})

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000)

このコードでは、JWTを使用したシンプルな認証システムを実装しています。/loginエンドポイントでユーザー認証を行い、トークンを発行します。@token_requiredデコレータを使用することで、特定のルートを保護することができます。

第8章：ファイルアップロード処理

Sanicでファイルアップロードを処理するのは非常に簡単です。request.filesオブジェクトを使用して、アップロードされたファイルにアクセスできます。

from sanic import Sanic
from sanic.response import json
import os

app = Sanic("FileUploadExample")
app.config.UPLOAD_FOLDER = "uploads"

@app.route("/upload", methods=["POST"])
async def upload_file(request):
    if "file" not in request.files:
        return json({"error": "No file part in the request"}, status=400)
    
    file = request.files["file"][0]
    if file.name == "":
        return json({"error": "No selected file"}, status=400)
    
    if file:
        filename = file.name
        file_path = os.path.join(app.config.UPLOAD_FOLDER, filename)
        await file.save(file_path)
        return json({"message": f"File {filename} uploaded successfully"})

if __name__ == "__main__":
    os.makedirs(app.config.UPLOAD_FOLDER, exist_ok=True)
    app.run(host="0.0.0.0", port=8000)

このコードでは、/uploadエンドポイントでファイルアップロードを処理しています。アップロードされたファイルは指定されたアップロードフォルダに保存されます。エラーハンドリングも含まれており、ファイルが選択されていない場合や、リクエストにファイルパートがない場合にエラーメッセージを返します。

第9章：WebSocket対応

Sanicは、WebSocketプロトコルもサポートしています。リアルタイム通信が必要なアプリケーションで使用できます。

from sanic import Sanic
from sanic.response import json
from sanic.websocket import WebSocketProtocol

app = Sanic("WebSocketExample")

@app.websocket("/ws")
async def websocket(request, ws):
    while True:
        data = await ws.recv()
        print(f"Received: {data}")
        await ws.send(f"Echo: {data}")

@app.route("/")
async def index(request):
    return json({"message": "Welcome to WebSocket server"})

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000, protocol=WebSocketProtocol)

このコードでは、/wsエンドポイントでWebSocket接続を受け付けます。クライアントからメッセージを受信すると、そのメッセージをエコーバックします。WebSocketを使用する際は、app.run()メソッドでprotocol=WebSocketProtocolを指定する必要があります。

第10章：非同期タスク処理

Sanicの非同期性を活かすため、バックグラウンドタスクを実行する方法を紹介します。これは長時間実行される処理や定期的なタスクに適しています。

import asyncio
from sanic import Sanic
from sanic.response import json

app = Sanic("AsyncTaskExample")

async def long_running_task(app):
    while True:
        print("Performing long running task...")
        await asyncio.sleep(60)  # 1分ごとに実行

@app.listener('before_server_start')
async def setup_bg_task(app, loop):
    app.add_task(long_running_task(app))

@app.route("/trigger-task")
async def trigger_task(request):
    app.add_task(asyncio.create_task(one_time_task()))
    return json({"message": "Task triggered"})

async def one_time_task():
    print("Starting one-time task...")
    await asyncio.sleep(10)
    print("One-time task completed")

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000)

このコードでは、サーバー起動時に定期的に実行されるバックグラウンドタスクと、APIエンドポイントから手動でトリガーできる一時的なタスクを実装しています。app.add_task()メソッドを使用して、非同期タスクをSanicのイベントループに追加しています。

第11章：エラーハンドリングとログ記録

適切なエラーハンドリングとログ記録は、アプリケーションの安定性と保守性を向上させます。Sanicでは、カスタムエラーハンドラーを定義し、組み込みのロガーを使用できます。

import logging
from sanic import Sanic
from sanic.response import json
from sanic.log import logger

app = Sanic("ErrorHandlingExample")

# カスタムエラーハンドラー
@app.exception(ValueError)
async def value_error_handler(request, exception):
    return json({"error": str(exception)}, status=400)

@app.exception(Exception)
async def generic_exception_handler(request, exception):
    logger.error(f"An error occurred: {str(exception)}")
    return json({"error": "Internal Server Error"}, status=500)

@app.route("/divide/<num:int>/<denom:int>")
async def divide(request, num, denom):
    if denom == 0:
        raise ValueError("Cannot divide by zero")
    return json({"result": num / denom})

@app.middleware("request")
async def log_request(request):
    logger.info(f"Received request: {request.method} {request.url}")

if __name__ == "__main__":
    # ログレベルの設定
    logging.basicConfig(level=logging.INFO)
    app.run(host="0.0.0.0", port=8000)

このコードでは、特定の例外（ValueError）に対するカスタムハンドラーと、一般的な例外に対するハンドラーを定義しています。また、リクエストごとにログを記録するミドルウェアも実装しています。Sanicの組み込みロガーを使用して、エラーや情報をログに記録しています。

第12章：テスト駆動開発（TDD）とSanic

Sanicアプリケーションのテストは、pytestとpytest-sanicを使用して効率的に行うことができます。ここでは、簡単なテストケースの例を示します。

# app.py
from sanic import Sanic
from sanic.response import json

app = Sanic("TestApp")

@app.route("/")
async def hello_world(request):
    return json({"message": "Hello, World!"})

@app.route("/user/<user_id:int>")
async def get_user(request, user_id):
    return json({"user_id": user_id})

# test_app.py
import pytest
from sanic.testing import SanicTestClient

from app import app

@pytest.fixture
def test_cli(loop):
    return SanicTestClient(app, loop)

async def test_hello_world(test_cli):
    response = await test_cli.get('/')
    assert response.status == 200
    assert response.json == {"message": "Hello, World!"}

async def test_get_user(test_cli):
    user_id = 123
    response = await test_cli.get(f'/user/{user_id}')
    assert response.status == 200
    assert response.json == {"user_id": user_id}

このテストコードでは、pytestとpytest-sanicを使用してSanicアプリケーションのエンドポイントをテストしています。SanicTestClientを使用することで、アプリケーションに対して非同期のリクエストを簡単に送信し、レスポンスを検証できます。

第13章：パフォーマンスチューニング

Sanicは高速なフレームワークですが、さらにパフォーマンスを向上させるためのテクニックがあります。ここでは、いくつかのパフォーマンス最適化の例を紹介します。

from sanic import Sanic
from sanic.response import json
from sanic.response import text
import ujson
import asyncio
import uvloop

asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

app = Sanic("PerformanceExample")

# UJSONを使用してJSONシリアライズを高速化
app.config.UJSON = True

# キャッシュデコレータの実装
def cache(seconds: int):
    def decorator(f):
        cache = {}
        async def wrapper(*args, **kwargs):
            key = str(args) + str(kwargs)
            if key not in cache:
                cache[key] = await f(*args, **kwargs)
                asyncio.create_task(clear_cache(key, seconds))
            return cache[key]
        async def clear_cache(key, seconds):
            await asyncio.sleep(seconds)
            cache.pop(key, None)
        return wrapper
    return decorator

@app.route("/cached")
@cache(60)  # 60秒間キャッシュ
async def cached_response(request):
    await asyncio.sleep(1)  # 重い処理をシミュレート
    return json({"message": "This response is cached"})

@app.route("/stream")
async def stream_response(request):
    async def streaming_fn(response):
        await response.write('START\n')
        for i in range(10):
            await asyncio.sleep(0.1)
            await response.write(f'Chunk {i}\n')
        await response.write('END\n')
    return await app.stream(streaming_fn, content_type='text/plain')

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000, workers=4)

このコードでは、以下のパフォーマンス最適化テクニックを使用しています：

  1. uvloopを使用して非同期処理を高速化
  2. UJSONを使用してJSONシリアライズを高速化
  3. カスタムキャッシュデコレータを実装して重い処理の結果をキャッシュ
  4. ストリーミングレスポンスを使用して大きなレスポンスを効率的に処理
  5. 複数のワーカーを使用してマルチコア処理を活用

これらの最適化により、Sanicアプリケーションのパフォーマンスをさらに向上させることができます。

第14章：セキュリティ対策

Webアプリケーションのセキュリティは非常に重要です。Sanicでは、いくつかの組み込みのセキュリティ機能と、追加のミドルウェアを使用してアプリケーションを保護できます。

from sanic import Sanic
from sanic.response import json
from sanic_cors import CORS
from sanic_limiter import Limiter, get_remote_address

app = Sanic("SecurityExample")

# CORSの設定
CORS(app)

# レートリミッターの設定
limiter = Limiter(app, key_func=get_remote_address)

# セキュリティヘッダーの追加
@app.middleware('response')
async def set_security_headers(request, response):
    response.headers['X-Frame-Options'] = 'SAMEORIGIN'
    response.headers['X-XSS-Protection'] = '1; mode=block'
    response.headers['X-Content-Type-Options'] = 'nosniff'
    response.headers['Strict-Transport-Security'] = 'max-age=31536000; includeSubDomains'

@app.route("/")
@limiter.limit("5 per minute")  # レートリミットの適用
async def index(request):
    return json({"message": "Welcome to the secure app"})

# CSRFトークンの生成と検証（簡易版）
import secrets

@app.route("/get-csrf-token")
async def get_csrf_token(request):
    token = secrets.token_hex(16)
    response = json({"csrf_token": token})
    response.cookies['csrf_token'] = token
    response.cookies['csrf_token']['httponly'] = True
    return response

@app.route("/protected", methods=["POST"])
async def protected_route(request):
    csrf_token = request.headers.get('X-CSRF-Token')
    if not csrf_token or csrf_token != request.cookies.get('csrf_token'):
        return json({"error": "Invalid CSRF token"}, status=403)
    return json({"message": "Action completed successfully"})

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000, ssl={'cert': 'cert.pem', 'key': 'key.pem'})

このコードでは、以下のセキュリティ対策を実装しています：

  1. CORSの設定
  2. レートリミッターの適用
  3. セキュリティヘッダーの追加
  4. CSRFトークンの生成と検証
  5. HTTPSの使用（SSL/TLS）

これらの対策により、クロスサイトスクリプティング（XSS）、クリックジャッキング、CSRFなどの一般的な攻撃からアプリケーションを保護できます。

第15章：デプロイメントとスケーリング

最後に、Sanicアプリケーションの本番環境へのデプロイメントとスケーリングについて説明します。ここでは、DockerとKubernetesを使用したデプロイメントの例を示します。

まず、Dockerfileを作成します：

# Dockerfile
FROM python:3.9

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["python", "app.py"]

次に、Kubernetes用のデプロイメント設定を作成します：

# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: sanic-app
spec:
  replicas: 3
  selector:
    matchLabels:
      app: sanic-app
  template:
    metadata:
      labels:
        app: sanic-app
    spec:
      containers:
      - name: sanic-app
        image: your-docker-registry/sanic-app:latest
        ports:
        - containerPort: 8000
        env:
        - name: SANIC_HOST
          value: "0.0.0.0"
        - name: SANIC_PORT
          value: "8000"
        - name: SANIC_WORKERS
          value: "4"
---
apiVersion: v1
kind: Service
metadata:
  name: sanic-app-service
spec:
  selector:
    app: sanic-app
  ports:
    - protocol: TCP
      port: 80
      targetPort: 8000
  type: LoadBalancer

この設定では、3つのレプリカを作成し、各ポッドで4つのワーカーを実行します。LoadBalancerタイプのServiceを使用して、外部からのトラフィックを処理します。

デプロイメントプロセス：

  1. Dockerイメージをビルドし、レジストリにプッシュします。

    docker build -t your-docker-registry/sanic-app:latest .
docker push your-docker-registry/sanic-app:latest

  2. Kubernetesクラスタにデプロイします。

    kubectl apply -f deployment.yaml

  3. サービスの外部IPアドレスを取得します。

    kubectl get services sanic-app-service

このアプローチにより、Sanicアプリケーションを簡単にスケールアップできます。Kubernetesは負荷に応じて自動的にポッドの数を調整し、高可用性を確保します。

以上で、Sanicの基本から応用まで、15章にわたって詳しく解説しました。Sanicは高速で柔軟性が高く、大規模なWebアプリケーションの開発に適したフレームワークです。この記事で学んだ知識を活かして、効率的で堅牢なWebアプリケーションを開発してください。

0
1
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?