FastAPI
高性能で学習しやすく、コーディングが高速で本番環境対応のモダンAPI開発フレームワーク。自動API文書生成と型ヒント活用が特徴。
フレームワーク
FastAPI
概要
FastAPIは、Python 3.6以降の型ヒントを活用した高性能でモダンなWebAPIフレームワークです。自動ドキュメント生成と高速な非同期処理を特徴とします。
詳細
FastAPIは2018年にSebastián Ramírezによって開発されたWebAPIフレームワークで、PydanticとStarletteを基盤として構築されています。Python型ヒントを最大限活用することで、データバリデーション、シリアライゼーション、自動ドキュメント生成を実現しています。ASGI（Asynchronous Server Gateway Interface）をベースとした非同期アーキテクチャにより、Node.jsやGoと同等の高いパフォーマンスを提供します。OpenAPI（Swagger）仕様に準拠した対話的APIドキュメントの自動生成、JSONスキーマの自動作成、リクエスト・レスポンスの検証などの機能を標準で備えています。Netflix、Microsoft、Uber等の大規模企業で採用されており、最近のAPIファーストアプローチのトレンドに最適化されています。「Fast to code」「Few bugs」「Intuitive」「Easy」「Short」「Robust」「Standards-based」の設計原則に基づいて開発されています。
メリット・デメリット
メリット
- 高性能: ASGI非同期処理によりNode.jsやGoと同等の速度
- 型安全性: Python型ヒントによる強力な型チェック機能
- 自動ドキュメント: OpenAPI準拠の対話的APIドキュメント自動生成
- 開発効率: エディタサポートと自動補完による高い開発生産性
- バリデーション: Pydanticによる自動リクエスト・レスポンス検証
- モダン設計: 最新のWeb標準とベストプラクティスに準拠
- 充実したエコシステム: 豊富な拡張機能とコミュニティサポート
デメリット
- 比較的新しい: 歴史が浅くFlaskやDjangoより事例が少ない
- 型ヒント必須: Python型システムへの理解が必要
- API特化: 従来のWebアプリケーション開発には適さない
- 学習コスト: 非同期プログラミングの知識が必要
- エコシステム: 一部のライブラリで非同期対応が不十分
主要リンク
書き方の例
Hello World
# main.py
from fastapi import FastAPI
app = FastAPI(
title="My API",
description="This is a very fancy API",
version="0.1.0",
)
@app.get("/")
async def read_root():
return {"Hello": "World"}
@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
return {"item_id": item_id, "q": q}
# 起動コマンド
# pip install fastapi uvicorn
# fastapi dev main.py
# または
# uvicorn main:app --reload
Pydanticモデルとバリデーション
from datetime import datetime
from typing import Optional, List
from pydantic import BaseModel, Field, EmailStr
from fastapi import FastAPI, HTTPException
from enum import Enum
app = FastAPI()
class ItemStatus(str, Enum):
draft = "draft"
published = "published"
archived = "archived"
class UserBase(BaseModel):
username: str = Field(..., min_length=3, max_length=50)
email: EmailStr
full_name: Optional[str] = None
age: Optional[int] = Field(None, ge=0, le=150)
class UserCreate(UserBase):
password: str = Field(..., min_length=8)
class UserResponse(UserBase):
id: int
created_at: datetime
is_active: bool = True
class Config:
from_attributes = True
class ItemBase(BaseModel):
title: str = Field(..., min_length=1, max_length=100)
description: Optional[str] = Field(None, max_length=500)
price: float = Field(..., gt=0)
tax: Optional[float] = None
status: ItemStatus = ItemStatus.draft
class ItemCreate(ItemBase):
pass
class ItemResponse(ItemBase):
id: int
owner_id: int
created_at: datetime
updated_at: datetime
# APIエンドポイント
@app.post("/users/", response_model=UserResponse)
async def create_user(user: UserCreate):
# パスワードハッシュ化などの処理
user_dict = user.dict()
user_dict["id"] = 1
user_dict["created_at"] = datetime.now()
return UserResponse(**user_dict)
@app.post("/items/", response_model=ItemResponse)
async def create_item(item: ItemCreate, user_id: int):
item_dict = item.dict()
item_dict.update({
"id": 1,
"owner_id": user_id,
"created_at": datetime.now(),
"updated_at": datetime.now()
})
return ItemResponse(**item_dict)
パス・クエリパラメータ
from fastapi import FastAPI, Query, Path, HTTPException
from typing import Optional, List
from enum import Enum
app = FastAPI()
class ModelName(str, Enum):
alexnet = "alexnet"
resnet = "resnet"
lenet = "lenet"
# パスパラメータ
@app.get("/items/{item_id}")
async def read_item(
item_id: int = Path(..., title="The ID of the item", ge=1),
q: Optional[str] = Query(None, min_length=3, max_length=50)
):
return {"item_id": item_id, "q": q}
# Enumを使ったパスパラメータ
@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):
if model_name == ModelName.alexnet:
return {"model_name": model_name, "message": "Deep Learning FTW!"}
return {"model_name": model_name, "message": "Have some residuals"}
# クエリパラメータ（複数・バリデーション付き）
@app.get("/items/")
async def read_items(
q: Optional[str] = Query(
None,
min_length=3,
max_length=50,
regex="^[a-zA-Z0-9 ]*$",
title="Query string"
),
skip: int = Query(0, ge=0),
limit: int = Query(10, gt=0, le=100),
tags: List[str] = Query([])
):
results = {"skip": skip, "limit": limit}
if q:
results.update({"q": q})
if tags:
results.update({"tags": tags})
return results
認証・認可
from datetime import datetime, timedelta
from typing import Optional
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel
from passlib.context import CryptContext
from jose import JWTError, jwt
app = FastAPI()
# セキュリティ設定
SECRET_KEY = "your-secret-key-here"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
# データモデル
class Token(BaseModel):
access_token: str
token_type: str
class User(BaseModel):
username: str
email: Optional[str] = None
full_name: Optional[str] = None
disabled: Optional[bool] = None
class UserInDB(User):
hashed_password: str
# 偽のユーザーデータベース
fake_users_db = {
"testuser": {
"username": "testuser",
"full_name": "Test User",
"email": "[email protected]",
"hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
"disabled": False,
}
}
def verify_password(plain_password, hashed_password):
return pwd_context.verify(plain_password, hashed_password)
def authenticate_user(fake_db, username: str, password: str):
user = get_user(fake_db, username)
if not user or not verify_password(password, user.hashed_password):
return False
return user
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
async def get_current_user(token: str = Depends(oauth2_scheme)):
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])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
except JWTError:
raise credentials_exception
user = get_user(fake_users_db, username=username)
if user is None:
raise credentials_exception
return user
# 認証エンドポイント
@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
user = authenticate_user(fake_users_db, form_data.username, form_data.password)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={"sub": user.username}, expires_delta=access_token_expires
)
return {"access_token": access_token, "token_type": "bearer"}
# 保護されたエンドポイント
@app.get("/users/me/", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_user)):
return current_user
データベース統合
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String, DateTime, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session, relationship
from sqlalchemy.sql import func
from pydantic import BaseModel
from datetime import datetime
from typing import List, Optional
# データベース設定
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
# SQLAlchemyモデル
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
email = Column(String, unique=True, index=True)
username = Column(String, unique=True, index=True)
hashed_password = Column(String)
is_active = Column(Boolean, default=True)
created_at = Column(DateTime(timezone=True), server_default=func.now())
items = relationship("Item", back_populates="owner")
class Item(Base):
__tablename__ = "items"
id = Column(Integer, primary_key=True, index=True)
title = Column(String, index=True)
description = Column(String, index=True)
owner_id = Column(Integer, ForeignKey("users.id"))
created_at = Column(DateTime(timezone=True), server_default=func.now())
owner = relationship("User", back_populates="items")
Base.metadata.create_all(bind=engine)
app = FastAPI()
# 依存性注入
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# CRUD操作
@app.post("/users/", response_model=UserResponse)
def create_user_endpoint(user: UserCreate, db: Session = Depends(get_db)):
db_user = get_user_by_email(db, email=user.email)
if db_user:
raise HTTPException(status_code=400, detail="Email already registered")
return create_user(db=db, user=user)
@app.get("/users/", response_model=List[UserResponse])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
users = get_users(db, skip=skip, limit=limit)
return users
テスト
# test_main.py
import pytest
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
def test_create_user():
response = client.post(
"/users/",
json={"email": "[email protected]", "username": "testuser", "password": "testpass"},
)
assert response.status_code == 200
data = response.json()
assert data["email"] == "[email protected]"
assert "id" in data
def test_read_item():
response = client.get("/items/1?q=test")
assert response.status_code == 200
data = response.json()
assert data["item_id"] == 1
assert data["q"] == "test"
# テスト実行コマンド
# pip install pytest httpx
# pytest