PythonでAPIを作成する完全ガイド - FastAPI編

はじめに

PythonでWebAPIを作成する方法を、初心者から中級者向けに詳しく解説します。今回はFastAPIを使用して、実用的なAPIサーバーを段階的に構築していきます。

目次

1. 環境構築
2. 基本的なAPIの作成
3. データベース連携
4. 認証機能の実装
5. エラーハンドリング
6. 実際の使用例
7. デプロイメント

環境構築

1. 必要なライブラリのインストール

pip install fastapi uvicorn sqlalchemy psycopg2-binary python-jose[cryptography] passlib[bcrypt] python-multipart

各ライブラリの役割：

• FastAPI: 高速なWebAPIフレームワーク
• uvicorn: ASGIサーバー（本番環境でも使用可能）
• sqlalchemy: ORM（データベース操作）
• psycopg2-binary: PostgreSQL接続ドライバ
• python-jose: JWT認証
• passlib: パスワードハッシュ化
• python-multipart: ファイルアップロード対応

2. プロジェクト構成

my_api/
├── app/
│   ├── __init__.py
│   ├── main.py          # メインアプリケーション
│   ├── models.py        # データベースモデル
│   ├── schemas.py       # Pydanticスキーマ
│   ├── crud.py          # データベース操作
│   ├── auth.py          # 認証関連
│   └── database.py      # データベース接続
├── requirements.txt
└── README.md

基本的なAPIの作成

1. シンプルなHello World API

# app/main.py
from fastapi import FastAPI

app = FastAPI(title="My API", version="1.0.0")

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

@app.get("/hello/{name}")
async def say_hello(name: str):
    """パラメータを受け取るエンドポイント"""
    return {"message": f"Hello {name}"}

# サーバー起動
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

2. 起動方法

# 開発環境での起動
uvicorn app.main:app --reload

# または
python app/main.py

ブラウザで http://localhost:8000 にアクセスすると、JSONレスポンスが表示されます。
http://localhost:8000/docs で自動生成されたAPI仕様書も確認できます。

データモデルの定義

1. Pydanticスキーマの作成

# app/schemas.py
from pydantic import BaseModel, EmailStr
from typing import Optional
from datetime import datetime

class UserBase(BaseModel):
    """ユーザーベース情報"""
    email: EmailStr
    name: str
    is_active: bool = True

class UserCreate(UserBase):
    """ユーザー作成用スキーマ"""
    password: str

class UserUpdate(BaseModel):
    """ユーザー更新用スキーマ"""
    email: Optional[EmailStr] = None
    name: Optional[str] = None
    is_active: Optional[bool] = None

class User(UserBase):
    """ユーザーレスポンス用スキーマ"""
    id: int
    created_at: datetime

    class Config:
        orm_mode = True  # SQLAlchemyモデルとの連携を有効化

class PostBase(BaseModel):
    """投稿ベース情報"""
    title: str
    content: str
    published: bool = False

class PostCreate(PostBase):
    pass

class Post(PostBase):
    """投稿レスポンス用スキーマ"""
    id: int
    created_at: datetime
    author_id: int
    author: User

    class Config:
        orm_mode = True

2. SQLAlchemyモデルの作成

# app/models.py
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String, Text, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func

Base = declarative_base()

class User(Base):
    """ユーザーテーブル"""
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True, nullable=False)
    name = Column(String, nullable=False)
    hashed_password = Column(String, nullable=False)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())

    # リレーション
    posts = relationship("Post", back_populates="author")

class Post(Base):
    """投稿テーブル"""
    __tablename__ = "posts"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, nullable=False)
    content = Column(Text, nullable=False)
    published = Column(Boolean, default=False)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    author_id = Column(Integer, ForeignKey("users.id"))

    # リレーション
    author = relationship("User", back_populates="posts")

データベース連携

1. データベース接続設定

# app/database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
import os

# データベースURL（環境変数から取得）
DATABASE_URL = os.getenv(
    "DATABASE_URL", 
    "postgresql://user:password@localhost/dbname"
)

engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def get_db():
    """データベースセッションの取得"""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

2. CRUD操作の実装

# app/crud.py
from sqlalchemy.orm import Session
from . import models, schemas
from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def get_password_hash(password: str) -> str:
    """パスワードをハッシュ化"""
    return pwd_context.hash(password)

def verify_password(plain_password: str, hashed_password: str) -> bool:
    """パスワード検証"""
    return pwd_context.verify(plain_password, hashed_password)

# ユーザー関連CRUD
def get_user(db: Session, user_id: int):
    """ユーザーIDでユーザーを取得"""
    return db.query(models.User).filter(models.User.id == user_id).first()

def get_user_by_email(db: Session, email: str):
    """メールアドレスでユーザーを取得"""
    return db.query(models.User).filter(models.User.email == email).first()

def get_users(db: Session, skip: int = 0, limit: int = 100):
    """ユーザー一覧を取得"""
    return db.query(models.User).offset(skip).limit(limit).all()

def create_user(db: Session, user: schemas.UserCreate):
    """新規ユーザー作成"""
    hashed_password = get_password_hash(user.password)
    db_user = models.User(
        email=user.email,
        name=user.name,
        hashed_password=hashed_password,
        is_active=user.is_active
    )
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

def update_user(db: Session, user_id: int, user_update: schemas.UserUpdate):
    """ユーザー情報更新"""
    db_user = get_user(db, user_id)
    if db_user:
        update_data = user_update.dict(exclude_unset=True)
        for field, value in update_data.items():
            setattr(db_user, field, value)
        db.commit()
        db.refresh(db_user)
    return db_user

def delete_user(db: Session, user_id: int):
    """ユーザー削除"""
    db_user = get_user(db, user_id)
    if db_user:
        db.delete(db_user)
        db.commit()
    return db_user

# 投稿関連CRUD
def get_posts(db: Session, skip: int = 0, limit: int = 100):
    """投稿一覧を取得"""
    return db.query(models.Post).offset(skip).limit(limit).all()

def create_user_post(db: Session, post: schemas.PostCreate, user_id: int):
    """新規投稿作成"""
    db_post = models.Post(**post.dict(), author_id=user_id)
    db.add(db_post)
    db.commit()
    db.refresh(db_post)
    return db_post

認証機能の実装

1. JWT認証の設定

# app/auth.py
from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from sqlalchemy.orm import Session
from . import crud, database, models

# JWT設定
SECRET_KEY = "your-secret-key-here"  # 本番環境では環境変数から取得
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

security = HTTPBearer()

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    """アクセストークン作成"""
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    """トークン検証"""
    token = credentials.credentials
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        email: str = payload.get("sub")
        if email is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    return email

def get_current_user(
    db: Session = Depends(database.get_db),
    email: str = Depends(verify_token)
):
    """現在のユーザーを取得"""
    user = crud.get_user_by_email(db, email=email)
    if user is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="User not found"
        )
    return user

def get_current_active_user(current_user: models.User = Depends(get_current_user)):
    """アクティブユーザーを取得"""
    if not current_user.is_active:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST, 
            detail="Inactive user"
        )
    return current_user

完全なAPIエンドポイントの実装

# app/main.py（完全版）
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy.orm import Session
from typing import List
from datetime import timedelta

from . import crud, models, schemas, auth
from .database import SessionLocal, engine, get_db

# テーブル作成
models.Base.metadata.create_all(bind=engine)

app = FastAPI(
    title="Blog API",
    description="ブログシステムのAPI",
    version="1.0.0"
)

# CORS設定
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 本番環境では適切に設定
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 認証エンドポイント
@app.post("/auth/login")
async def login(user_credentials: schemas.UserLogin, db: Session = Depends(get_db)):
    """ログイン"""
    user = crud.get_user_by_email(db, email=user_credentials.email)
    if not user or not crud.verify_password(user_credentials.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect email or password"
        )
    
    access_token_expires = timedelta(minutes=auth.ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = auth.create_access_token(
        data={"sub": user.email}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

# ユーザー関連エンドポイント
@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    """新規ユーザー作成"""
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)

@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    """ユーザー一覧取得"""
    users = crud.get_users(db, skip=skip, limit=limit)
    return users

@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    """特定ユーザー取得"""
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

@app.get("/users/me/", response_model=schemas.User)
def read_users_me(current_user: models.User = Depends(auth.get_current_active_user)):
    """現在のユーザー情報取得"""
    return current_user

@app.put("/users/{user_id}", response_model=schemas.User)
def update_user(
    user_id: int, 
    user_update: schemas.UserUpdate,
    db: Session = Depends(get_db),
    current_user: models.User = Depends(auth.get_current_active_user)
):
    """ユーザー情報更新"""
    if current_user.id != user_id:
        raise HTTPException(status_code=403, detail="Permission denied")
    
    db_user = crud.update_user(db, user_id=user_id, user_update=user_update)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

# 投稿関連エンドポイント
@app.post("/users/{user_id}/posts/", response_model=schemas.Post)
def create_post_for_user(
    user_id: int, 
    post: schemas.PostCreate, 
    db: Session = Depends(get_db),
    current_user: models.User = Depends(auth.get_current_active_user)
):
    """新規投稿作成"""
    if current_user.id != user_id:
        raise HTTPException(status_code=403, detail="Permission denied")
    return crud.create_user_post(db=db, post=post, user_id=user_id)

@app.get("/posts/", response_model=List[schemas.Post])
def read_posts(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    """投稿一覧取得"""
    posts = crud.get_posts(db, skip=skip, limit=limit)
    return posts

# ヘルスチェック
@app.get("/health")
async def health_check():
    """ヘルスチェック"""
    return {"status": "healthy"}

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

実際の使用例

1. ユーザー登録

curl -X POST "http://localhost:8000/users/" \
     -H "Content-Type: application/json" \
     -d '{
       "email": "[e-mail]",
       "name": "田中太郎",
       "password": "securepassword123"
     }'

2. ログイン

curl -X POST "http://localhost:8000/auth/login" \
     -H "Content-Type: application/json" \
     -d '{
       "email": "[e-mail]",
       "password": "securepassword123"
     }'

3. 認証が必要なエンドポイントへのアクセス

# レスポンスで取得したトークンを使用
TOKEN="your-jwt-token-here"

curl -X GET "http://localhost:8000/users/me/" \
     -H "Authorization: Bearer $TOKEN"

4. 投稿作成

curl -X POST "http://localhost:8000/users/1/posts/" \
     -H "Authorization: Bearer $TOKEN" \
     -H "Content-Type: application/json" \
     -d '{
       "title": "初めての投稿",
       "content": "FastAPIで作ったAPIです！",
       "published": true
     }'

エラーハンドリング

# app/main.py に追加
from fastapi import Request
from fastapi.responses import JSONResponse

@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    """HTTP例外ハンドラー"""
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": {
                "code": exc.status_code,
                "message": exc.detail,
                "timestamp": datetime.utcnow().isoformat()
            }
        }
    )

@app.exception_handler(ValidationError)
async def validation_exception_handler(request: Request, exc: ValidationError):
    """バリデーション例外ハンドラー"""
    return JSONResponse(
        status_code=422,
        content={
            "error": {
                "code": 422,
                "message": "Validation Error",
                "details": exc.errors()
            }
        }
    )

テストの作成

# tests/test_main.py
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from app.main import app
from app.database import get_db
from app import models

# テスト用データベース設定
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

models.Base.metadata.create_all(bind=engine)

def override_get_db():
    try:
        db = TestingSessionLocal()
        yield db
    finally:
        db.close()

app.dependency_overrides[get_db] = override_get_db

client = TestClient(app)

def test_create_user():
    """ユーザー作成テスト"""
    response = client.post(
        "/users/",
        json={"email": "[e-mail]", "name": "テストユーザー", "password": "testpass"}
    )
    assert response.status_code == 200
    data = response.json()
    assert data["email"] == "[e-mail]"
    assert "id" in data

def test_login():
    """ログインテスト"""
    # まずユーザーを作成
    client.post(
        "/users/",
        json={"email": "[e-mail]", "name": "ログインユーザー", "password": "loginpass"}
    )
    
    # ログインテスト
    response = client.post(
        "/auth/login",
        json={"email": "[e-mail]", "password": "loginpass"}
    )
    assert response.status_code == 200
    data = response.json()
    assert "access_token" in data
    assert data["token_type"] == "bearer"

デプロイメント

1. Docker化

# Dockerfile
FROM python:3.9

WORKDIR /app

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

COPY ./app ./app

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

2. docker-compose.yml

version: '3.8'

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://user:password@db:5432/mydb
    depends_on:
      - db

  db:
    image: postgres:13
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: password
      POSTGRES_DB: mydb
    volumes:
      - postgres_data:/var/lib/postgresql/data

volumes:
  postgres_data:

3. 起動

docker-compose up -d

まとめ

この記事では、FastAPIを使用してPythonでWebAPIを作成する方法を詳しく解説しました。

主なポイント：

1. FastAPIの特徴

   • 高速なパフォーマンス
   • 自動的なAPI仕様書生成
   • 型ヒントによる安全性
   • 非同期処理対応

2. 実装した機能

   • RESTful API設計
   • データベース連携（SQLAlchemy）
   • JWT認証
   • エラーハンドリング
   • テスト機能

3. 本番環境対応

   • Docker化
   • 環境変数管理
   • セキュリティ設定
   • ヘルスチェック

FastAPIを使用することで、高品質で保守性の高いAPIを効率的に開発できます。自動生成されるAPI仕様書により、フロントエンド開発者との連携もスムーズになります。

次のステップ

• Redis（キャッシュ）の導入
• ファイルアップロード機能
• WebSocket通信
• GraphQL対応
• マイクロサービス化

皆さんもぜひ試してみてください！