FastAPIで完璧なブログを構築する:ユーザーシステムを追加
Lukas Schneider
DevOps Engineer · Leapcell

前の記事では、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)
ステップ5:テスト
これで、基本的なユーザー登録とログイン検証ロジックが完成しました。
プロジェクトを再起動します。
uvicorn main:app --reload
http://localhost:3000/users/register
にアクセスして登録します。
登録が成功すると、自動的に http://localhost:3000/auth/login
にリダイレクトされ、ログインします。
正しいアカウント情報と間違ったアカウント情報を入力して結果をテストできます。たとえば、間違った情報を入力すると、401 Unauthorizedエラーが表示されます。
しかし、現在のログインは単なる一回限りの検証プロセスであり、サーバーはユーザーのログイン状態を「覚えて」いません。ブラウザを閉じたり、他のページにアクセスしたりすると、未認証の状態になります。
次の記事では、セッション管理を導入して真のユーザーログイン状態の永続化を実現し、ユーザー権限に基づいてページアクセスや操作を制限します。