🐧

FastAPIアプリケーション〜環境設定からログイン処理まで〜

2024/01/28に公開

Python

FastAPI

sqlmodel

tech

開発環境の設定(docker)

dockerの設定をしていきます。

backend/Dockerfile

FROM python:latest
WORKDIR /backend
ENV PYTHONDONTWRITEBYTECODE 1
ENV PYTHONBUFFERED 1

# pipを使ってpoetryをインストール
RUN pip install poetry

# poetryの定義ファイルをコピー (存在する場合)
COPY pyproject.toml* poetry.lock* ./

# poetryでライブラリをインストール (pyproject.tomlが既にある場合)
RUN poetry config virtualenvs.in-project false
RUN if [ -f pyproject.toml ]; then poetry install; fi

# poetryの実行用コマンド
ENTRYPOINT ["poetry", "run", "uvicorn", "v1.app.main:app", "--host", "0.0.0.0", "--reload"]
# ENTRYPOINT ["poetry", "run"]

COPY . /backend

compose.yml

services:
    app_backend: # 任意の名称で設定
    container_name: app_backend # 任意の名称で設定
    build:
      context: ./backend # Dockerfileのあるdir
      dockerfile: Dockerfile # Dockerfileの名称
    volumes: #[ホスト側の相対Path]:[コンテナの絶対Path]
      - ./backend/:/backend/
    environment:
      TZ: Asia/Tokyo
    ports: #[hostのport]:[docker内のport]
      - 8000:8000

FastAPI用のdocker設定はとりあえずこれで動くかと思います。

記載したら、

bash

docker compose build

を実行します。

docker経由でpoetryを実行し、必要なライブラリなどをインストール

FastAPIで開発するためのライブラリなどをインストールしていきます。

bash

docker compose run --entrypoint "poetry init --name [コンテナ名] [インストールするライブラリ]" [コンテナ名]

【インストールするライブラリ】
--dependency fastapi --dependency uvicorn[standard] --dependency sqlmodel --dependency psycopg2 --dependency alembic --dependency psycopg2-binary --dependency uuid

これを実行すると、pyproject.tomlが生成される
生成されたpyproject.tomlにはインストールするライブラリが記載されているため、これを使用してライブラリをインストールします。

bash

docker compose run --entrypoint "poetry install" [コンテナ名]

上記コマンドを実行後は、コンテナの再buildをしておく

bash

docker compose build --no-cache

bash

docker compose up

これを実行し、コンテナを起動します。

FastAPIの初期設定

メインの処理となるファイルを作成します。
今回はbackendディレクトにv1ディレクトリを作成し、その中にコードを書いていきます。

コードはこちら

backend/v1/app/main.py

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

import time

from ..core.config import settings

def get_application():
  
  app = FastAPI(
    title = settings.AUTH_API_PROJECT_NAME,
    version = settings.AUTH_API_PROJECT_VERSION,
    description = settings.AUTH_API_DESCRIPTION,
  )
  
  origins = [
    'http://localhost:3000',
  ]
  
  app.add_middleware (
    CORSMiddleware,
    allow_origins = origins,
    allow_credentials = True,
    allow_methods = ['*'],
    allow_headers = ['*'],
  )
  
  return app

app = get_application()

@app.middleware('http')
async def add_process_time_header(request: Request, call_next):
  start_time = time.time()
  response = await call_next(request)
  process_time = time.time() - start_time
  response.headers['X-Process-Time'] = str(process_time)
  return response

参考にした記事はこちら

アプリケーションの設定

FastAPIのアプリケーションを作っていきます。
まずは全体の共通設定とデータベースの設定を書いていきます。

コードはこちら

backend/v1/core/config.py

import os

from passlib.context import CryptContext
from fastapi.security import OAuth2PasswordBearer

from dotenv import load_dotenv

load_dotenv() # ()には必要に応じてファイルパスを記載

class Settings:
  # 
  AUTH_API_PROJECT_NAME: str = "Application"
  AUTH_API_PROJECT_VERSION: str = "0.1.0"
  AUTH_API_DESCRIPTION: str = """
  アプリケーションに関する説明などを記載（markdownで書けば、反映される）
  """

  # DB関連の設定
  POSTGRES_USER: str = os.environ['POSTGRES_USER']
  POSTGRES_PASSWORD: str = os.environ['POSTGRES_PASSWORD']
  POSTGRES_DOCKER: str = os.environ['POSTGRES_DOCKER']
  POSTGRES_PORT: str = os.environ['POSTGRES_PORT']
  POSTGRES_DB: str = os.environ['POSTGRES_DB']
  
  pwd_context = CryptContext(schemes = ['bcrypt'], deprecated = 'auto')
  oauth2_schema = OAuth2PasswordBearer(tokenUrl = '/user/token')

# インスタンス化
settings = Settings()

PostgresQLのための環境変数設定

os.environ['key']は、環境変数（keyで指定）の読み込み（環境変数がない場合はkeyErrorを返す）
os.environ['key']は、keyのほか、第２引数としてデフォルト値の設定が可能
環境変数キーが存在しない場合はデフォルト値を返す
compose.ymlにデータベース関連の情報を書いている場合、その情報に変更を加えたら、.envファイルの情報も更新が必要

PostgreSQLのデータベースURLの設定

【同期処理の場合】
DATABASE_URL = f"postgresql://{POSTGRES_USER}:{POSTGRES_PASSWORD}@{POSTGRES_DOCKER}:{POSTGRES_PORT}/{POSTGRES_DB}"

【非同期処理の場合】
DATABASE_URL = f"postgresql+asyncpg://{POSTGRES_USER}:{POSTGRES_PASSWORD}@{POSTGRES_DOCKER}:{POSTGRES_PORT}/{POSTGRES_DB}"

データベースの設定

settingsに書いた設定を使用して、データベースの設定をしていきます。

コードはこちら

backend/v1/db/db.py

from sqlmodel import SQLModel, create_engine, Session
from sqlmodel.sql.expression import Select, SelectOfScalar

from ..core.config import settings

DATABASE_URL = settings.DATABASE_URL

SelectOfScalar.inherit_cache = True
Select.inherit_cache = True

# create_engineの初期化
engine = create_engine(DATABASE_URL, echo = True)

# tableの作成
def init():
  SQLModel.metadata.create_all(engine)
  
# FastAPIのDependencyの設定
def get_session():
  with Session(engine) as session:
    yield session

compose.ymlに追記

はじめに作成したcompose.ymlにデータベース関連の設定を追記します。

コードはこちら

compose.yml

services:    
  app_backend:
    container_name: app_backend
    build:
      context: ./backend # Dockerfileのあるdir
      dockerfile: Dockerfile # Dockerfileの名称
    volumes: #[ホスト側の相対Path]:[コンテナの絶対Path]
      - ./backend/:/backend/
    environment:
      TZ: Asia/Tokyo
    ports: #[hostのport]:[docker内のport]
      - 8000:8000
+    depends_on:
+      - app_db
  
+  app_db:
+    container_name: app_postgres
+    image: postgres:latest
+    volumes:
+      - postgres:/db
+    env_file:
+      - ./backend/.env
+    environment:
+      POSTGRES_DB: postgres
+      POSTGRES_USER: postgres
+      POSTGRES_PASSWORD: postgres
+      TZ: Asia/Tokyo
+      PGTZ: Asia/Tokyo
+    ports:
+      - 5432:5432

+ volumes: # add
+   postgres:

【追記内容】

データベースの設定として、app_dbを追記
app_dbとbackendを接続するため、backendの設定に「depends_on」を追記
データベースの永続化のために、volumesを設定

FastAPIのスキーマ設定

データベースの構造にもなるスキーマを書いていきます。
今回はSQLModelを使用しています。

コードはこちら

backend/v1/model/user_model.py

from sqlmodel import SQLModel, Field, Relationship, AutoString
from pydantic import EmailStr
from typing import Optional, List, TYPE_CHECKING
from datetime import datetime, date

import uuid as uuid_pkg

class BaseUser(SQLModel):
  username: str = Field(nullable = False, default = None, unique = True)
  email: EmailStr =Field(unique = True, index = True, sa_type=AutoString)
  name: str = Field(index = True, unique = False, nullable = True)
  is_active: bool = Field(default = True)
  is_superuser: bool = Field(default = False)

class BaseUserModel(BaseUser, table = True):
  id: Optional[int] = Field(default = None, primary_key = True)
  uuid: str = Field(default_factory = uuid_pkg.uuid4, nullable = False)
  hashed_password: str = Field(nullable = False)
  member_id: str = Field(nullable = True)
  created_at: datetime = Field(default = datetime.now(), nullable = False)
  updated_at: datetime = Field(default_factory = datetime.now, nullable = False, sa_column_kwargs = {'onupdate': datetime.now})
  refresh_token: str = Field(nullable = True)
  
class BaseUserOut(SQLModel):  # response用のスキーマ（パスワードを除く）
  id: int
  uuid: str
  name: str
  username: str
  email: str
  member_id: str
  is_active: bool
  is_superuser: bool
  created_at: datetime
  updated_at: datetime
  refresh_token: str

class BaseUserCreate(SQLModel):
  username: str
  email: EmailStr
  password: str

class BaseUserRead(BaseUser):
  id: int
  uuid: str
  created_at: datetime
  updated_at: datetime
  
class BaseUserUpdate(SQLModel): # データ更新の項目は任意のため全てOptionalで設定
  name: Optional[str] = None
  username: Optional[str] = None
  email: Optional[EmailStr] = None
  member_id: Optional[str] = None

class Token(SQLModel):
  access_token: str
  refresh_token: str
  token_type: str

class TokenData(SQLModel):
  uuid: str | None = None

FastAPIでユーザー情報取得するためのコードを書く

APIとしてデータの取得をするためのコードを書きます。

コードはこちら

backend/v1/crud/auth.py

import os
from fastapi import Depends, HTTPException, status
from datetime import datetime, timedelta

# schema
from ..model.user_model import BaseUserModel, BaseUser, BaseUserOut

# sqlmodel
from sqlmodel import select

# settings
from ..core.config import settings

# .env
from dotenv import load_dotenv
load_dotenv()

# User登録前確認用
def check_user(userIn, session):
  # BaseUserModel(=Table)からデータを探して、既に登録済みかどうかを検証
  checked_user = session.exec(
    select(BaseUserModel)
    .where(BaseUserModel.username == userIn.username)
    .where(BaseUserModel.email == userIn.email)
  ).first()
  return checked_user

# passwordのハッシュ化
def get_hashed_password(password):
  return settings.pwd_context.hash(password)

# ハッシュ化されたpasswordの検証
def verify_password(password, hashed_password):
  return settings.pwd_context.verify(password, hashed_password)

# 全てのユーザーを取得
def get_users(session):
  users = select(BaseUserModel)
  user_list: list[BaseUserOut] = session.exec(users)
  return user_list

# ユーザー情報の更新
def update_user(uuid, user, session):
    db_user = select(BaseUserModel).where(BaseUserModel.uuid == uuid)
    result = session.exec(db_user)
    pickuped_user = result.first()
    if not pickuped_user:
      raise HTTPException(
      status_code = 404,
      detail = 'User not found'
    )
    user_data = user.model_dump(exclude_unset = True)
    for key, value in user_data.items():
      setattr(pickuped_user, key, value)
    session.add(pickuped_user)
    session.commit()
    session.refresh(pickuped_user)
    return pickuped_user

# 登録済みユーザーの削除
def delete_user(uuid, session):
  delete_user = get_user_by_uuid(session, uuid)
  if not delete_user:
    raise HTTPException(
      status_code = 404,
      detail = 'User not found.'
    )
  session.delete(delete_user)
  session.commit()
  return {"message": "User was deleted."}

# usernameからユーザーを取得
def get_user(session, username: str):
  return session.query(BaseUserModel).filter(BaseUserModel.username == username).first()

# uuidからユーザーを取得
def get_user_by_uuid(uuid: str, session):
  return session.query(BaseUserModel).filter(BaseUserModel.uuid == uuid).first()

FastAPIでAPI出力

次にAPIドキュメント（OpenAPIで出力される）を生成するためのコードを書いていきます。

コードはこちら

backend/v1/router/user.py

from fastapi import APIRouter, Depends, HTTPException

# functions
from ..crud.user import (
  check_user,
  get_users,
  delete_user,
  update_user,
  get_hashed_password
)

# schema
from ..model.user_model import (
  BaseUserOut,
  BaseUserCreate,
  BaseUserRead,
  BaseUserModel,
  BaseUserUpdate
)

import uuid as uuid_pkg

# db
from ..db.db import get_session

# SQLModel
from sqlmodel import Session, select

router = APIRouter(
  prefix = '/user',
  tags = ['User']
)

# ユーザー登録
@router.post('/signup', response_model = BaseUserOut)
def create_user(userIn: BaseUserCreate, session: Session = Depends(get_session)):
  """サインアップ用API"""
  user = check_user(userIn, session)
  if user:
    raise HTTPException(
      status_code = 409,
      detail = 'UsernameまたはEmailアドレスが既に使用されています'
    )
  else:
    new_user = BaseUserModel(
      name = '',
      username = userIn.username,
      email = userIn.email,
      member_id = '',
      is_active = True,
      is_superuser = False,
      hashed_password = get_hashed_password(userIn.password),
      refresh_token = ''
    )

    session.add(new_user)
    session.commit()
    session.refresh(new_user)
    return new_user

# 全てのユーザーを取得
@router.get('/all_user', response_model = list[BaseUserRead])
def get_all_user(
  *,
  session: Session = Depends(get_session)
) -> list[BaseUserRead]:
  """
  全てのUser情報を取得できます
  """
  users = get_users(session=session)
  return users

# ユーザー情報の更新
@router.patch('/user/{uuid}', response_model = BaseUserRead)
def update_db_user(
  *,
  session: Session = Depends(get_session),
  user: BaseUserUpdate,
  uuid: str
):
  return update_user(uuid, user, session)

# ユーザー情報の削除
@router.delete('/delete/{uuid}')
def delete_db_user(
  *,
  session: Session = Depends(get_session),
  uuid: str,
):
  return delete_user(uuid, session)

main.pyで設定したrouterを読み込む

main.pyでrouterを読み込みます。

コードはこちら

backend/v1/app/main.py

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware

import time

from ..core.config import settings
+ from ..router import user

def get_application():
  
  app = FastAPI(
    title = settings.AUTH_API_PROJECT_NAME,
    version = settings.AUTH_API_PROJECT_VERSION,
    description = settings.AUTH_API_DESCRIPTION,
  )
  
  origins = [
    'http://localhost:3000',
  ]
  
  app.add_middleware (
    CORSMiddleware,
    allow_origins = origins,
    allow_credentials = True,
    allow_methods = ['*'],
    allow_headers = ['*'],
  )
  
  return app

app = get_application()

# router settings
+ app.include_router(user.router)

@app.middleware('http')
async def add_process_time_header(request: Request, call_next):
  start_time = time.time()
  response = await call_next(request)
  process_time = time.time() - start_time
  response.headers['X-Process-Time'] = str(process_time)
  return response

２箇所追記して、設定したrouterを読み込みます。

データベースの設定

データベースはalembicを使用します。
まずは、dockerを立ち上げ、alembicをインストールします。

bash

docker compose exec [コンテナ名] poetry add alembic

立ち上げたdockerのコンテナのbashに入ります（別のターミナルから実行）。

bash

docker compose exec [コンテナ名] bash

コンテナのbashに入ったら、以下のコマンドを実行します。

bash

poetry run alembic init migrations

migrationsディレクトリが作成され、いくつかファイルが作成されるため、これらのファイルに追記していきます。

コードはこちら

backend/migrations/env.py

+ import os
from logging.config import fileConfig

from sqlalchemy import engine_from_config
from sqlalchemy import pool

+ from v1.model.user_model import BaseUserModel
from alembic import context
+ from sqlmodel import SQLModel

# add (env file settings)
+ from dotenv import load_dotenv
+ load_dotenv()

# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config

# Interpret the config file for Python logging.
# This line sets up loggers basically.
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# add your model's MetaData object here
# for 'autogenerate' support
# from myapp import mymodel
# target_metadata = mymodel.Base.metadata
- target_metadata = None
+ target_metadata = BaseUserModel.metadata

# other values from the config, defined by the needs of env.py,
# can be acquired:
# my_important_option = config.get_main_option("my_important_option")
# ... etc.


def run_migrations_offline() -> None:
    """Run migrations in 'offline' mode.

    This configures the context with just a URL
    and not an Engine, though an Engine is acceptable
    here as well.  By skipping the Engine creation
    we don't even need a DBAPI to be available.

    Calls to context.execute() here emit the given string to the
    script output.

    """
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )

    with context.begin_transaction():
        context.run_migrations()


def run_migrations_online() -> None:
    """Run migrations in 'online' mode.

    In this scenario we need to create an Engine
    and associate a connection with the context.

    """

+    # set_section_option (セクション: str 、名前: str 、値: str )
+    config.set_section_option('alembic', 'POSTGRES_USER', os.environ['POSTGRES_USER']) # add
+    config.set_section_option('alembic', 'POSTGRES_PASSWORD', os.environ['POSTGRES_PASSWORD']) # add
+    config.set_section_option('alembic', 'POSTGRES_DOCKER', os.environ['POSTGRES_DOCKER']) # add
+    config.set_section_option('alembic', 'POSTGRES_PORT', os.environ['POSTGRES_PORT']) # add
+    config.set_section_option('alembic', 'POSTGRES_DB', os.environ['POSTGRES_DB']) # add
    
    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection, target_metadata=target_metadata
        )

        with context.begin_transaction():
            context.run_migrations()


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

コードはこちら

script.py.mako

"""${message}

Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}

"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa
+ import sqlmodel
${imports if imports else ""}

# revision identifiers, used by Alembic.
revision: str = ${repr(up_revision)}
down_revision: Union[str, None] = ${repr(down_revision)}
branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)}
depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)}


def upgrade() -> None:
    ${upgrades if upgrades else "pass"}


def downgrade() -> None:
    ${downgrades if downgrades else "pass"}

backend/alembic.ini

# A generic, single database configuration.

[alembic]
# path to migration scripts
script_location = migrations

prepend_sys_path = .

version_path_separator = os  # Use os.pathsep. Default configuration used for new projects.

+ sqlalchemy.url = postgresql://%(POSTGRES_USER)s:%(POSTGRES_PASSWORD)s@%(POSTGRES_DOCKER)s:%(POSTGRES_PORT)s/%(POSTGRES_DB)s


[post_write_hooks]

# Logging configuration
[loggers]
keys = root,sqlalchemy,alembic

[handlers]
keys = console

[formatters]
keys = generic

[logger_root]
level = WARN
handlers = console
qualname =

[logger_sqlalchemy]
level = WARN
handlers =
qualname = sqlalchemy.engine

[logger_alembic]
level = INFO
handlers =
qualname = alembic

[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic

[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S

上記のとおり追記したら、下記により再度コンテナに入り、マイグレートしていきます。

bash

docker compose exec [コンテナ名] bash

bash

alembic revision --autogenerate -m "init"
# poetryを使用する場合は、先頭に「poetry run」をつけて実行します
# "init"の部分はマイグレーションファイル名に記載されます（いわゆるコメント）

bash

alembic upgrade head
# poetryを使用する場合は、先頭に「poetry run」をつけて実行します
# マイグレートを実行

【参考記事】
（alembicについて）

（%記法について）

全体の流れはこれまでのステップにより実行・反映可能です。
上記までの流れを行えば、基本的に新たにAPIの設定を書いても、同じように反映させることができると思います。

認証機能の設定

認証に関するコードを書いていきます。

コードはこちら

backend/v1/crud/auth.py

import os
from fastapi import Depends, HTTPException
from datetime import datetime, timedelta

# auth
from jose import jwt

# sqlmodel
from sqlmodel import Session

# db
from ..db.db import get_session

# settings
from ..core.config import settings

# functions
from .user import verify_password, get_user, get_user_by_uuid

# .env
from dotenv import load_dotenv
load_dotenv()

def authenticate(username, password, session):
  # usernameとpasswordで認証し、User情報を返す
  user = get_user(session, username)
  if verify_password(password, user.hashed_password) != True:
    raise HTTPException(
      status_code = 401,
      detail = 'パスワードが一致しません'
    )
  return user

def create_tokens(uuid: str):
  """パスワード認証を行いtokenを生成"""
  access_payload = {
    'token_type': 'access_token',
    'exp': datetime.utcnow() + timedelta(minutes=60),
    'uuid': uuid,
  }
  
  refresh_payload = {
    'token_type': 'refresh_token',
    'exp': datetime.utcnow() + timedelta(days=90),
    'uuid': uuid,
  }
  
  # tokenを生成
  access_token = jwt.encode(access_payload, os.environ['SECRET_KEY'], algorithm = os.environ['ALGORITHM'])
  refresh_token = jwt.encode(refresh_payload, os.environ['SECRET_KEY'], algorithm = os.environ['ALGORITHM'])
  
  return {
    'access_token': access_token,
    'refresh_token': refresh_token,
    'token_type': 'bearer'
  }

  
def get_current_user_from_token(session, token: str, token_type: str):
  # access_tokenをdecodeしてpayloadを取得（有効期限と署名は自動検証）
  payload = jwt.decode(
    token,
    os.environ['SECRET_KEY'],
    algorithms = os.environ['ALGORITHM']
  )
  
  # DBからユーザーを取得
  user = get_user_by_uuid(payload['uuid'], session)
  
  # refresh_tokenの場合、受け取ったtokenとDBに保存されているものを照合
  if payload['token_type'] != token_type:
    raise HTTPException(
      status_code = 401,
      detail = 'トークンタイプが一致していません'
    )
  if token_type == 'refresh_token' and user.refresh_token != token:
    print(user.refresh_token, '¥n', token)
    raise HTTPException(
      status_code = 401,
      detail = 'refresh_tokenが一致しません'
    )

  return user


def get_current_user(session: Session = Depends(get_session), token: str = Depends(settings.oauth2_schema)):
  print(token)
  return get_current_user_from_token(session, token, 'access_token')


def get_user_from_refresh_token(
  token: str,
  session: Session = Depends(get_session),
):
  return get_current_user_from_token(session, token, 'refresh_token')

認証に関するAPIの設定

コードはこちら

backend/v1/router/auth.py

from fastapi import APIRouter, Depends
from fastapi.security import OAuth2PasswordRequestForm

# functions
from ..crud.auth import (
  get_current_user,
  get_user_from_refresh_token,
  create_tokens,
  authenticate,
)

# schema
from ..model.user_model import (Token, BaseUserModel)

# db
from ..db.db import get_session

# SQLModel
from sqlmodel import Session

router = APIRouter(
  prefix = '/user',
  tags = ['Token認証']
)

@router.post('/token', response_model = Token)
def login(
  username: str = Form(...),
  password: str = Form(...),
  session: Session = Depends(get_session),
):
  """Tokenの発行(基本的には直接のアクセスはない)"""
  user = authenticate(username, password, session)
  token = create_tokens(user.uuid)
  user.refresh_token = token['refresh_token']
  session.add(user)
  session.commit()
  session.refresh(user)
  return token


@router.get('/refresh_token', response_model = Token)
def refresh_token(
  session: Session = Depends(get_session),
  current_user: BaseUserModel = Depends(get_user_from_refresh_token)
):
  token = create_tokens(current_user.uuid)
  current_user.refresh_token = token['refresh_token']
  session.add(current_user)
  session.commit()
  session.refresh(current_user)
  return token


@router.get('/user/me', response_model = BaseUserModel)
def get_me(current_user: BaseUserModel = Depends(get_current_user)):
  """access_tokenからログインUserを取得"""
  return current_user

main.pyでもこのrouterの設定を読み込みます。

コードはこちら

backend/v1/app/main.py

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.templating import Jinja2Templates

import time

from ..core.config import settings
from ..router import auth, user

def get_application():
  
  app = FastAPI(
    title = settings.AUTH_API_PROJECT_NAME,
    version = settings.AUTH_API_PROJECT_VERSION,
    description = settings.AUTH_API_DESCRIPTION,
  )
  
  origins = [
    'http://localhost:3000',
  ]
  
  app.add_middleware (
    CORSMiddleware,
    allow_origins = origins,
    allow_credentials = True,
    allow_methods = ['*'],
    allow_headers = ['*'],
  )
  
  return app

app = get_application()

# router settings
+ app.include_router(auth.router)
app.include_router(user.router)

@app.middleware('http')
async def add_process_time_header(request: Request, call_next):
  start_time = time.time()
  response = await call_next(request)
  process_time = time.time() - start_time
  response.headers['X-Process-Time'] = str(process_time)
  return response

jinja2でログインを試す

作成したAPIを使用して、実際にログイン処理ができるかどうかを確認します。
まずは表示用のHTML、JavaScriptを書いていきます。

コードはこちら

backend/templates/index.html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Login Page</title>
</head>
<body>
    <h2>Login</h2>
    <form id="loginForm">
        <label for="username">Username:</label>
        <input type="text" id="username" name="username" required>
        <br>
        <label for="password">Password:</label>
        <input type="password" id="password" name="password" required>
        <br>
        <button type="button" onclick="login()">Login</button>
    </form>

    <script>
        async function login() {
            const username = document.getElementById("username").value;
            const password = document.getElementById("password").value;

            const response = await fetch('http://0.0.0.0:8000/user/token', {
                method: 'POST',
                headers: {
                    'Content-Type': 'application/x-www-form-urlencoded',
                },
                body: `username=${encodeURIComponent(username)}&password=${encodeURIComponent(password)}`,
            });

            if (!response.ok) {
              throw new Error("Invalid username or password");
            }

            const data = await response.json();

            if (response.ok) {
                // ログイン成功
                console.log("Access Token:", data.access_token);
                console.log("Refresh Token:", data.refresh_token);

                // アクセストークンをCookieに保存
                document.cookie = `access_token=${data.access_token}; path=/`;
                document.cookie = `refresh_token=${data.refresh_token}; path=/`;
               refreshedData.access_token);
                // ログイン後のページに遷移
                window.location.href = "/userpage.html";
            } else {
                // ログイン失敗
                console.error("Login failed:", data.detail);
            }
        }
    </script>
</body>
</html>

同じく、ログイン後のページも用意します。

コードはこちら

backend/templates/userpage.html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>User Info Page</title>
</head>
<body>
    <h1>User Info Page</h1>
    <h2 id="username"></h2>
    <h2 id="email"></h2>

    <script>
        // Cookieから指定した名前のCookie値を取得する関数
        function getCookie(name) {
            const cookies = document.cookie.split(';');
            for (const cookie of cookies) {
                const [cookieName, cookieValue] = cookie.trim().split('=');
                if (cookieName === name) {
                    return decodeURIComponent(cookieValue);
                }
            }
            return null;
        }

        // ページが読み込まれた際にCookieからユーザー情報を取得して表示
        document.addEventListener("DOMContentLoaded", async function() {
            const accessToken = getCookie("access_token");

            if (accessToken) {
                const response = await fetch('http://0.0.0.0:8000/user/user/me', {
                method: 'GET',
                headers: {
                    'Content-Type': 'application/json',
                    'Authorization': `Bearer ${accessToken}`,
                },
                });

                if (!response.ok) {
                    throw new Error("Invalid username or password");
                }
    
                const data = await response.json();

                if (data) {
                    // ユーザー情報を表示
                    document.getElementById("username").innerText = `User name: ${data.username}さんがログイン中です`;
                    document.getElementById("email").innerText = `Email: ${data.email}`;
                } else {
                    console.error("User info not found in the cookie.");
                }
            }else {
                console.error("Access token not found in the cookie.");
            }
        });
    </script>
</body>
</html>

次に、templateの表示用の設定をしていきます。
FastAPIでは、jinja2を使用した設定方法が用意されていますので、これに従います。

コードは以下のとおりです。

コードはこちら

backend/v1/app/main.py

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.templating import Jinja2Templates

import time

from ..core.config import settings
from ..router import auth, user

def get_application():
  
  app = FastAPI(
    title = settings.AUTH_API_PROJECT_NAME,
    version = settings.AUTH_API_PROJECT_VERSION,
    description = settings.AUTH_API_DESCRIPTION,
  )
  
  origins = [
    'http://localhost:3000',
  ]
  
  app.add_middleware (
    CORSMiddleware,
    allow_origins = origins,
    allow_credentials = True,
    allow_methods = ['*'],
    allow_headers = ['*'],
  )
  
  return app

app = get_application()

# jinja
+ templates = Jinja2Templates(directory = 'templates')

# router settings
app.include_router(auth.router)
app.include_router(user.router)

@app.middleware('http')
async def add_process_time_header(request: Request, call_next):
  start_time = time.time()
  response = await call_next(request)
  process_time = time.time() - start_time
  response.headers['X-Process-Time'] = str(process_time)
  return response


+@app.get("/", tags = ['template example'])
+async def index(name: str, request: Request):
+    return templates.TemplateResponse("index.html", {
+        "name": name,
+        "request": request
+    })

+@app.get("/index.html", tags = ['login form'])
+async def index(request: Request):
+    return templates.TemplateResponse("index.html", {
+        "request": request
+    })

+@app.get("/userpage.html", tags = ['login form'])
+async def index(request: Request):
+    return templates.TemplateResponse("userpage.html", {
+        "request": request
+    })

dockerを起動して、指定したURL（今回の例では、ローカルの場合、http://0.0.0.0:8000/index.html）にアクセスすると設定したテンプレートが表示されます。

実際にログインできれば成功です。

以上が、FastAPIを使用したユーザー認証までの流れとなります。

開発環境の設定(docker)

docker経由でpoetryを実行し、必要なライブラリなどをインストール

FastAPIの初期設定

アプリケーションの設定

PostgresQLのための環境変数設定

PostgreSQLのデータベースURLの設定

データベースの設定

compose.ymlに追記

FastAPIのスキーマ設定

FastAPIでユーザー情報取得するためのコードを書く

FastAPIでAPI出力

main.pyで設定したrouterを読み込む

データベースの設定

認証機能の設定

認証に関するAPIの設定

jinja2でログインを試す

Discussion