FastAPIとは
高速で現代的なPython Web APIフレームワーク。型ヒントを活用した自動検証と自動ドキュメント生成が特徴です。
高速
型安全
自動ドキュメント
非同期対応
基本セットアップ
インストール
# FastAPIとUvicornをインストール
pip install fastapi uvicorn[standard]
# 開発用依存関係も含める場合
pip install fastapi[all]
最小構成(main.py)
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def read_root():
return {"message": "Hello World"}
サーバー起動
# 開発モード(自動リロード)
uvicorn main:app --reload
# ポート指定
uvicorn main:app --reload --port 8080
# ホスト指定(外部アクセス許可)
uvicorn main:app --host 0.0.0.0 --port 8000
自動ドキュメントURL
Swagger UI:
http://127.0.0.1:8000/docs
ReDoc:
http://127.0.0.1:8000/redoc
OpenAPI JSON:
http://127.0.0.1:8000/openapi.json
ルーティングとHTTPメソッド
GET(データ取得)
@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
return {"item_id": item_id, "q": q}
POST(データ作成)
from pydantic import BaseModel
class Item(BaseModel):
name: str
price: float
@app.post("/items/")
def create_item(item: Item):
return {"name": item.name, "price": item.price}
PUT(データ更新)
@app.put("/items/{item_id}")
def update_item(item_id: int, item: Item):
return {"item_id": item_id, "item": item}
DELETE(データ削除)
@app.delete("/items/{item_id}")
def delete_item(item_id: int):
return {"message": f"Item {item_id} deleted"}
パラメータ
パスパラメータ
# URLパスから値を取得
@app.get("/users/{user_id}")
def read_user(user_id: int):
return {"user_id": user_id}
# 複数のパスパラメータ
@app.get("/users/{user_id}/items/{item_id}")
def read_user_item(user_id: int, item_id: str):
return {"user_id": user_id, "item_id": item_id}
クエリパラメータ
# ?key=value 形式
@app.get("/items/")
def read_items(skip: int = 0, limit: int = 10):
return {"skip": skip, "limit": limit}
# URL例: /items/?skip=20&limit=5
# 必須パラメータ(デフォルト値なし)
@app.get("/search/")
def search(q: str):
return {"query": q}
リクエストボディ(Pydantic)
from pydantic import BaseModel, Field
class Item(BaseModel):
name: str
description: str | None = None
price: float = Field(gt=0, description="価格は0より大きい")
tax: float | None = None
@app.post("/items/")
def create_item(item: Item):
return item
ヘッダー・Cookie
from fastapi import Header, Cookie
@app.get("/headers/")
def read_headers(user_agent: str = Header(None)):
return {"User-Agent": user_agent}
@app.get("/cookies/")
def read_cookies(session_id: str = Cookie(None)):
return {"session_id": session_id}
バリデーション
数値バリデーション
from pydantic import Field
class Item(BaseModel):
price: float = Field(gt=0, le=1000000)
quantity: int = Field(ge=1, le=9999)
# gt: greater than (より大きい)
# ge: greater than or equal (以上)
# lt: less than (より小さい)
# le: less than or equal (以下)
文字列バリデーション
from pydantic import Field, EmailStr
class User(BaseModel):
username: str = Field(min_length=3, max_length=20)
email: EmailStr
password: str = Field(min_length=8, pattern=r"^(?=.*[A-Za-z])(?=.*\d)")
レスポンスモデル
レスポンスモデル定義
class UserIn(BaseModel):
username: str
password: str
email: EmailStr
class UserOut(BaseModel):
username: str
email: EmailStr
@app.post("/users/", response_model=UserOut)
def create_user(user: UserIn):
# パスワードは自動的に除外される
return user
ステータスコード
from fastapi import status
@app.post("/items/", status_code=status.HTTP_201_CREATED)
def create_item(item: Item):
return item
@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_item(item_id: int):
return None
エラーハンドリング
HTTPException
from fastapi import HTTPException
@app.get("/items/{item_id}")
def read_item(item_id: int):
if item_id not in items:
raise HTTPException(
status_code=404,
detail="Item not found",
headers={"X-Error": "Custom error"}
)
return items[item_id]
カスタムエラーハンドラ
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"Oops! {exc.name} did something wrong."}
)
非同期処理
async/await基本
import asyncio
@app.get("/async-items/")
async def read_async_items():
# 非同期処理
await asyncio.sleep(1)
return {"message": "Async response"}
# データベースクエリ例
@app.get("/users/{user_id}")
async def read_user(user_id: int):
user = await db.fetch_one(f"SELECT * FROM users WHERE id = {user_id}")
return user
バックグラウンドタスク
from fastapi import BackgroundTasks
def write_log(message: str):
with open("log.txt", "a") as log:
log.write(message + "\n")
@app.post("/send-notification/")
async def send_notification(
email: str,
background_tasks: BackgroundTasks
):
background_tasks.add_task(write_log, f"Notification sent to {email}")
return {"message": "Notification sent in background"}
依存性注入(Dependency Injection)
基本的な依存性
from fastapi import Depends
def get_query_param(q: str = None):
return q
@app.get("/items/")
def read_items(query: str = Depends(get_query_param)):
return {"query": query}
# クラスベースの依存性
class CommonQueryParams:
def __init__(self, skip: int = 0, limit: int = 10):
self.skip = skip
self.limit = limit
@app.get("/users/")
def read_users(commons: CommonQueryParams = Depends()):
return commons
データベース依存性
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.get("/users/")
def read_users(db: Session = Depends(get_db)):
users = db.query(User).all()
return users
認証とセキュリティ
OAuth2 + JWT
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
# ユーザー認証処理
user = authenticate_user(form_data.username, form_data.password)
if not user:
raise HTTPException(status_code=401, detail="Invalid credentials")
access_token = create_access_token(data={"sub": user.username})
return {"access_token": access_token, "token_type": "bearer"}
@app.get("/users/me")
async def read_users_me(token: str = Depends(oauth2_scheme)):
user = decode_token(token)
return user
CORS設定
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:3000"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# すべてのオリジンを許可(開発用のみ)
# allow_origins=["*"]
実務Tips
プロジェクト構成
• app/ - アプリケーションコード
• routers/ - エンドポイント定義
• models/ - Pydanticモデル
• services/ - ビジネスロジック
• dependencies/ - 共通依存性
• tests/ - テストコード
• requirements.txt - 依存関係
ベストプラクティス
✓ 型ヒントを必ず使用する
✓ Pydanticモデルでバリデーション
✓ 依存性注入でコードを分離
✓ 環境変数で設定を管理
✓ テストコードを書く(pytest)
✓ ドキュメント文字列を追加
✓ ロギングを適切に設定
パフォーマンス最適化
✓ 非同期処理を活用(async/await)
✓ データベース接続プールを使用
✓ レスポンスモデルで不要データ除外
✓ キャッシュを適切に実装
✓ Gunicorn + Uvicorn で本番運用
セキュリティ
✓ 環境変数でシークレット管理
✓ HTTPS通信を強制
✓ レート制限を実装
✓ 入力バリデーションを徹底
✓ SQLインジェクション対策(ORM使用)