FastAPI – MySQLとCRUD

モデルを作成してMySQLをDBとしてCRUDまでやりたい。

ユーザーを作る。最終的なディレクトリ構成は以下。

root
 |- user
 |    |- __init__.py
 |    |- models.py
 |    |- schemas.py
 |    |- routes.py
 |
 |- myfunc
 |    |- __init__.py
 |    |- hash.py
 |
 |- database.py
 |- main.py

必要なパッケージをインストール

pip install sqlalchemy sqlalchemy_utils pymysql alembic passlib[bcrypt]

sqlalchemy: ORM

sqlalchemy_utils: idにUUIDを使うため

pymysql: MySQLドライバ

alembic: sqlalchemyでmigration管理

passlib[bcrypt]: パスワードのハッシュ化

database.py作成

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

host = '127.0.0.1:3306'
db_name = 'fast_api'
user = 'root'
password = 'password'

DATABASE_URL = 'mysql+pymysql://%s:%s@%s/%s?charset=utf8' % (
    user,
    password,
    host,
    db_name
)

engine = create_engine(DATABASE_URL)
sessionLocal = sessionmaker(engine)
Base = declarative_base()


def get_db():
    db = sessionLocal()

    try:
        yield db
    finally:
        db.close()

あらかじめfast_apiデータベースをmysqlで作っておく。

model.py作成

テーブルのスキーマを定義するためのモデルを作成する。

from sqlalchemy import Column, Boolean, String, DateTime
from sqlalchemy_utils import UUIDType
from database import Base
import uuid
from datetime import datetime


class User(Base):
    __tablename__ = 'users'

    id = Column(UUIDType(binary=False), primary_key=True, default=uuid.uuid4)
    email = Column(String(255), unique=True)
    password = Column(String(255))
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.now)
    updated_at = Column(DateTime, default=datetime.now)

schemas.py作成

データをAPIとやり取りするために、Pydanticモデルを使用してデータ構造を定義しておく。

from pydantic import BaseModel
from datetime import datetime
from uuid import UUID


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


class UserShow(UserBase):
    id: UUID
    is_active: bool
    created_at: datetime
    updated_at: datetime

    class Config:
        orm_mode = True

routes.py作成

ルートを作成する。

from fastapi import APIRouter, Depends, status, Response, HTTPException
from sqlalchemy.orm import Session
from database import get_db
from .schemas import UserShow, UserBase, UserCreate
from .models import User
from typing import List
from myfunc.hash import get_password_hash
from datetime import datetime


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


@router.get('/', response_model=List[UserShow])
def all_fetch(db: Session = Depends(get_db)):
    users = db.query(User).all()
    return users


@router.get('/{id}', status_code=status.HTTP_200_OK, response_model=UserShow)
def show(id, response: Response, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == id).first()

    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f'User with hte id={id} is not available.'
        )

    return user


@router.post('/')
def create_user(request: UserCreate, db: Session = Depends(get_db)):
    new_user = User(
        email=request.email,
        password=get_password_hash(request.password),
    )
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return new_user


@router.put('/{id}', status_code=status.HTTP_202_ACCEPTED)
def update(id, request: UserBase, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == id)

    if not user.first():
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f'User with the id={id} is not found'
        )

    param = request.dict()
    param['update_at'] = datetime.now()

    user.update(param)
    db.commit()

    return 'Updated'


@router.delete('/{id}', status_code=status.HTTP_202_ACCEPTED)
def delete(id, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == id)

    if not user.first():
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f'User with the id={id} is not found'
        )

    param = {'is_active': False}
    user.update(param)
    db.commit()

    return 'Deleted'

myfunc/hash.py作成

routes.pyのcreate_user中でパスワードをハッシュ化しているが、そのための関数を作成する。

from passlib.context import CryptContext


pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")


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)

main.py修正

main.pyを修正して、routes.pyにで指定したアドレスにアクセスできるようにする。

from fastapi import FastAPI
from user.routes import router as user_router

app = FastAPI()

app.include_router(user_router)

alembicでマイグレーション

alembic init migrations

alembic.ini内のsqlalchemy.urlを修正する。

sqlalchemy.url = mysql+pymysql://root:password@localhost:3306/fast_api

migrations/env.py内のtarget_metadataを修正する。

from user.models import User

target_metadata = [User.metadata]

複数モデルある場合は、リストに追加する。

マイグレーションスクリプトファイルを作成する。

alembic revision --autogenerate -m 'create user table'

migrations/versions/にマイグレーションスクリプトファイルが作成されるので、中身を修正する。sqlalchemy_utilsを使っているため、import文を追記する。

import sqlalchemy_utils

マイグレーションを実行する。

alembic upgrade head

データベースにusersとalembic_versionテーブルが作成される。

APIを確認

以上でUserモデルが作成され、CRUDができる。

uvicorn main:app --reload

http://127.0.0.1:8000/docsにアクセスして動作確認。