【備忘録】FastAPI×MySQLでユーザーの新規作成およびログイン機能を実装する方法

備考

Python:3.9
MySQL:8.0.31
動作環境:MacOS
開発環境:PyCharm

事前準備

データベース作成済み

マイグレーション実行済み

1.Pytharmにて新規プロジェクトを作成する

2.プロジェクトを以下の構成になるようにファイルを作成する

フォルダ構成.

.
├── main.py
├── env.py
├── .env
├── requirements.txt
├── api
│  ├── certification_api.py
│  └── user_api.py
├── crud
│  ├── certification_crud.py
│  └── user_crud.py
├── database
│  ├── database.py
│  └── models.py
└── schema
   ├── certification_schema.py
   └── user_schema.py

3. requirements.txtファイルを以下の内容にする

requirements.txt

fastapi
uvicorn
sqlalchemy
pytz
python-dotenv
pydantic
passlib[bcrypt]
mysqlclient
itsdangerous
python-jose[cryptography]
jose

4. ターミナルにてコマンドを入力してモジュールをインストールする

ターミナル.

pip install -r requirements.txt

5. .envファイルを以下の内容にする

.env

# DB接続情報
DATABASE = 'mysql'
DB_USER = 'root'
DB_PASSWORD = 'MySQLのパスワード'
DB_HOST = 'localhost'
DB_PORT = '3306'
DB_NAME = '対象のDB名'

# JWT認証で使用するシークレットキー
SECRET_KEY = "任意の文字列"

6. env.pyファイルを以下の内容にする

env.py

import os
from dotenv import load_dotenv
from os.path import join, dirname

dotenv_path = join(dirname(__file__), '.env')
load_dotenv(dotenv_path)


class Env:
    # DB接続情報
    DATABASE = os.environ.get("DATABASE")
    DB_USER = os.environ.get("DB_USER")
    DB_PASSWORD = os.environ.get("DB_PASSWORD")
    DB_HOST = os.environ.get("DB_HOST")
    DB_PORT = os.environ.get("DB_PORT")
    DB_NAME = os.environ.get("DB_NAME")

    # JWT認証で使用するシークレットキー（任意の文字列）
    SECRET_KEY = os.environ.get("SECRET_KEY")

7. database.pyファイルを以下の内容にする

database.py

from env import Env
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.ext.declarative import declarative_base

# データベースに接続するための情報
database_url = '{}://{}:{}@{}:{}/{}?charset=utf8'.format(Env.DATABASE, Env.DB_USER, Env.DB_PASSWORD, Env.DB_HOST, Env.DB_PORT, Env.DB_NAME)
engine = create_engine(database_url, connect_args={"connect_timeout": 15}, echo=False, pool_recycle=10, pool_size=10, max_overflow=20)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()


# データベースに接続するための処理
def db_session():
    return scoped_session(sessionmaker(autocommit=False, autoflush=False, bind=engine))

8. models.pyファイルを以下の内容にする

models.py

from database.database import Base
from datetime import datetime
from sqlalchemy import Column, VARCHAR, INT, TEXT, DATETIME
import pytz

jst = pytz.timezone('Asia/Tokyo')


# 仮にユーザーを用意
class User(Base):
    __tablename__ = 'users'
    __table_args__ = {"mysql_collate": "utf8_general_ci"}
    id = Column(INT, primary_key=True, autoincrement=True, unique=True, nullable=False, index=True)
    name = Column(VARCHAR(255), primary_key=False, autoincrement=False, unique=False, nullable=False, index=False)
    email = Column(VARCHAR(255), primary_key=False, autoincrement=False,  unique=True, nullable=False, index=True)
    password = Column(TEXT, primary_key=False, autoincrement=False, unique=False, nullable=False, index=False)
    create_time = Column(DATETIME, primary_key=False, autoincrement=False, unique=False, nullable=False, index=False, default=lambda: datetime.now(jst))

9. certification_crud.pyファイルを以下の内容にする

certification_crud.py

from env import Env
from passlib.context import CryptContext
from datetime import datetime, timedelta
from jose import jwt

# ハッシュ化などの処理を使用するために定義する
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# JWT認証のシークレットキー（任意の文字列）
SECRET_KEY = Env.SECRET_KEY

# JWT認証のハッシュ化のアルゴリズム
ALGORITHM = 'HS256'


# ユーザーが入力したパスワードをハッシュ化したバスワードにする
def create_password_hash(user_password: str):
    return pwd_context.hash(user_password)


# ユーザーが入力したパスワードとハッシュ化したパスワードが一致するか確認する
def verify_password(plain_password: str, hashed_password: str):
    return pwd_context.verify(plain_password, hashed_password)


# アクセストークンを発行する
def create_access_token(data: dict, expires_delta: int):
    to_encode = data.copy()
    expire = datetime.now() + timedelta(days=expires_delta)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

10. user_crud.pyファイルを以下の内容にする

user_crud.py

from database import models
from schema import user_schema
from . import certification_crud
from sqlalchemy.orm import Session


# ユーザーを作成する処理
def create_user(db: Session, user: user_schema.CreateUser):
    db_user = models.User(**user.dict())

    # パスワードをハッシュ化する
    db_user.password = certification_crud.create_password_hash(db_user.password)

    db.add(db_user)
    db.commit()
    db.refresh(db_user)

    # 作成したユーザー情報を返す
    return db_user

11. certification_schema.pyファイルを以下の内容にする

certification_schema.py

from pydantic import BaseModel


# ユーザーのログイン
class UserLogin(BaseModel):
    email: str
    password: str
    is_remember_login: bool


# ユーザーのメールアドレスの被りチェック
class EmailCheck(BaseModel):
    email: str

12. user_schema.pyファイルを以下の内容にする

user_schema.py

from pydantic import BaseModel


# ユーザーを作成
class CreateUser(BaseModel):
    name: str
    email: str
    password: str

13. certification_api.pyファイルを以下の内容にする

certification_api.py

from database import models
from database import database
from fastapi import Depends
from fastapi import APIRouter
from schema import certification_schema
from sqlalchemy.orm.session import Session
from crud import certification_crud
from schema.certification_schema import EmailCheck

router = APIRouter()


# ユーザーのログイン
@router.post(path='/user/login', summary="ユーザーのログイン")
async def login_user(request: certification_schema.UserLogin, db: Session = Depends(database.db_session)):
    db_user = db.query(models.User).filter(models.User.email == request.email).first()
    if not db_user:
        # ユーザーが存在しない場合
        return {
            'status': False,
            'access_token': None,
            'message': 'ユーザーが存在しません',
            'data': None,
        }
    if not certification_crud.verify_password(request.password, db_user.password):
        # パスワードが一致しない場合
        return {
            'status': False,
            'access_token': None,
            'message': 'パスワードが一致しません',
            'data': None,
        }
    if request.is_remember_login:
        # ログイン情報を記憶する場合は30日間保存する
        access_token = certification_crud.create_access_token(data={'sub': str(db_user.id)}, expires_delta=30)
    else:
        # ログイン情報を記憶しない場合は1日間保存する
        access_token = certification_crud.create_access_token(data={'sub': str(db_user.id)}, expires_delta=1)
    return {
        'status': True,
        'access_token': access_token,
        'message': 'ログインに成功しました',
        'data': db_user,
    }


# メールアドレスの被りチェック
@router.post(path='/email/check', summary="メールアドレスの被りチェック")
def check_email(request: EmailCheck, db: Session = Depends(database.db_session)):
    db_user = db.query(models.User).filter(models.User.email == request.email).first()
    if db_user:
        return {
            'status': True,
            'message': '既に使用されているメールアドレスです',
            'data': True,
        }
    else:
        return {
            'status': True,
            'message': '使用が可能なメールアドレスです',
            'data': False,
        }

14. user_api.pyファイルを以下の内容にする

user_api.py

from schema import user_schema
from database import database
from fastapi import APIRouter, Depends
from sqlalchemy.orm.session import Session
from crud import user_crud, certification_crud
from sqlalchemy.exc import IntegrityError

router = APIRouter()


# ユーザーの作成
@router.post(path="/user/create", summary="ユーザーの作成")
async def post_user(db: Session = Depends(database.db_session), user: user_schema.CreateUser = None, is_remember_login: bool = None):
    try:
        db_user = user_crud.create_user(db, user)
        if db_user:
            if is_remember_login:
                # ログイン情報を記憶する場合は30日間保存する
                access_token = certification_crud.create_access_token(data={'sub': str(db_user.id)}, expires_delta=30)
            else:
                # ログイン情報を記憶しない場合は1日間保存する
                access_token = certification_crud.create_access_token(data={'sub': str(db_user.id)}, expires_delta=1)
            return {
                "status": True,
                'access_token': access_token,
                "message": "ユーザーの作成に成功しました",
                "data": db_user,
            }
        else:
            return {
                "status": False,
                'access_token': None,
                "message": "ユーザーの作成に失敗しました",
                "data": None,
            }
    except IntegrityError:
        return {
            "status": False,
            'access_token': None,
            "message": "このメールアドレスは既に登録されています",
            "data": None,
        }

15. main.pyファイルを以下の内容にする

main.py

import uvicorn
from env import Env
from database.database import db_session
from fastapi import FastAPI, Request
from starlette.middleware import Middleware
from starlette.middleware.cors import CORSMiddleware
from starlette.middleware.sessions import SessionMiddleware
from api import certification_api, user_api

app = FastAPI(
    title="FastAPI",
    description="SampleAPI",
    middleware=[Middleware(SessionMiddleware, secret_key=Env.SECRET_KEY)],
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

prefix = "/api"
app.include_router(certification_api.router, prefix=prefix)
app.include_router(user_api.router, prefix=prefix)


@app.middleware("http")
async def db_session_middleware(request: Request, call_next):
    try:
        request.state.db = db_session()
        response = await call_next(request)
    finally:
        request.state.db.close()
    return response

if __name__ == "__main__":
    uvicorn.run(app)

16. ターミナルにてコマンドを入力してをFastAPIを起動する

ターミナル.

Python main.py

17. 正常にFastAPIが起動した場合

ターミナル.

(プロジェクト名) (base) ユーザー名@MacBook プロジェクト名 % Python main.py
INFO:     Started server process [4119]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

18. ブラウザでURLを確認する：http://localhost:8000/docs#/

19. ユーザーの作成を実行する

Try it outをクリックする

is_remember_loginのTrue/Falseを選択 / Request bodyの内容を入力をする

Executeをクリックしユーザーの作成を実行し、Responsesを確認する

20.ユーザーが作成されているかデータベースを確認する