FastAPI入門 – 初心者でもわかるPython Web APIフレームワーク

FastAPI入門 – 初心者でもわかるPython Web APIフレームワーク

FastAPIは、Pythonで高速かつ現代的なWeb APIを構築するためのフレームワークです。自動的なAPIドキュメント生成、型ヒントによる検証、高いパフォーマンスなど、多くの特徴を持っています。

FastAPIとは?

FastAPIは以下の特徴を持つPython Webフレームワークです:

  • 高速: NodeJS並みの高いパフォーマンス
  • 型安全: Python型ヒントによる自動検証
  • 自動ドキュメント: SwaggerUIとReDocの自動生成
  • 直感的: Pythonの標準的な型ヒントを使用
  • プロダクション対応: 本格的なAPI開発に適している

環境構築

必要な環境

  • Python 3.7以上
  • pip(Pythonパッケージマネージャー)

インストール

# FastAPIとASGIサーバー(uvicorn)をインストール
pip install fastapi uvicorn[standard]

最初のFastAPIアプリケーション

Hello World API

最もシンプルなFastAPIアプリケーションを作成してみましょう。

# main.py
from fastapi import FastAPI

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

# ルートエンドポイント
@app.get("/")
async def read_root():
    return {"message": "Hello World"}

# アイテム取得エンドポイント
@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

アプリケーションの実行

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

# オプション説明:
# main:app - main.pyファイルのappオブジェクト
# --reload - コード変更時の自動リロード

ブラウザで以下のURLにアクセスできます:

  • API: http://localhost:8000
  • 自動生成ドキュメント: http://localhost:8000/docs
  • ReDoc: http://localhost:8000/redoc

基本的なHTTPメソッド

GET – データの取得

from fastapi import FastAPI

app = FastAPI()

# 全ユーザー取得
@app.get("/users")
async def get_users():
    return [
        {"id": 1, "name": "田中太郎"},
        {"id": 2, "name": "佐藤花子"}
    ]

# 特定ユーザー取得
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    return {"id": user_id, "name": f"ユーザー{user_id}"}

POST – データの作成

from pydantic import BaseModel

# リクエストボディのモデル定義
class User(BaseModel):
    name: str
    email: str
    age: int

@app.post("/users")
async def create_user(user: User):
    return {
        "message": "ユーザーが作成されました",
        "user": user
    }

PUT – データの更新

@app.put("/users/{user_id}")
async def update_user(user_id: int, user: User):
    return {
        "message": f"ユーザー{user_id}が更新されました",
        "user": user
    }

DELETE – データの削除

@app.delete("/users/{user_id}")
async def delete_user(user_id: int):
    return {"message": f"ユーザー{user_id}が削除されました"}

Pydanticモデルの活用

Pydanticモデルを使用することで、リクエスト・レスポンスデータの型安全性を確保できます。

基本的なモデル

from pydantic import BaseModel, EmailStr
from typing import Optional
from datetime import datetime

class UserBase(BaseModel):
    name: str
    email: EmailStr
    age: int

class UserCreate(UserBase):
    password: str

class UserResponse(UserBase):
    id: int
    created_at: datetime

    # ORMオブジェクトからの変換を有効化
    class Config:
        orm_mode = True

バリデーション機能

from pydantic import BaseModel, validator

class Product(BaseModel):
    name: str
    price: float
    category: str

    @validator('price')
    def price_must_be_positive(cls, v):
        if v <= 0:
            raise ValueError('価格は正の値である必要があります')
        return v

    @validator('name')
    def name_must_not_be_empty(cls, v):
        if not v.strip():
            raise ValueError('商品名は空にできません')
        return v

クエリパラメータとパスパラメータ

パスパラメータ

# 必須パスパラメータ
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

# 複数のパスパラメータ
@app.get("/users/{user_id}/items/{item_id}")
async def read_user_item(user_id: int, item_id: int):
    return {"user_id": user_id, "item_id": item_id}

クエリパラメータ

@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10, q: Optional[str] = None):
    items = get_items(skip=skip, limit=limit, query=q)
    return items

# 使用例: /items/?skip=0&limit=10&q=search_term

レスポンスモデルとステータスコード

レスポンスモデル

from fastapi import status

@app.post("/users/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(user: UserCreate):
    # ユーザー作成ロジック
    created_user = create_user_in_db(user)
    return created_user

カスタムステータスコード

from fastapi import HTTPException

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = get_user_from_db(user_id)
    if user is None:
        raise HTTPException(
            status_code=404, 
            detail="ユーザーが見つかりません"
        )
    return user

エラーハンドリング

HTTPExceptionの使用

from fastapi import HTTPException

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id < 1:
        raise HTTPException(
            status_code=400,
            detail="アイテムIDは1以上である必要があります"
        )

    item = get_item(item_id)
    if not item:
        raise HTTPException(
            status_code=404,
            detail="アイテムが見つかりません"
        )

    return item

カスタム例外ハンドラー

from fastapi import Request
from fastapi.responses import JSONResponse

class CustomException(Exception):
    def __init__(self, name: str):
        self.name = name

@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):
    return JSONResponse(
        status_code=418,
        content={"message": f"カスタムエラー: {exc.name}"}
    )

実践的なAPIの例

商品管理APIの実装例:

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel
from typing import List, Optional

app = FastAPI(title="商品管理API", version="1.0.0")

# データモデル
class Product(BaseModel):
    id: Optional[int] = None
    name: str
    price: float
    description: Optional[str] = None
    category: str

# インメモリデータストア(実際の開発ではデータベースを使用)
products = []
next_id = 1

@app.get("/products", response_model=List[Product])
async def get_products(category: Optional[str] = None):
    if category:
        return [p for p in products if p.category == category]
    return products

@app.get("/products/{product_id}", response_model=Product)
async def get_product(product_id: int):
    product = next((p for p in products if p.id == product_id), None)
    if not product:
        raise HTTPException(
            status_code=404,
            detail="商品が見つかりません"
        )
    return product

@app.post("/products", response_model=Product, status_code=status.HTTP_201_CREATED)
async def create_product(product: Product):
    global next_id
    product.id = next_id
    next_id += 1
    products.append(product)
    return product

@app.put("/products/{product_id}", response_model=Product)
async def update_product(product_id: int, product: Product):
    index = next((i for i, p in enumerate(products) if p.id == product_id), None)
    if index is None:
        raise HTTPException(
            status_code=404,
            detail="商品が見つかりません"
        )

    product.id = product_id
    products[index] = product
    return product

@app.delete("/products/{product_id}")
async def delete_product(product_id: int):
    index = next((i for i, p in enumerate(products) if p.id == product_id), None)
    if index is None:
        raise HTTPException(
            status_code=404,
            detail="商品が見つかりません"
        )

    products.pop(index)
    return {"message": "商品が削除されました"}

まとめ

FastAPIは以下の点で優れたWebフレームワークです:

  1. 開発効率: 型ヒントによる自動補完とドキュメント生成
  2. パフォーマンス: 非同期処理による高速なAPI
  3. 保守性: Pydanticモデルによる型安全なコード
  4. 標準準拠: OpenAPI仕様に完全対応

次回は、FastAPIでのデータベース連携について詳しく解説します。SQLAlchemyを使用したORMの活用方法や、マイグレーション、リレーションの扱い方などを学んでいきましょう。

参考リンク