FastAPIで完璧なブログを構築する：ユーザーシステムを追加

前の記事では、FastAPIを使用して基本的な個人ブログを構築し、正常にデプロイしました。

しかし、このブログには深刻なセキュリティ上の問題があります。誰でも自由に記事を作成できてしまいます。

次のチュートリアルでは、このブログにユーザーおよび認証システムを追加して、より安全にしていきます。

早速始めましょう。

認証方法の紹介

Web開発において、最も一般的な認証方法は トークンベース（例：JWT） と セッションベース（Cookie） の2つです。

JWT（JSON Web Tokens）: 現在最も人気のある認証方法です。ユーザーがログインすると、サーバーはトークンを生成し、クライアントに返します。クライアントは後続のリクエストにこのトークンを含め、サーバーはトークンを検証するだけでユーザーの身元を確認できます。サーバーはユーザーの状態を保存する必要がないため、この方法は分散型で水平スケーラブルな大規模アプリケーションに非常に適しています。
セッション-Cookie: ユーザーがログインすると、サーバーはセッションを作成し、Cookieを介してセッションIDをブラウザに返します。ブラウザは後続のリクエストにこのCookieを自動的に含めます。サーバーはセッションIDに基づいて対応するセッション情報を検索することで、ユーザーを識別します。

このチュートリアルでは、従来のセッション-Cookie方式を選択します。ブログはシンプルなアーキテクチャのモノリスであるため、認証にセッション-Cookiesを使用するのが最も直接的で古典的、かつ十分に安全なアプローチです。

ステップ1：ユーザーモジュールの作成

認証を処理できるようになる前に、まずユーザーシステムが必要です。

1. ユーザーデータモデルの作成

models.py ファイルを開き、Post クラスの上に User モデルを追加します。

# models.py
import uuid
from datetime import datetime
from typing import Optional
from sqlmodel import Field, SQLModel

class User(SQLModel, table=True):
    id: Optional[uuid.UUID] = Field(default_factory=uuid.uuid4, primary_key=True)
    username: str = Field(unique=True, index=True)
    password: str # 格納されるパスワードは暗号化されたハッシュです

class Post(SQLModel, table=True):
    id: Optional[uuid.UUID] = Field(default_factory=uuid.uuid4, primary_key=True)
    title: str
    content: str
    createdAt: datetime = Field(default_factory=datetime.utcnow, nullable=False)

最初の記事で main.py に create_db_and_tables 関数を実装したため、アプリケーション起動時にすべてのSQLModelモデルが自動的に検出され、対応するデータベーステーブルが作成されます。そのため、SQLステートメントを手動で実行する必要はありません。

手動でSQLを実行する必要があり、データベースがLeapcellで作成された場合、グラフィカルインターフェイスを使用してSQLステートメントを簡単に実行できます。ウェブサイトのデータベース管理ページにアクセスし、上記のステートメントをSQLインターフェイスに貼り付けて実行するだけです。

2. パスワード暗号化ライブラリのインストール

セキュリティのため、ユーザーのパスワードをデータベースに平文で保存することは絶対に避けるべきです。パスワードのハッシュ化には bcrypt ライブラリを使用します。

まず、requirements.txt ファイルに bcrypt を追加します。

# requirements.txt
fastapi
uvicorn[standard]
sqlmodel
psycopg2-binary
jinja2
python-dotenv
python-multipart
bcrypt

次に、インストールコマンドを実行します。

pip install -r requirements.txt

ステップ2：ユーザー登録および検証ロジックの実装

次に、ユーザーデータを処理し、パスワードを検証する関数を作成します。

プロジェクトのルートディレクトリに users_service.py という新しいファイルを作成し、ユーザー関連のビジネスロジックを格納します。

# users_service.py
import bcrypt
from sqlmodel import Session, select
from models import User

def get_user_by_username(username: str, session: Session) -> User | None:
    """ユーザー名でユーザーを検索"""
    statement = select(User).where(User.username == username)
    return session.exec(statement).first()

def create_user(user_data: dict, session: Session) -> User:
    """新しいユーザーを作成し、パスワードをハッシュ化する"""
    # 平文パスワードをバイトに変換
    password_bytes = user_data["password"].encode('utf-8')
    # ソルトを生成し、パスワードをハッシュ化
    salt = bcrypt.gensalt()
    hashed_password = bcrypt.hashpw(password_bytes, salt)
    
    new_user = User(
        username=user_data["username"],
        # ハッシュ化されたパスワード（バイト）をデコードして文字列にし、データベースに保存
        password=hashed_password.decode('utf-8')
    )
    
    session.add(new_user)
    session.commit()
    session.refresh(new_user)
    return new_user

次に、ユーザー認証を処理するために auth_service.py ファイルを作成します。

# auth_service.py
import bcrypt
from sqlmodel import Session
from models import User
from users_service import get_user_by_username

def validate_user(username: str, plain_password: str, session: Session) -> User | None:
    """ユーザー名とパスワードが一致するか検証する"""
    user = get_user_by_username(username, session)
    if not user:
        return None
        
    # 入力された平文パスワードと保存されているハッシュ化パスワードの両方をバイトにエンコード
    plain_password_bytes = plain_password.encode('utf-8')
    hashed_password_bytes = user.password.encode('utf-8')
    
    # bcrypt.checkpw を使用して比較
    if bcrypt.checkpw(plain_password_bytes, hashed_password_bytes):
        return user # 検証成功、ユーザー情報を返す
        
    return None # 検証失敗

ステップ3：ログインおよび登録ページの作成

ユーザーが登録およびログインするためのインターフェースを提供する必要があります。templates フォルダ内に login.html と register.html ファイルを作成します。

register.html

{% include "_header.html" %}
<form action="/users/register" method="POST" class="post-form">
  <h2>Register</h2>
  <div class="form-group">
    <label for="username">Username</label>
    <input type="text" id="username" name="username" required />
  </div>
  <div class="form-group">
    <label for="password">Password</label>
    <input type="password" id="password" name="password" required />
  </div>
  <button type="submit">Register</button>
</form>
<p style="text-align: center; margin-top: 1rem;">
  Already have an account? <a href="/auth/login">Login here</a>.
</p>
{% include "_footer.html" %}

login.html

{% include "_header.html" %}
<form action="/auth/login" method="POST" class="post-form">
  <h2>Login</h2>
  <div class="form-group">
    <label for="username">Username</label>
    <input type="text" id="username" name="username" required />
  </div>
  <div class="form-group">
    <label for="password">Password</label>
    <input type="password" id="password" name="password" required />
  </div>
  <button type="submit">Login</button>
</form>
<p style="text-align: center; margin-top: 1rem;">
  Don't have an account? <a href="/users/register">Register here</a>.
</p>
{% include "_footer.html" %}

同時に、ヘッダーの右上隅に登録とログインへのリンクを追加するために _header.html を更新しましょう。

<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="UTF-8" />
  <meta name="viewport" content="width=device-width, initial-scale=1.0" />
  <title>{{ title }}</title>
  <link rel="stylesheet" href="/static/css/style.css" />
</head>
<body>
  <header>
    <h1><a href="/">My Blog</a></h1>
    <nav>
      <a href="/posts/new" class="new-post-btn">New Post</a>
      <a href="/users/register" class="nav-link">Register</a>
      <a href="/auth/login" class="nav-link">Login</a>
    </nav>
  </header>
  <main>

ステップ4：ルーティングとコントローラーロジックの実装

プロジェクト構造を明確にするために、ルーティングロジックを異なるファイルに分割します。

プロジェクトのルートディレクトリに routers フォルダを作成します。

main.py から Post 関連のすべてのルート（@app.get("/posts", ...) など）をカットし、routers/posts.py ファイルに貼り付けます。

# routers/posts.py
import uuid
from fastapi import APIRouter, Request, Depends, Form
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.templating import Jinja2Templates
from sqlmodel import Session, select

from database import get_session
from models import Post

router = APIRouter()
templates = Jinja2Templates(directory="templates")

@router.get("/", response_class=HTMLResponse)
def root():
    return RedirectResponse(url="/posts", status_code=302)

@router.get("/posts", response_class=HTMLResponse)
def get_all_posts(request: Request, session: Session = Depends(get_session)):
    statement = select(Post).order_by(Post.createdAt.desc())
    posts = session.exec(statement).all()
    return templates.TemplateResponse("index.html", {"request": request, "posts": posts, "title": "Home"})

@router.get("/posts/new", response_class=HTMLResponse)
def new_post_form(request: Request):
    return templates.TemplateResponse("new-post.html", {"request": request, "title": "New Post"})

@router.post("/posts", response_class=HTMLResponse)
def create_post(
    title: str = Form(...), 
    content: str = Form(...),
    session: Session = Depends(get_session)
):
    new_post = Post(title=title, content=content)
    session.add(new_post)
    session.commit()
    return RedirectResponse(url="/posts", status_code=302)

@router.get("/posts/{post_id}", response_class=HTMLResponse)
def get_post_by_id(request: Request, post_id: uuid.UUID, session: Session = Depends(get_session)):
    post = session.get(Post, post_id)
    return templates.TemplateResponse("post.html", {"request": request, "post": post, "title": post.title})

注意：@app デコレータを @router に置き換えています。

routers フォルダ内に、ユーザー登録を処理する users.py を作成します。

# routers/users.py
from fastapi import APIRouter, Request, Depends, Form
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.templating import Jinja2Templates
from sqlmodel import Session

from database import get_session
import users_service

router = APIRouter()
templates = Jinja2Templates(directory="templates")

@router.get("/users/register", response_class=HTMLResponse)
def show_register_form(request: Request):
    return templates.TemplateResponse("register.html", {"request": request, "title": "Register"})

@router.post("/users/register")
def register_user(
    username: str = Form(...),
    password: str = Form(...),
    session: Session = Depends(get_session)
):
    # 簡単のため、複雑な検証は行いません
    users_service.create_user({"username": username, "password": password}, session)
    return RedirectResponse(url="/auth/login", status_code=302)

routers フォルダ内に、ユーザーログインを処理する auth.py を作成します。

# routers/auth.py
from fastapi import APIRouter, Request, Depends, Form, HTTPException
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.templating import Jinja2Templates
from sqlmodel import Session

from database import get_session
import auth_service

router = APIRouter()
templates = Jinja2Templates(directory="templates")

@router.get("/auth/login", response_class=HTMLResponse)
def show_login_form(request: Request):
    return templates.TemplateResponse("login.html", {"request": request, "title": "Login"})

@router.post("/auth/login")
def login(
    username: str = Form(...),
    password: str = Form(...),
    session: Session = Depends(get_session)
):
    user = auth_service.validate_user(username, password, session)
    if not user:
        raise HTTPException(status_code=401, detail="Incorrect username or password")
    
    # 検証成功
    return RedirectResponse(url="/posts", status_code=302)

最後に、main.py を更新して古いルートを削除し、新しいルーターファイルを含めます。

# main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

from database import create_db_and_tables
from routers import posts, users, auth

@asynccontextmanager
async def lifespan(app: FastAPI):
    print("Creating tables..")
    create_db_and_tables()
    yield

app = FastAPI(lifespan=lifespan)

# 静的ファイルディレクトリをマウント
app.mount("/static", StaticFiles(directory="public"), name="static")

# ルーターを含める
app.include_router(posts.router)
app.include_router(users.router)
app.include_router(auth.router)