API クラス¶
lambapi の中心となるクラスです。HTTP リクエストの処理、ルーティング、レスポンス生成を管理します。
基本的な使用方法¶
API インスタンスの作成¶
from lambapi import API, create_lambda_handler
def create_app(event, context):
app = API(event, context)
# ルートを定義...
return app
lambda_handler = create_lambda_handler(create_app)
HTTP メソッドの定義¶
@app.get("/users")
def get_users():
return {"users": []}
@app.post("/users")
def create_user():
return {"message": "User created"}
@app.put("/users/{user_id}")
def update_user(user_id: str):
return {"message": f"User {user_id} updated"}
@app.delete("/users/{user_id}")
def delete_user(user_id: str):
return {"message": f"User {user_id} deleted"}
@app.patch("/users/{user_id}")
def patch_user(user_id: str):
return {"message": f"User {user_id} patched"}
高度な機能¶
カスタムレスポンス¶
from lambapi import Response
@app.get("/custom")
def custom_response():
return Response(
{"data": "custom"},
status_code=201,
headers={"X-Custom": "header"}
)
エラーハンドリング¶
from lambapi import ValidationError
@app.get("/error-example")
def error_example():
# 自動的に適切な HTTP ステータスコードで返される
raise ValidationError("Invalid input")
リクエストオブジェクトへのアクセス¶
@app.post("/upload")
def upload_file(request):
# 生のリクエストオブジェクトにアクセス
content_type = request.headers.get("content-type")
body = request.body
return {"received": len(body)}
API クラスの詳細¶
コンストラクタ¶
class API:
def __init__(self, event: dict, context: Any, cors_config: Optional[CORSConfig] = None):
"""
Args:
event: Lambda event オブジェクト
context: Lambda context オブジェクト
cors_config: CORS 設定(オプション)
"""
メソッド一覧¶
| メソッド | 説明 |
|---|---|
get(path) |
GET リクエストのハンドラを登録 |
post(path) |
POST リクエストのハンドラを登録 |
put(path) |
PUT リクエストのハンドラを登録 |
delete(path) |
DELETE リクエストのハンドラを登録 |
patch(path) |
PATCH リクエストのハンドラを登録 |
route(methods, path) |
複数の HTTP メソッドを一度に登録 |
include_router(router) |
Router インスタンスを追加 |
error_handler(exception_class) |
エラーハンドラを登録 |
ルートパラメータ¶
パスパラメータは {} で囲みます:
@app.get("/users/{user_id}")
def get_user(user_id: str):
return {"user_id": user_id}
@app.get("/posts/{post_id}/comments/{comment_id}")
def get_comment(post_id: str, comment_id: str):
return {"post_id": post_id, "comment_id": comment_id}
Router との組み合わせ¶
from lambapi import Router
# 別ファイルで定義
user_router = Router()
@user_router.get("/")
def list_users():
return {"users": []}
@user_router.get("/{user_id}")
def get_user(user_id: str):
return {"user_id": user_id}
# メインアプリで使用
def create_app(event, context):
app = API(event, context)
app.include_router(user_router, prefix="/users")
return app
CORS 設定¶
from lambapi import CORSConfig
cors_config = CORSConfig(
allow_origins=["https://example.com"],
allow_methods=["GET", "POST"],
allow_headers=["Authorization", "Content-Type"],
max_age=3600
)
app = API(event, context, cors_config=cors_config)
レスポンス形式¶
自動 JSON 変換¶
@app.get("/data")
def get_data():
# 自動的に JSON に変換される
return {
"message": "success",
"data": [1, 2, 3],
"timestamp": "2024-01-01T00:00:00Z"
}
カスタムレスポンス¶
@app.get("/xml")
def get_xml():
return Response(
"<xml>data</xml>",
status_code=200,
headers={"Content-Type": "application/xml"}
)
パフォーマンス最適化¶
関数シグネチャキャッシュ¶
lambapi は関数のシグネチャ情報をキャッシュして、Lambda のコールドスタート時間を最小限に抑えます。
# キャッシュの恩恵を受けるため、関数定義は一貫性を保つ
@app.get("/fast")
def fast_handler(user_id: str = Path(...)):
return {"user_id": user_id}
メモリ使用量の最適化¶
# 大きなオブジェクトは関数外で定義してメモリを節約
LARGE_CONFIG = load_config()
@app.get("/config")
def get_config():
return {"config": LARGE_CONFIG}
デバッグとログ¶
リクエスト情報の確認¶
@app.post("/debug")
def debug_request(request):
return {
"method": request.method,
"path": request.path,
"headers": dict(request.headers),
"query_params": dict(request.query_params),
"body": request.body
}
カスタムログ¶
import logging
logger = logging.getLogger(__name__)
@app.get("/logged")
def logged_handler():
logger.info("Handler called")
return {"status": "logged"}
実践的な例¶
RESTful API¶
from lambapi import API, Query, Path, Body, ValidationError
from typing import List, Optional
def create_app(event, context):
app = API(event, context)
# データストレージ(本来は DB)
users = []
@app.get("/users")
def list_users(
page: int = Query(1, ge=1),
limit: int = Query(10, ge=1, le=100)
):
start = (page - 1) * limit
end = start + limit
return {
"users": users[start:end],
"page": page,
"limit": limit,
"total": len(users)
}
@app.post("/users")
def create_user(user_data: dict = Body(...)):
if not user_data.get("name"):
raise ValidationError("Name is required")
user_id = len(users) + 1
user = {"id": user_id, **user_data}
users.append(user)
return Response(user, status_code=201)
@app.get("/users/{user_id}")
def get_user(user_id: int = Path(..., ge=1)):
user = next((u for u in users if u["id"] == user_id), None)
if not user:
return Response({"error": "User not found"}, status_code=404)
return user
return app