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

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

### 事前準備
データベース作成済み

https://qiita.com/ume1017da/items/feaad4d0dc35f15ca4ba

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

https://qiita.com/ume1017da/items/ec76466f36fc707e5a56

### 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#/
![localhost_8000_docs.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3470136/2831c884-2fce-8ed0-ba49-89501401bd7a.png)

### 19. ユーザーの作成を実行する
#### Try it outをクリックする
![localhost_8000_docs (1).png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3470136/e0669ff4-17e1-7ea4-3090-b8017e0c07a3.png)

#### is_remember_loginのTrue/Falseを選択 / Request bodyの内容を入力をする
![localhost_8000_docs (2).png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3470136/2a746466-4ff0-b795-09f9-fd5c1a618f20.png)

#### Executeをクリックしユーザーの作成を実行し、Responsesを確認する
![localhost_8000_docs (3).png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3470136/5deb2139-a1be-cef3-82cd-649b47568d7a.png)

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