FastAPIを使ってAPI開発の勉強を始めた話。 - CCCMKホールディングス TECH Labの Tech Blog

こんにちは、CCCMKホールディングス TECH LABの三浦です。

8月も後半に入り、そろそろ夏休みも終わりの頃ですね。私が子どもの頃は夏休みがもうすぐ終わってしまうこの頃が嫌で、時間が流れるのが止まっちゃえばいいのに、といつも思っていたことを思い出しました。

LLMsについては今年度に入ってからいろんなことを試してきました。Azure OpenAIで提供されているChatGPTのモデルを使ってプロンプトエンジニアリングを試したり、外部データを連携したRAGを試したり。さらにオープンソースのLLMsをQLoRAといったテクニックを使って特定のタスクでの使用を想定したチューニングを行ったりもしました。

LLMsの面白さはその柔軟性にあるように感じています。入力と、それに対して望ましい出力がある程度定義出来ていれば、概ね同じ手続きで、その要件をそこそこのレベルで満たすモデルが作れてしまいます。

LLMsに関連する様々なロジックを他のシステムと連携出来たらな・・・とこの頃考えるようになりました。そのためにはLLMsを使ったロジックをAPIとして提供する必要が出てきます。

具体的には以下のようなことが出来たらいいな・・・と考えています。

ユーザーはフロントのチャット風アプリケーションを通じて色々なLLMsを使ったアプリにアクセスして会話形式で操作することが出来ます。恐らくユーザーごとに使えるLLMsアプリケーションは分けた方がいいと思うので、ユーザーの認証機能はLLMsアプリケーション側に持った方が良さそうです。

色々と考え始めると、API開発の勉強をちゃんとしないと・・・と思うようになりました。Pythonの軽量なWebAPI開発フレームワークとしてはFastAPIがとても有名です。FastAPIはチュートリアルがとても豊富で、FastAPIの使い方を学びながら、API開発に必要な知識を身に着けることが出来ます。

今回はFastAPIのチュートリアルのSecurityの項目について、一通りソースコードを書きながら勉強してみたので、自分の理解の整理の意味も込めて、まとめてみたいと思います。

全体の流れ

FastAPIのSecurityのチュートリアルを一通り試すと、最終的には開発したAPIをユーザー登録済みのユーザーに対し、トークンを発行してアクセスが出来るようにするフローを理解することが出来ます。このフローをもう少し細かく落とし込むと、

username, passwordでログインし、ユーザーの認証を行う
登録済みのユーザーであることが確認出来たら期限付きのトークンを発行する
ユーザーはAPIへのリクエストのヘッダに発行されたトークンを含めることで、特定のエンドポイントが利用可能になる

となります。この中で発行するトークンはJson Web Token(JWT)という形式で、ユーザー識別子とセッション有効期間をそれぞれsub, expというキーに格納したJSON形式のデータを秘密鍵で暗号化したものです。トークン内にユーザーを特定出来る情報が含まれているため、一度username, passwordでの認証が済んだ後は発行したトークンでユーザーを識別することが出来ます。またトークンを所有していることがそのトークンの利用権利を表すBearerトークンを使用します。

このフローはOAuth2のAuthorization Grant(認可グラント)の中のResource Owner Password Credentialsに対応しているようです。Resource Owner Password Credentialsはレガシーな方法で、クライアントアプリケーション側でユーザーのパスワードを入力する必要があることから、リスクもあると言えます。しかし基礎的な部分を知るためにはとても参考になると思いますので、以降実際のコードを交えながらどうやってFastAPIでこのフローを組んでいくのかをまとめていきたいと思います。

必要なライブラリ

以下のライブラリが必要になります。

fastapi
pydantic
uvicorn
python-jose[cryptography]
passlib[bcrypt]
python-multipart

コード

以下のコードはFastAPI - Tutorial - User Guide - Securityの"OAuth2 with Password (and hashing), Bearer with JWT tokens"まで進むと完成するもので、Tutorialに掲載されているものとほぼ同じ内容になっています。

fastapi.tiangolo.com

Dependenciesなどの定義

FastAPIではDependenciesという形でAPIのエンドポイントに対するリクエストを処理する関数を、各エンドポイントに紐づいた関数と切り分けて定義して使用することが出来ます。まずはそれらの関数やこのアプリケーションで使用するデータモデルなどを定義します。

import os
import uuid
from typing import Annotated, Union
from datetime import datetime, timedelta

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel

# 以下はJWTトークンの生成に使用する

## 暗号化に使用する秘密鍵
SECRET_KEY = "xxxx" 

## 暗号化方式
ALGORITHM = "HS256"

## トークン有効期間(分)
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# ユーザー情報が記録されたDB(を模した変数)
fake_users_db = {
    "johndoe":{
        "username":"johndoe",
        "full_name":"John Doe",
        "email":"johndoe@example.com",
        # plain_password: "secret"をハッシュ化した値
        "hashed_password":"xxxx",
        "disable":False
    },
}

# Data Models
class User(BaseModel):
    username: str
    email: Union[str,None] = None
    full_name: Union[str,None] = None
    disabled: Union[bool,None] = None

class UserInDB(User):
    hashed_password: str

class Token(BaseModel):
    access_token: str
    token_type: str

class TokenData(BaseModel):
    username: Union[str, None] = None

# パスワードのハッシュ処理に使用
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# リクエストヘッダの"Authorization"に含まれるBearerトークンを取得する
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# 平文のパスワードとハッシュ化されたパスワードの一致判定
def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password,hashed_password)

# パスワードのハッシュ化を行う
def get_password_hash(password):
    return pwd_context.hash(password)

# ユーザー認証を行い、認証が正常に済んだらユーザー情報を返す
def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user

# トークン有効期限を加えたJSONデータを暗号化してトークンとして返す
def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

# ユーザー情報を用いてユーザーDBから該当ユーザーを検索する
def get_user(db, username:str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)

# リクエストヘッダに含まれるJWTからユーザー情報を取得し、該当するユーザーの情報を返す
async def get_current_user(token: Annotated[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(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    return user 

# ユーザー情報が有効かどうかを調べ、有効だったらユーザー情報を返す
async def get_current_active_user(
    current_user: Annotated[User, Depends(get_current_user)]
):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

app = FastAPI()

パスの定義

次はこのアプリに必要になるパスと、そこにアクセスした際の処理を定義します。

最初は"/token"というパスです。ユーザーの認証を行い、トークンを発行する処理を行います。OAuth2PasswordRequestFormというクラスの変数を受け付けますが、このクラスはメンバー変数にusernameとpasswordを持つクラスです。login関数の中でそれらの値を使ってユーザー認証を行い、最終的にBearerトークンを発行します。

@app.post("/token")
async def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=400, 
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"}
        )
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    return {"access_token":access_token, "token_type": "bearer"}

"/users/me/items"は発行されたトークンを使ってユーザー自身が所有するアイテムの情報を閲覧するためのパスです。このパスで実行される関数では、依存関係をたどるとOAuth2PasswordBearerクラスのオブジェクトoauth2_schemeの__call__が最初に実行されます。oauth2_schemeの__call__の中ではリクエストのヘッダの中の"Authorization"に指定されているBearerトークンを取得しています。そのトークンを使って次に呼ばれるget_current_user関数の中で該当するユーザー情報を取得しUserクラスのオブジェクトを返し、get_current_active_user関数によって有効な会員( disableがFalse)かのチェックを行い、有効であればUserを、無効であれば400エラーをあげます。

そしてこのread_own_itemsで該当Userのメンバ変数usernameを使って戻り値を生成します。

@app.get("/users/me/items")
async def read_own_items(
    current_user: Annotated[User, Depends(get_current_active_user)]
):
    return [{"item_id": "Foo", "owner": current_user.username}]

動作をInteractive API docsで確認する

ここまでのコードでアプリケーションにどんな動作が実装されたのかはアプリケーションを起動するとアクセスできるInteractive API docs上で確認することが出来ます。ローカル環境でアプリケーションをポート8000で立ち上げて、"http://localhost:8000/docs"にブラウザからアクセスします。