ユーザーシステム — 登録、ログイン、JWT認証
この章では、FastAPI の JWT 認証方式を学び、Django/Flask の Session 認証との違いを理解します。
JWT vs Session 認証
Django と Flask が使用するのはSession 認証:サーバーがユーザー状態を保存し、Cookie には Session ID のみが保存されます。
FastAPI が推奨するのはJWT 認証:ユーザーはログイン後に Token(トークン)を取得し、以降のリクエストに Token を付ければ認証できます。
| 特徴 | Session(Django/Flask) | JWT(FastAPI) |
|---|---|---|
| 保存場所 | サーバー側(メモリ/データベース) | クライアント側(Cookie/LocalStorage) |
| 拡張性 | サーバーは Session を共有する必要がある | ステートレスで、分散システムを自然にサポート |
| 適用シナリオ | サーバーサイドレンダリングのWebサイト | API + SPA / モバイル端末 |
| 有効期限の制御 | サーバー側でいつでも無効化できる | Token は有効期限内は常に有効 |
依存関係のインストール
(venv) $ pip install passlib python-jose python-multipart
- passlib:パスワードのハッシュ化(werkzeug.security に相当)
- python-jose:JWT の生成と検証
- python-multipart:フォーム送信の処理(OAuth2 ログインフォーム)
User モデルの定義
実例
# ファイルパス: models.py に追加
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
username = Column(String(50), unique=True, nullable=False)
email = Column(String(120), unique=True, nullable=False)
hashed_password = Column(String(256), nullable=False)
def set_password(self, password: str):
"""パスワードをハッシュ化"""
self.hashed_password = pwd_context.hash(password)
def verify_password(self, password: str) -> bool:
"""パスワードを検証"""
return pwd_context.verify(password, self.hashed_password)
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
username = Column(String(50), unique=True, nullable=False)
email = Column(String(120), unique=True, nullable=False)
hashed_password = Column(String(256), nullable=False)
def set_password(self, password: str):
"""パスワードをハッシュ化"""
self.hashed_password = pwd_context.hash(password)
def verify_password(self, password: str) -> bool:
"""パスワードを検証"""
return pwd_context.verify(password, self.hashed_password)
マイグレーションを生成して実行:
(venv) $ alembic revision --autogenerate -m "新增 User 模型" (venv) $ alembic upgrade head
JWT ユーティリティ関数
実例
# ファイルパス: auth.py(新規ファイル)
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
# シークレットキー(本番環境では環境変数から読み取る)
SECRET_KEY = "your-secret-key-change-in-production"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 # Token の有効期限: 24 時間
# OAuth2PasswordBearer は FastAPI に、リクエストヘッダー Authorization: Bearer <token> から JWT を抽出することを伝える
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
def create_access_token(data: dict) -> str:
"""JWT Token を生成"""
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
def decode_access_token(token: str) -> dict | None:
"""JWT Token を検証してデコード"""
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
return payload
except JWTError:
return None
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
# シークレットキー(本番環境では環境変数から読み取る)
SECRET_KEY = "your-secret-key-change-in-production"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 # Token の有効期限: 24 時間
# OAuth2PasswordBearer は FastAPI に、リクエストヘッダー Authorization: Bearer <token> から JWT を抽出することを伝える
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
def create_access_token(data: dict) -> str:
"""JWT Token を生成"""
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
def decode_access_token(token: str) -> dict | None:
"""JWT Token を検証してデコード"""
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
return payload
except JWTError:
return None
登録ルート
実例
# ファイルパス: routers/users.py
from fastapi import APIRouter, Depends, HTTPException, Request, Form
from fastapi.templating import Jinja2Templates
from sqlalchemy.orm import Session
from database import get_db
from models import User
from schemas import UserCreate, UserResponse
from auth import create_access_token
router = APIRouter(prefix="/users", tags=["ユーザー"])
templates = Jinja2Templates(directory="templates")
@router.post("/register", response_model=UserResponse)
def register(
username: str = Form(...),
email: str = Form(...),
password: str = Form(...),
db: Session = Depends(get_db)
):
"""ユーザー登録"""
# ユーザー名とメールアドレスが既に存在するか確認
if db.query(User).filter(User.username == username).first():
raise HTTPException(status_code=400, detail="ユーザー名は既に使用されています")
if db.query(User).filter(User.email == email).first():
raise HTTPException(status_code=400, detail="メールアドレスは既に登録されています")
user = User(username=username, email=email)
user.set_password(password)
db.add(user)
db.commit()
db.refresh(user)
return user
from fastapi import APIRouter, Depends, HTTPException, Request, Form
from fastapi.templating import Jinja2Templates
from sqlalchemy.orm import Session
from database import get_db
from models import User
from schemas import UserCreate, UserResponse
from auth import create_access_token
router = APIRouter(prefix="/users", tags=["ユーザー"])
templates = Jinja2Templates(directory="templates")
@router.post("/register", response_model=UserResponse)
def register(
username: str = Form(...),
email: str = Form(...),
password: str = Form(...),
db: Session = Depends(get_db)
):
"""ユーザー登録"""
# ユーザー名とメールアドレスが既に存在するか確認
if db.query(User).filter(User.username == username).first():
raise HTTPException(status_code=400, detail="ユーザー名は既に使用されています")
if db.query(User).filter(User.email == email).first():
raise HTTPException(status_code=400, detail="メールアドレスは既に登録されています")
user = User(username=username, email=email)
user.set_password(password)
db.add(user)
db.commit()
db.refresh(user)
return user
ログインルート(JWT 発行)
実例
# ファイルパス: routers/users.py に追記
from fastapi.security import OAuth2PasswordRequestForm
from auth import create_access_token
@router.post("/token")
def login_for_access_token(
form_data: OAuth2PasswordRequestForm = Depends(),
db: Session = Depends(get_db)
):
"""
ログインして Token を取得
OAuth2PasswordRequestForm を使用:フィールド名は username と password(フォーム形式)
"""
user = db.query(User).filter(User.username == form_data.username).first()
if not user or not user.verify_password(form_data.password):
raise HTTPException(
status_code=401,
detail="ユーザー名またはパスワードが正しくありません",
headers={"WWW-Authenticate": "Bearer"}
)
# JWT を発行:user_id のみを含める(機密情報を入れてはいけない)
access_token = create_access_token(data={"sub": str(user.id)})
return {"access_token": access_token, "token_type": "bearer"}
from fastapi.security import OAuth2PasswordRequestForm
from auth import create_access_token
@router.post("/token")
def login_for_access_token(
form_data: OAuth2PasswordRequestForm = Depends(),
db: Session = Depends(get_db)
):
"""
ログインして Token を取得
OAuth2PasswordRequestForm を使用:フィールド名は username と password(フォーム形式)
"""
user = db.query(User).filter(User.username == form_data.username).first()
if not user or not user.verify_password(form_data.password):
raise HTTPException(
status_code=401,
detail="ユーザー名またはパスワードが正しくありません",
headers={"WWW-Authenticate": "Bearer"}
)
# JWT を発行:user_id のみを含める(機密情報を入れてはいけない)
access_token = create_access_token(data={"sub": str(user.id)})
return {"access_token": access_token, "token_type": "bearer"}
JWT Token 内
subは標準の subject フィールドで、ユーザー識別子を格納します。Token 自体は暗号化されていない(単なる Base64 エンコード)ため、絶対にパスワードなどの機密情報を入れてはいけません。暗号化したい場合は、JWT ではなく JWE を使用してください。
ルートで JWT 保護を使用する
実例
# 現在のユーザーを取得する依存関数
from auth import oauth2_scheme, decode_access_token
def get_current_user(
token: str = Depends(oauth2_scheme),
db: Session = Depends(get_db)
):
"""JWT Token から現在のユーザーを解析"""
payload = decode_access_token(token)
if payload is None:
raise HTTPException(status_code=401, detail="Token が無効か期限切れです")
user_id = payload.get("sub")
user = db.query(User).filter(User.id == int(user_id)).first()
if user is None:
raise HTTPException(status_code=401, detail="ユーザーが存在しません")
return user
# ログインが必要なルートで注入する
@router.get("/me", response_model=UserResponse)
def read_current_user(current_user: User = Depends(get_current_user)):
"""現在のユーザー情報を取得(ログインが必要)"""
return current_user
from auth import oauth2_scheme, decode_access_token
def get_current_user(
token: str = Depends(oauth2_scheme),
db: Session = Depends(get_db)
):
"""JWT Token から現在のユーザーを解析"""
payload = decode_access_token(token)
if payload is None:
raise HTTPException(status_code=401, detail="Token が無効か期限切れです")
user_id = payload.get("sub")
user = db.query(User).filter(User.id == int(user_id)).first()
if user is None:
raise HTTPException(status_code=401, detail="ユーザーが存在しません")
return user
# ログインが必要なルートで注入する
@router.get("/me", response_model=UserResponse)
def read_current_user(current_user: User = Depends(get_current_user)):
"""現在のユーザー情報を取得(ログインが必要)"""
return current_user
SSR ページの場合、JWT は通常 Cookie に保存され(Authorization Header ではなく)、テンプレートルートでは Cookie から Token を抽出して検証します。
本章のまとめ
この章では、FastAPI の JWT 認証を実装しました:passlib によるパスワードのハッシュ化、python-jose による JWT の発行・検証、OAuth2PasswordBearer による Token の抽出、Depends による認証の依存性注入です。
Django/Flask の Session 認証とは異なり、JWT はステートレスな方式であり、フロントエンドとバックエンドを分離したシナリオに適しています。
その他の拡張