FastAPIで完璧なブログを構築：認可機能の追加

前の記事では、FastAPIブログのユーザー登録システムと基本的なログイン検証ロジックを正常に構築しました。ユーザーはアカウントを作成でき、アプリケーションはユーザー名とパスワードを検証できます。

しかし、現在のログインは一度限りの検証であり、サーバーはユーザーのログイン状態を「覚えて」いません。ページがリフレッシュされるたびに、または新しいページにアクセスするたびに、ユーザーは認証されていないゲストに戻ってしまいます。

この記事では、ミドルウェアを使用してブログの真のユーザーログイン状態管理を実装します。ログインが必要なページや機能を保護し、ユーザーのログイン状態に基づいてインターフェースを動的に更新する方法を学びます。

セッションの設定

FastAPIでセッション管理を処理するために、StarletteのSessionMiddlewareを使用します。StarletteはFastAPIが構築されているASGIフレームワークであり、SessionMiddlewareはセッションを処理するための公式標準ツールです。

まず、itsdangerousライブラリをインストールします。SessionMiddlewareはこれを使用してセッションデータを暗号化署名し、そのセキュリティを確保します。

pip install itsdangerous

次に、requirements.txtファイルに追加します。

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

セッションにRedisを使用する

デフォルトでは、SessionMiddlewareはセッションデータを暗号化し、クライアントサイドのCookieに保存します。このアプローチはシンプルでバックエンドストレージを必要としませんが、Cookieのサイズ制限（通常4KB）という欠点があり、大量のデータ保存には適していません。

より優れたスケーラビリティとセキュリティのために、Redis（高性能インメモリデータベース）を使用してサーバーサイドでセッションを永続化します。これにより、ユーザーがブラウザを閉じたり、サーバーが再起動したりしても、ログイン状態を維持できます。

Redisがない場合は？

LeapcellでRedisインスタンスを作成できます。Leapcellはバックエンドアプリケーションに必要なほとんどのツールを提供します！

インターフェースの「Redisの作成」ボタンをクリックして、新しいRedisインスタンスを作成します。

Redisの詳細ページには、Redisコマンドを直接実行できるオンラインCLIが用意されています。

現時点でRedisサービスが利用できない場合、SessionMiddlewareはデフォルトで署名付きCookieを使用します。このチュートリアルの目的においては、機能に影響はありません。

Redis関連の依存関係をインストールします。

pip install redis

次に、main.pyファイルを開いてSessionMiddlewareをインポートして設定します。

# main.py
import os
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from starlette.middleware.sessions import SessionMiddleware # ミドルウェアをインポート
from dotenv import load_dotenv

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

# 環境変数をロード
load_dotenv()

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

app = FastAPI(lifespan=lifespan)

# 秘密鍵を環境変数から読み込む
# 'your-secret-key' を本当に安全なランダムな文字列に置き換えてください。 `openssl rand -hex 32` を使用して生成できます。
SECRET_KEY = os.getenv("SECRET_KEY", "your-secret-key")

# SessionMiddleware を追加
app.add_middleware(
    SessionMiddleware,
    secret_key=SECRET_KEY,
    session_cookie="session_id", # Cookieに保存されるセッションIDの名前
    max_age=60 * 60 * 24 * 7  # セッションは7日間後に期限切れ
)

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

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

注意: セキュリティのため、secret_keyは複雑なランダムに生成された文字列であるべきです。データベースURLと同様に、ハードコーディングするのではなく、環境変数を通じて管理する必要があります。

設定後、SessionMiddlewareは各リクエストを自動的に処理し、リクエストのCookieからセッションデータを解析してrequest.sessionオブジェクトにアタッチして使用できるようにします。

実際のログインとログアウトルートの実装

次に、実際のログインとログアウトロジックを処理するために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(
    request: Request, # セッションにアクセスするためにRequestオブジェクトを注入
    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")
    
    # 検証成功、ユーザー情報をセッションに保存
    # SessionMiddleware が後続の暗号化とCookie設定を自動的に処理します
    request.session["user"] = {"username": user.username, "id": str(user.id)}
    
    return RedirectResponse(url="/posts", status_code=302)

@router.get("/auth/logout")
def logout(request: Request):
    # セッションをクリア
    request.session.clear()
    return RedirectResponse(url="/", status_code=302)

login関数では、ユーザーが正常に検証された後、基本的なユーザー情報を含む辞書をrequest.session["user"]に保存します。SessionMiddlewareは、このセッションデータを自動的に暗号化および署名し、それを含むCookieをブラウザに設定します。ブラウザは、後続のすべてのリクエストにこのCookieを自動的に含めるため、サーバーはユーザーのログイン状態を認識できます。

logout関数では、request.session.clear()を呼び出します。これによりセッションデータがクリアされ、事実上ユーザーはログアウトします。

ルートの保護とUIの更新

ログインメカニズムができたので、最後のステップは、それを使用して「投稿の作成」機能を保護し、ログイン状態に基づいて異なるUI要素を表示することです。

認証依存関係の作成

FastAPIでルートを保護する最もエレガントな方法は、依存関係注入を使用することです。ユーザーがログインしているかどうかを確認する依存関係関数を作成します。

プロジェクトのルートディレクトリで、auth_dependencies.pyという名前の新しいファイルを作成します。

# auth_dependencies.py
from fastapi import Request, Depends, HTTPException, status
from fastapi.responses import RedirectResponse

def login_required(request: Request):
    """
    ユーザーがログインしているかどうかを確認する依存関係。
    ログインしていない場合は、ログインページにリダイレクトします。
    """
    if not request.session.get("user"):
        # HTTPException を発生させることも選択できます
        # raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated")
        return RedirectResponse(url="/auth/login", status_code=status.HTTP_302_FOUND)
    return request.session.get("user")

def get_user_from_session(request: Request) -> dict | None:
    """
    セッションからユーザー情報を取得します（存在する場合）。
    この依存関係はログインを強制しません。テンプレートでユーザー情報を便利に取得するためだけです。
    """
    return request.session.get("user")

最初の関数login_requiredのロジックはシンプルです。request.sessionにuserが存在しない場合は、ユーザーをログインページにリダイレクトします。存在する場合は、ユーザー情報が返され、ルート関数が直接使用できるようになります。

依存関係の適用

routers/posts.pyを開いて、保護が必要なルートにlogin_required依存関係を適用します。

# 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
from auth_dependencies import login_required # 依存関係をインポート

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

# ... 他のルート ...

# このルートを保護するために依存関係を適用
@router.get("/posts/new", response_class=HTMLResponse)
def new_post_form(request: Request, user: dict = Depends(login_required)):
    return templates.TemplateResponse("new-post.html", {"request": request, "title": "New Post", "user": user})

# このルートを保護するために依存関係を適用
@router.post("/posts", response_class=HTMLResponse)
def create_post(
    title: str = Form(...), 
    content: str = Form(...),
    session: Session = Depends(get_session),
    user: dict = Depends(login_required) # ログインユーザーのみが投稿を作成できるようにする
):
    new_post = Post(title=title, content=content)
    session.add(new_post)
    session.commit()
    return RedirectResponse(url="/posts", status_code=302)

# ... 他のルート ...

これで、認証されていないユーザーが/posts/newにアクセスしようとすると、自動的にログインページにリダイレクトされます。

フロントエンドUIの更新

最後に、UIを更新して、ユーザーのログイン状態に基づいて異なるボタンを表示しましょう。get_user_from_session依存関係を使用してユーザー情報を取得し、テンプレートに渡します。

templates/_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>
      {% if user %}
        <span class="welcome-msg">Welcome, {{ user.username }}</span>
        <a href="/posts/new" class="new-post-btn">New Post</a>
        <a href="/auth/logout" class="nav-link">Logout</a>
      {% else %}
        <a href="/users/register" class="nav-link">Register</a>
        <a href="/auth/login" class="nav-link">Login</a>
      {% endif %}
    </nav>
  </header>
  <main>

上記のテンプレートが正しく機能するには、ビューにユーザー情報を渡して、ビューをレンダリングするすべてのルートを更新する必要があります。

routers/posts.pyで、ビューをレンダリングするすべてのメソッドを修正します。

# routers/posts.py
# ... imports ...
from auth_dependencies import get_user_from_session, login_required # 新しい依存関係をインポート

# ...

@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),
    user: dict | None = Depends(get_user_from_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", "user": user})

# ... new_post_form ルートは上記で更新されました ...

@router.get("/posts/{post_id}", response_class=HTMLResponse)
def get_post_by_id(
    request: Request, 
    post_id: uuid.UUID, 
    session: Session = Depends(get_session),
    user: dict | None = Depends(get_user_from_session) # セッションユーザー情報を取得
):
    post = session.get(Post, post_id)
    # ユーザーをテンプレートに渡す
    return templates.TemplateResponse("post.html", {"request": request, "post": post, "title": post.title, "user": user})