Authentication System with FastAPI + SQLModel

FastAPIでログイン機能を15分で実装するためのTips。

Offitial Tutorial

fastapi.tiangolo.com

ログイン機能実装の勉強にはなるけど、わざわざ自分で書く必要もないよね。

pip install fastapi

SQLModel

sqlmodel.tiangolo.com

FastAPIの開発者が手掛けているSQLデータベースをPythonから操作するためのライブラリ。

SQLAlchemyをベースとしていて、それにPydanticを加えてより使いやすくしたもの。

SQLライブラリで迷っているならとりあえずチュートリアルをやってみるとよい。

Installation

pip install sqlmodel

FastAPI-Users

fastapi-users.github.io

I'll use FastAPI-Users though it's a smaller community than other major libraries such as FastAPI. This library sounds best for what I'm trying to do.

小さめのコミュニティだけど、FastAPI-Usersというライブラリを使わせてもらう。

SQLModelは一応サポートしているけどドキュメントはないので、ソースコード読みながら勘で実装した。 そんな人の時間を節約するためのメモ。

Installation

pip install fastapi-users
pip install fastapi-users-db-sqlmodel  # additional package for SQLModel

Concepts

FastAPI-Users`は次の4つの要素で構成することで、ログイン機能を単純化している。

  • モデル
    データベースに保存する情報をまとめたユーザモデル。
  • データベースアダプタ
    データベースとやり取りする。
    SQLAlchemy, MonagoDB, SQLModel, etc...
    今回はSQLModelで実装。
  • 認証バックエンド
    JWT or CookieCookie実装はうまく機能しなかったのでJWTで実装。
  • ユーザマネージャ
    データベース操作前後の処理のインターフェースを提供。
    今回はいじらない。

Code

インポート関係がわかりやすいように、敢えて分割してインポートしています。

login.py

## Define Models ##
# See also https://fastapi-users.github.io/fastapi-users/configuration/models/
from fastapi_users import models
from fastapi_users_db_sqlmodel import SQLModelBaseUserDB

class User(models.BaseUser):
    pass


class UserCreate(models.BaseUserCreate):
    pass


class UserUpdate(models.BaseUserUpdate):
    pass


class UserDB(SQLModelBaseUserDB, table=True):
    pass
    # See also https://sqlmodel.tiangolo.com/#create-a-sqlmodel-model



## Database Adapter ##
import sqlmodel
from fastapi_users_db_sqlmodel import SQLModelUserDatabase

# create engine. See also https://sqlmodel.tiangolo.com/#write-to-the-database
sqlengine = sqlmodel.create_engine('sqlite:///database.db', connect_args={"check_same_thread": False})

def get_user_db(): # used later
    with sqlmodel.Session(sqlengine) as session:
        yield SQLModelUserDatabase(UserDB, session)


## Authentication Backend(JWT)##
# See also https://fastapi-users.github.io/fastapi-users/configuration/authentication/jwt/
# About JWT, see https://fastapi.tiangolo.com/tutorial/security/oauth2-jwt/#jwt
from fastapi_users.authentication.jwt import JWTAuthentication

SECRET = 'SECRET_KEY_FOR_AUTH'
jwt_authentication = JWTAuthentication(secret=SECRET, lifetime_seconds=3600)


## User Manager ##
# See also https://fastapi-users.github.io/fastapi-users/configuration/user-manager/
from fastapi_users import BaseUserManager
from fastapi import Depends

class UserManager(BaseUserManager[UserCreate, UserDB]):
    user_db_model = UserDB
    reset_password_token_secret = SECRET
    verification_token_secret = SECRET


def get_user_manager(user_db=Depends(get_user_db)):
    yield UserManager(user_db)


## Add to FastAPI ##
from fastapi import FastAPI
from fastapi_users import FastAPIUsers

app = FastAPI() # create an app

fastapi_users = FastAPIUsers(  # Interface of FastAPI-Users
    get_user_manager,
    [jwt_authentication],
    User,
    UserCreate,
    UserUpdate,
    UserDB,
)

# Add basic routers for authentication
app.include_router(
    fastapi_users.get_auth_router(jwt_authentication),
    prefix="/auth",
    tags=["auth"],
)

app.include_router(
    fastapi_users.get_register_router(),
    prefix="/auth",
    tags=["auth"],
)

app.include_router(
    fastapi_users.get_verify_router(),
    prefix="/auth",
    tags=["auth"],
)

app.include_router(
    fastapi_users.get_reset_password_router(),
    prefix="/auth",
    tags=["auth"],
)

# Add routers returning user information.
app.include_router(
    fastapi_users.get_users_router(),
    prefix="/users",
    tags=["users"]
)

# Create models in database
sqlmodel.SQLModel.metadata.create_all(sqlengine) 

Run

pip install uvicorn
uvicorn login:app

Getting access to http://localhost:8000/docs, you can see web documents automatically created by FastAPI.

Usage

The following is a sample using requests library.

import requests

URL = 'http://localhost:8000'

# User Registeration
def register(email: str, password: str) -> requests.Response:
    data = {
        'email': email,
        'password': password
    }
    response = requests.post(f'{URL}/auth/register', json=data)
    return response


# Login
def login(email: str, password: str) -> str:
    data = {
        'email': email,
        'password': password
    }
    response = requests.post(f'{URL}/auth/login', data=data)
    token = response.json().get('access_token', '')
    return token


# Get User Information
def get_me(token: str) -> dict:
    headers = {'Authorization': f'Bearer {token}'}
    response = requests.get(f'{URL}/users/me', headers=headers}
    return response.json()

ToDo

  • OAuth2の実装
  • SQLエンジンをAsync化

結局追加機能つけてアプリケーションに組み込もうと思ったら公式ドキュメント見に行くしかなくてこの記事の意味はあるのかと。

SQLAlchemy使って実装するよりはSQLModelで実装した方がシンプルだから一応意味はあるのかな?