はじめに
前回の記事で、ユーザーのCRUD APIが完成しました。
ただ、今の状態には大きな問題があります。
DELETE /users/1 ← 誰でも叩ける。知らない人に消されたら困る。
「ログインした人だけが使えるAPI」 にしたい。
それを実現するのが今回のテーマ「認証」です。
1. そもそも「認証」って何？
現実世界で考えてみましょう。
会社のビルに入るとき、社員証をかざしますよね。
セキュリティは「この社員証を持っているなら、この人は社員だ」と判断して扉を開けます。
APIの認証もまったく同じです。
社員証 = トークン（token）
扉 = 認証が必要なエンドポイント
かざす = Authorization ヘッダーにトークンを入れて送る
「トークンを持っているか？」を確認するだけです。
今回使う JWT（JSON Web Token） は、このトークンの形式のひとつです。
2. ログインの流れ（3ステップ）
認証の全体像は、たったの3ステップです。
【ステップ1】ログイン
ユーザーがメールアドレスとパスワードを送る
→ サーバーが「本物だ」と確認したら、トークンを発行して返す
【ステップ2】トークンを保持
ユーザー（アプリ）はトークンを手元に保存しておく
【ステップ3】APIを使う
リクエストのたびにトークンを一緒に送る
→ サーバーが「このトークンは本物だ」と確認して、処理する
一度ログインすればトークンが発行され、それ以降はトークンを見せるだけでOKです。
毎回パスワードを送る必要はありません。
3. JWTトークンとは何か
JWTトークンは、見た目はこんな文字列です。
eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJhbGljZUBleGFtcGxlLmNvbSJ9.SflKxwRJSMeKKF2QT4fwpM
ランダムに見えますが、.（ドット）で3つのパーツに分かれています。
[ヘッダー].[ペイロード].[署名]
| パーツ | 中身 | 役割 |
|---|---|---|
| ヘッダー | アルゴリズム情報 | 「どうやって作ったか」 |
| ペイロード | メールアドレス・有効期限など | 「誰が・いつまで有効か」 |
| 署名 | SECRET_KEYで生成した値 | 「偽物でないことの証明」 |
⚠️ 重要：ペイロードは誰でも中身を見られる
Base64という方法でエンコードされているだけで、暗号化ではありません。
パスワードや個人情報は絶対に入れないでください。
4. パスワードはどう保存する？
「ログイン時にパスワードを確認するなら、DBにパスワードを保存しないといけない」
でも、パスワードをそのまま保存するのは危険です。
# ❌ 絶対NG：DBが流出したら全員のパスワードが漏れる
User(password="password123")
そこで ハッシュ化 という方法を使います。
"password123" → ハッシュ化 → "$2b$12$eW3Kz..." （元に戻せない）
ハッシュ化は一方向です。元の文字列に戻すことができません。
ログイン時は「入力されたパスワードをハッシュ化して、DBの値と照合する」だけです。
# ✅ 正しい保存方法
User(hashed_password=hash_password("password123"))
# ✅ 正しい照合方法（ログイン時）
verify_password("password123", "$2b$12$eW3Kz...") # → True/False
5. 実装（少しずつ積み上げる）
ここからコードを書いていきます。一気に全部見せるのではなく、なぜこのコードが必要かを説明しながら進めます。
インストール
pip install fastapi uvicorn sqlalchemy python-jose[cryptography] passlib[bcrypt] python-multipart
追加するライブラリは3つだけです。
| ライブラリ | 役割 |
|---|---|
python-jose |
JWTトークンを作る・検証する |
passlib[bcrypt] |
パスワードをハッシュ化する |
python-multipart |
ログインフォームのデータを受け取る |
前回からの変更点
前回の構成に auth.py を1ファイル追加するだけです。
my_api/
├── main.py
├── database.py
├── models.py ← パスワード列を追加
├── schemas.py ← パスワード・トークン用の型を追加
├── crud.py ← ハッシュ化して保存するよう変更
└── auth.py ← 今回新しく追加
Step 1：パスワードのハッシュ化だけ書く（auth.pyの一部）
まず一番シンプルな部分から。「パスワードをハッシュ化する」「照合する」の2つだけです。
# auth.py（まずはこれだけ）
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def hash_password(password: str) -> str:
"""パスワードをハッシュ化して返す"""
return pwd_context.hash(password)
def verify_password(plain: str, hashed: str) -> bool:
"""入力パスワードとハッシュを照合する"""
return pwd_context.verify(plain, hashed)
CryptContext は「どのハッシュ方式を使うか」の設定です。
bcrypt は現在最も広く使われている安全な方式です。
Step 2：JWTトークンの生成を追加する
次にログイン成功時にトークンを発行する関数を追加します。
# auth.py（Step 2を追加）
from passlib.context import CryptContext
from datetime import datetime, timedelta, timezone
from typing import Optional
from jose import jwt
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
# ── JWT設定 ────────────────────────────────────
import os
SECRET_KEY = os.environ.get("SECRET_KEY", "dev-secret-key-change-in-production")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
def hash_password(password: str) -> str:
return pwd_context.hash(password)
def verify_password(plain: str, hashed: str) -> bool:
return pwd_context.verify(plain, hashed)
def create_access_token(data: dict) -> str:
"""JWTトークンを生成して返す"""
payload = data.copy()
expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
payload.update({"exp": expire}) # 有効期限をペイロードに追加
return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
SECRET_KEY はトークンの署名に使う秘密の鍵です。
外部に漏れると他人がトークンを偽造できるので、本番では必ず環境変数で管理します。
💡
os.environ.getとは？
環境変数から値を取得します。
("SECRET_KEY", "dev-secret...")の2番目の引数はデフォルト値で、環境変数が設定されていないときに使われます。
Step 3：トークンを検証して「今誰がアクセスしているか」を取得する
これが認証の核心部分です。リクエストに含まれるトークンを検証して、「このリクエストは誰からか」を特定します。
# auth.py（Step 3を追加）
from passlib.context import CryptContext
from datetime import datetime, timedelta, timezone
from typing import Optional
from jose import JWTError, jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session
from database import get_db
import crud, os
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
SECRET_KEY = os.environ.get("SECRET_KEY", "dev-secret-key-change-in-production")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
def hash_password(password: str) -> str:
return pwd_context.hash(password)
def verify_password(plain: str, hashed: str) -> bool:
return pwd_context.verify(plain, hashed)
def create_access_token(data: dict) -> str:
payload = data.copy()
expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
payload.update({"exp": expire})
return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
# ── ここからStep 3 ──────────────────────────────
# "Authorization: Bearer <token>" からトークンを取り出してくれる仕組み
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")
def get_current_user(
token: str = Depends(oauth2_scheme), # ヘッダーからトークンを自動取得
db: Session = Depends(get_db)
):
"""トークンを検証して現在のユーザーを返す。失敗したら401を返す"""
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
email: str = payload.get("sub") # "sub"にメールアドレスを入れた
if email is None:
raise ValueError
except (JWTError, ValueError):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="ログインしてください",
headers={"WWW-Authenticate": "Bearer"},
)
user = crud.get_user_by_email(db, email=email)
if user is None:
raise HTTPException(status_code=404, detail="ユーザーが見つかりません")
return user
get_current_user はエンドポイントの引数に Depends(get_current_user) と書くだけで自動で動きます。トークンが無効なら自動で 401 を返してくれます。
Step 4：models.py にパスワード列を追加
# models.py
from sqlalchemy import Column, Integer, String, Boolean
from database import Base
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, nullable=False)
email = Column(String, unique=True, index=True, nullable=False)
hashed_password = Column(String, nullable=False) # ← 追加
is_active = Column(Boolean, default=True)
Step 5：schemas.py に認証用の型を追加
# schemas.py
from pydantic import BaseModel, Field
from typing import Optional
class UserResponse(BaseModel):
id: int
name: str
email: str
is_active: bool
model_config = {"from_attributes": True}
class UserCreate(BaseModel):
name: str = Field(min_length=1, max_length=50)
email: str = Field(pattern=r"^[\w\.-]+@[\w\.-]+\.\w+$")
password: str = Field(min_length=8, description="8文字以上")
class UserUpdate(BaseModel):
name: Optional[str] = Field(None, min_length=1, max_length=50)
is_active: Optional[bool] = None
# ── 今回追加 ─────────────────────────────────
class Token(BaseModel):
access_token: str
token_type: str = "bearer"
Step 6：crud.py をハッシュ化対応に更新
# crud.py
from sqlalchemy.orm import Session
from models import User
from schemas import UserCreate, UserUpdate
from auth import hash_password # ← インポート追加
def get_user(db: Session, user_id: int):
return db.get(User, user_id)
def get_user_by_email(db: Session, email: str):
return db.query(User).filter(User.email == email).first()
def get_users(db: Session, skip: int = 0, limit: int = 100):
return db.query(User).offset(skip).limit(limit).all()
def create_user(db: Session, user_data: UserCreate):
db_user = User(
name=user_data.name,
email=user_data.email,
hashed_password=hash_password(user_data.password) # ← ハッシュ化して保存
)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
def update_user(db: Session, user_id: int, user_data: UserUpdate):
db_user = db.get(User, user_id)
if not db_user:
return None
for key, value in user_data.model_dump(exclude_unset=True).items():
setattr(db_user, key, value)
db.commit()
db.refresh(db_user)
return db_user
def delete_user(db: Session, user_id: int):
db_user = db.get(User, user_id)
if not db_user:
return None
db.delete(db_user)
db.commit()
return db_user
Step 7：main.py にログインエンドポイントと認証保護を追加
# main.py
from fastapi import FastAPI, HTTPException, Depends, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from typing import List
import models, schemas, crud
from database import engine, get_db
from auth import verify_password, create_access_token, get_current_user
models.Base.metadata.create_all(bind=engine)
app = FastAPI(title="認証付きユーザー管理API")
# ── 認証エンドポイント ────────────────────────────
@app.post("/auth/register", response_model=schemas.UserResponse, status_code=201)
def register(user: schemas.UserCreate, db: Session = Depends(get_db)):
"""新規ユーザー登録"""
if crud.get_user_by_email(db, user.email):
raise HTTPException(status_code=400, detail="このメールアドレスは既に登録済みです")
return crud.create_user(db, user)
@app.post("/auth/login", response_model=schemas.Token)
def login(
form_data: OAuth2PasswordRequestForm = Depends(),
db: Session = Depends(get_db)
):
"""ログイン → JWTトークンを返す"""
user = crud.get_user_by_email(db, form_data.username)
if not user or not verify_password(form_data.password, user.hashed_password):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="メールアドレスまたはパスワードが違います",
)
token = create_access_token(data={"sub": user.email})
return {"access_token": token, "token_type": "bearer"}
# ── 認証が必要なエンドポイント ────────────────────
@app.get("/users/me", response_model=schemas.UserResponse)
def read_me(current_user=Depends(get_current_user)):
"""自分のプロフィールを取得（ログイン必須）"""
return current_user
@app.get("/users", response_model=List[schemas.UserResponse])
def read_users(
skip: int = 0,
limit: int = 100,
db: Session = Depends(get_db),
current_user=Depends(get_current_user) # ← この1行を追加するだけで認証必須になる
):
return crud.get_users(db, skip=skip, limit=limit)
@app.delete("/users/{user_id}", status_code=204)
def delete_user(
user_id: int,
db: Session = Depends(get_db),
current_user=Depends(get_current_user) # ← 認証必須
):
"""ユーザー削除（自分のアカウントのみ）"""
if current_user.id != user_id:
raise HTTPException(status_code=403, detail="他のユーザーは削除できません")
deleted = crud.delete_user(db, user_id)
if not deleted:
raise HTTPException(status_code=404, detail="ユーザーが見つかりません")
6. 動かしてみる
uvicorn main:app --reload
http://localhost:8000/docs を開くと右上に Authorize ボタンが表示されます。
手順
① POST /auth/register でユーザーを作成
{"name": "Alice", "email": "alice@example.com", "password": "password123"}
② POST /auth/login でログイン
username: alice@example.com ／ password: password123
→ {"access_token": "eyJ...", "token_type": "bearer"} が返ってくる
③ Swagger UIの [Authorize] をクリックして
② のトークンを貼り付ける
④ GET /users/me にアクセス → 自分の情報が返る ✅
⑤ トークンなしで GET /users にアクセス → 401 Unauthorized ✅
7. ハマりやすいポイント
ハマり1：SECRET_KEY をGitHubに上げてしまう
# ❌ コードに直書きしてpushするとアウト
SECRET_KEY = "my-secret-key"
.env ファイルを作って、そこに書くようにしましょう。
# .env ファイル
SECRET_KEY=my-actual-secret-key-here
# main.py
from dotenv import load_dotenv
load_dotenv()
SECRET_KEY = os.environ.get("SECRET_KEY")
.gitignore に .env を追加することも忘れずに。
ハマり2：/auth/login の username フィールドに戸惑う
OAuth2PasswordRequestForm はHTTPの仕様で username というフィールド名が固定されています。
今回はメールアドレスをそこに入れる設計にしているので、Swagger UI上では「username」と表示されますが、実際にはメールアドレスを入力します。
ハマり3：DBのテーブルにカラムが反映されない
models.py に hashed_password を追加した後、既存の app.db を一度削除してから起動し直す必要があります。
rm app.db
uvicorn main:app --reload
8. まとめ
今回やったことを整理します。
パスワードの保存 → hash_password() でハッシュ化してから保存
ログイン処理 → verify_password() で照合 → create_access_token() でトークン発行
認証チェック → get_current_user() をエンドポイントの引数に Depends() で渡すだけ
認証の追加は「1行足すだけ」にできるのがFastAPIの強みです。
# この1行を引数に追加するだけで、そのエンドポイントが認証必須になる
current_user=Depends(get_current_user)
次回は FastAPIでAI APIのラッパーを作る を解説します。
AnthropicのAPIをFastAPIでラップして、社内向けAIツールに組み込む構成を作ります。
参考
この記事は学習日記として書いています。間違いや補足があればコメントいただけると嬉しいです！