モデルを作成して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
にアクセスして動作確認。
[…] FastAPI – MySQLとCRUD […]