🧫

FastAPI✕SQLite✕SQLModelで作ったJWT認証付きREST APIで、簡単なテストをする流れ

2023/10/04に公開

Python

はじめに

今回の記事では、FastAPIとSQLiteで作ったJWT認証付きのREST APIをテストする方法を解説する。APIのテストは具体的な手順や実際のコードなどよく忘れてしまう部分なので、執筆者のためにここに簡単な手順を備忘録として残す。

あくまで一個人の見解にすぎないので、参考程度に。

対象とする読者

FastAPIをこれから学ぶひと
FastAPIで実際にテストしたい人
これからFastAPIを実務で導入したいひと

REST APIを開発する手順

FastAPIを使用して、SQLModelとSQLiteを用いた簡単なJWT認証付きのREST APIを開発する手順を以下に示す。

(1) 必要なライブラリのインストール

pip install fastapi[all] uvicorn sqlmodel

(2) モデルの定義

必要なライブラリのインストールが終了したら、REST APIに組み込むデータを実装する。今回はあくまで簡単な認証方法を実装するだけに過ぎないので、以下のような簡単なデータを使って認証を実装する。

ID
ユーザ名
パスワード(ハッシュ化されているもの)

上述の情報をPythonのコードに落とし込む。

/app/models.py

from sqlmodel import SQLModel, Field

class User(SQLModel, table=True):
    id: int = Field(primary_key=True, auto_increment=True) # 主キーに設定し、自動でデータが追加されるように設定する。
    username: str = Field(unique=True) # usernameのデータが被らないようにするため
    hashed_password: str

今回の記事で、データベースに情報を追加するORMにSQLModelを採用している。データを格納するためのデータベースを実際に作るので、UserモデルにSQLModelを継承させ、table=Trueと書いてデータベースの設定を有効化している。

上述の/app/models.pyで定義したUserモデルは以下の通りである。

id：primary_key=Trueで、同じidであるデータを登録できないように設計する。auto_increment=Trueで使えないようにする。
username：ユーザ名。文字列で表示する。
hashed_password：ハッシュ化されたパスワード。

(3) 依存関係の定義:

/app/dependencies.pyで、APIの依存関係を実装する。言い換えれば、APIで実際に行いたい処理をこのファイルに書くのだ。

/app/dependencies.py

# (1) 必要なライブラリをインポートする
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from passlib.context import CryptContext
from .models import User

# (2) 適用する秘密鍵(SECRET_KEY)、認証のアルゴリズム(ALGORITHM)、トークンの有効期限(ACCESS_TOKEN_EXPIRE_MINUTES)を設定
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# (3) 認証に必要なスキーマを定義
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") # password flowを実現するためのインスタンス
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") # パスワードのハッシュ化と検証を行うためのシステム

# (4) 平文のパスワードとハッシュ化されたパスワードを比較して、一致するかどうかを確認する
def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

# (5) データベースから指定されたユーザ名のユーザを取得する
def get_user(db, username: str):
    return db.query(User).filter(User.username == username).first()

# (6) 与えられたトークンを検証し、トークンが有効であれば関連するユーザを返す
# トークンが無効ならば401エラーを出力する
def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception
    user = get_user(db, username=token_data.username)
    if user is None:
        raise credentials_exception
    return user

上述のコードで、特に注目するべきポイントはget_current_user()関数になる。

def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception
    user = get_user(db, username=token_data.username)
    if user is None:
        raise credentials_exception
    return user

get_current_user()関数は、以下の処理を実行する。

与えられたトークンを検証し、トークンが有効であれば関連するユーザを返す
1.で扱うトークンが無効であれば401エラーを返す

上述の処理を実行するには、try...except構文を使って表現しなければならない。

try:
    # ここに実行処理を書く
except Error: # Errorは該当するエラーがあれば何でもかまわない
    # 該当するエラーが出力された場合の処理を書く

try...except構文で得られる最大のメリットはデバッグしたり、プログラムの問題を特定したりしやすくなることにある。try...except構文では対応する例外を特定し、例外が発生した場合のメッセージやログを出力できる。try...except構文はエラーが発生した場合の振る舞いや実行するべき処理をコントロールできるのが最大の強みだ。

(4) メインアプリケーションの作成

/app/main.pyに、FastAPIアプリケーションで実行する処理を書いていく。

/app/main.py

# (1) 必要なライブラリをインポートする
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlmodel import Session, SQLModel, create_engine

# ファイル関係はこのようにインポートする
# 例：「.models」 -> 「app/models.py」
from .models import User
from .dependencies import verify_password, get_user, get_current_user

# (2) アプリケーションのインスタンスを作成する
app = FastAPI()

# (3) データベースURLを設定する
DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL)

@app.on_event("startup")
def on_startup():
    SQLModel.metadata.create_all(engine)

# (4) POSTエンドポイントを実装する
@app.post("/token")
def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = get_user(db, form_data.username)
    if not user:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect username or password")
    if not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect username or password")
    # ここでトークンを生成して返す処理を書く

# (5) GETエンドポイントを実装する
@app.get("/users/me")
def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

(5) Uvicornを使用してアプリケーションを実行

uvicorn app.main:app --reload

テストを書く

FastAPIを用いて開発したAPIのテストは、httpxやrequestsライブラリで実施できる。以下は、ユーザーの作成、ログイン、そして認証付きエンドポイントのテストの一例だ。

(1) エンドポイントの追加

テストユーザ用の情報を作るためのエンドポイントをmain.pyに追加する。

/app/main.py

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlmodel import Session, SQLModel, create_engine
from .models import User
from .dependencies import verify_password, get_user, get_current_user, pwd_context

# ...

# 新規でコードを追加する。
# POSTエンドポイントを作成する
@app.post("/users/")
def create_user(user: User):
    user.hashed_password = pwd_context.hash(user.hashed_password)
    with Session(engine) as session:
        session.add(user)
        session.commit()
        session.refresh(user)
        return user

(2) テストコード

test_auth.pyに、テストコードを以下のように書く。

/tests/test_auth.py

import requests

BASE_URL = "http://127.0.0.1:8000"

def test_auth_flow():
    # ユーザーの作成
    user_data = {"username": "testuser", "hashed_password": "testpassword"}
    response = requests.post(f"{BASE_URL}/users/", json=user_data)
    assert response.status_code == 200, response.text
    user = response.json()
    assert user["username"] == user_data["username"]
    
    # ログイン
    login_data = {"username": "testuser", "password": "testpassword"}
    response = requests.post(f"{BASE_URL}/token", data=login_data)
    assert response.status_code == 200, response.text
    token = response.json()["access_token"]
    
    # 認証付きエンドポイントのテスト
    headers = {"Authorization": f"Bearer {token}"}
    response = requests.get(f"{BASE_URL}/users/me", headers=headers)
    assert response.status_code == 200, response.text
    current_user = response.json()
    assert current_user["username"] == user_data["username"]

if __name__ == "__main__":
    test_auth_flow()

上述のテストコードで特に重要となる部分はtest_auth_flow()関数である。

def test_auth_flow():
    # (1) ユーザーの作成
    user_data = {"username": "testuser", "hashed_password": "testpassword"}
    response = requests.post(f"{BASE_URL}/users/", json=user_data)
    assert response.status_code == 200, response.text
    user = response.json()
    assert user["username"] == user_data["username"]
    
    # (2) ログイン
    login_data = {"username": "testuser", "password": "testpassword"}
    response = requests.post(f"{BASE_URL}/token", data=login_data)
    assert response.status_code == 200, response.text
    token = response.json()["access_token"]
    
    # (3) 認証付きエンドポイントのテスト
    headers = {"Authorization": f"Bearer {token}"}
    response = requests.get(f"{BASE_URL}/users/me", headers=headers)
    assert response.status_code == 200, response.text
    current_user = response.json()
    assert current_user["username"] == user_data["username"]

上述のtest_auth_flow()関数では、認証をテストする上で必要なステップを以下のように分けて実行している。

ユーザを作成する
ログイン機能を実装する
エンドポイントをテストする

上述のどのステップでもassert文が使われているのがわかるだろう。assertとは、プログラムの条件が真であることを確認するためのデバッグを行う上で重宝する文法だ。

assert文は以下のようにして書く。

assert <conditional>, <error_message>

<conditional>：条件式をここに書く。
<error_message>：エラーメッセージを書く。

以下にassert文を使った簡単な例を紹介する。

x = 1
assert x == 1, "x is not 1"

上述のコードは問題なく実行される。ところが、xの値を2に変更するとAssertionErrorが発生するのだ。

x = 2
assert x == 1, "x is not 1" # この場合は文字列の部分が出力される

この場合はエラーメッセージとして"x is not 1"が表示される。このように、assert文はデバッグ時に役立つツールとして適切に使える。

上述のtest_auth.pyにある以下のコードで考える。

def test_auth_flow():
    # (1) ユーザーの作成
    user_data = {"username": "testuser", "hashed_password": "testpassword"}
    response = requests.post(f"{BASE_URL}/users/", json=user_data)
    assert response.status_code == 200, response.text
    user = response.json()
    assert user["username"] == user_data["username"]

上述のコードのこの部分に焦点を当てて考える。

assert response.status_code == 200, response.text

こちらのコードは、ユーザの作成が成功したときに200を出力し、失敗したときにresponse.textを出力する、というような流れになっている。assert文は、処理が成功した時と失敗した時に行う処理を1行でまとめられるのが強みだ。

このテストコードは、ユーザーを作成し、そのユーザーでログインしてトークンを取得、そしてそのトークンを使用して認証付きエンドポイントにアクセスする。各リクエストのレスポンスが期待通りであることを確認する。

テストを実行するには、以下のコマンドを入力する。

python -m tests.test_auth

このテストコードは単純な例なので、実際のプロジェクトでは、エラーハンドリングや異なるシナリオに対するテストケースを追加する必要がある。

最終ディレクトリ

project_root/
│
├── app/
│   ├── models.py          # データベースのモデル定義
│   ├── dependencies.py    # 依存関係と認証関連のヘルパー関数
│   └── main.py            # FastAPIアプリケーションのメインファイル
│
├── tests/
│   └── test_auth.py       # 認証のテストコード（オプション）
│
└── test.db                # SQLiteデータベースファイル

開発環境

Windows 11
FastAPI 0.103
SQLite 3.43.1
SQLmodel 0.0.8

参考サイト

GitHubで編集を提案