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：ルーティングとコントローラーロジックの実装

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

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

2. 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 に置き換えています。

3. 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)

4. 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)

5. 最後に、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)

ステップ5：テスト

これで、基本的なユーザー登録とログイン検証ロジックが完成しました。

プロジェクトを再起動します。

uvicorn main:app --reload

http://localhost:3000/users/register にアクセスして登録します。

登録が成功すると、自動的に http://localhost:3000/auth/login にリダイレクトされ、ログインします。

正しいアカウント情報と間違ったアカウント情報を入力して結果をテストできます。たとえば、間違った情報を入力すると、401 Unauthorizedエラーが表示されます。

しかし、現在のログインは単なる一回限りの検証プロセスであり、サーバーはユーザーのログイン状態を「覚えて」いません。ブラウザを閉じたり、他のページにアクセスしたりすると、未認証の状態になります。

次の記事では、セッション管理を導入して真のユーザーログイン状態の永続化を実現し、ユーザー権限に基づいてページアクセスや操作を制限します。