【FastAPI】SQLAlchemyのCRUD操作【入門】
FastAPIとSQLAlchemyを用いて、データベースへの
createreadupdatedelete
操作を実行してみます。
FastAPIの基礎についての記事まとめ
モデルの作成
簡単なブログを想定したarticleモデルを作成します。models.pyを作成します。
from db.database import Base
from sqlalchemy import Column
from sqlalchemy.sql.sqltypes import Boolean, Integer, String
class Article(Base):
__tablename__ = 'articles'
id = Column(Integer, primary_key=True, index=True)
title = Column(String)
content = Column(String)
is_display = Column(Boolean, default=False)データベースの作成の下記を忘れずに。
models.Base.metadata.create_all(engine)スキーマの定義
modelが作成出来たら、スキーマの定義を行います。shemas.pyを作成します。
from pydantic import BaseModel
class ArticleBase(BaseModel):
title: str
content: str
is_display: bool新規作成 – Create
dbなどのディレクトリを作成し、db_article.pyを作成し、データベース操作用の関数を作成します。
from sqlalchemy.orm.session import Session
from schemas import ArticleBase
from db.models import Article
def create_article(db: Session, request: ArticleBase):
new_article = Article(
title = request.title,
content = request.content,
is_display = request.is_display
)
db.add(new_article)
db.commit()
db.refresh(new_article)
return new_articlecreate_article関数の引数には、データベースセッションと、リクエストを受け付けるようにし
titlecontentis_display
を受け取れるようにします。
エンドポイントの作成
新規追加用関数が作成出来たらエンドポイントを作成します。
from fastapi import APIRouter, Depends
from sqlalchemy.orm.session import Session
from db import db_article
from db.database import get_db
from schemas import ArticleBase
router = APIRouter(
prefix='/article',
tags=['article']
)
@router.post('/')
def get_article(article: ArticleBase, db: Session = Depends(get_db)):
return db_article.create_article(db, article)docsから確認すると・・・
いい感じですね！
レスポンスモデル
任意の返り値を返す場合には、BaseModelを継承したクラスを作成し、routerの引数にresponse_modelを追加します。
返り値用のクラスを作成し
from pydantic import BaseModel
class ArticleBase(BaseModel):
title: str
content: str
class ArticleCreate(ArticleBase):
is_display: bool
class ArticleDisplay(ArticleBase):
id: int
class Config():
orm_mode = Trueresponse_modelを追記します。
from schemas import ArticleBase, ArticleDisplay
@router.post('/', response_model=ArticleDisplay)
def create_article(article: ArticleCreate, db: Session = Depends(get_db)):
return db_article.create_article(db, article)これでドキュメントからAPIを叩いてみると・・・
返り値がBaseModelで定義した内容に変わりました！便利！
{
"id": 7,
"title": "sample_title",
"content": "sample_content"
}読込み – Read
次は読み込み用のAPIを作成します。
is_displayでTrueになっているものだけを全件取得- 特定の
idの1件を取得
2つを作成します。
全件取得
全件取得は下記の様になります。
def get_article_all(db: Session):
return db.query(Article).filter(Article.is_display == True)db.query(model).all()で全件取得が可能で、今回はfilter() で公開が有効になっているものだけ表示されるようにします。
@router.get('/', response_model=List[ArticleDisplay])
def get_article_all(db: Session = Depends(get_db)):
return db_article.get_article_all(db)エンドポイントを作成しました。response_modelにはリストで返却するためList[ArticleDisplay]としています。確認してみましょう。
取得できていますね！id:3はis_displayをfalseに設定していますので、表示されませんでした。
１件取得
１件取得も同様の流れで作成します。違いとしては、パスパラメーターでidをとるぐらいでしょうか。
def get_article(db: Session, id: int):
return db.query(Article).filter(Article.id == id).first()データ取得用の関数を作成し
@router.get('/{id}', response_model=ArticleDisplay)
def get_article(id: int, db: Session = Depends(get_db)):
return db_article.get_article(db, id)エンドポイントを作成しました。
更新 – Update
def update_article(db: Session, id: int, request: ArticleCreate):
article = db.query(Article).filter(Article.id == id)
article.update({
Article.title: request.title,
Article.content: request.content,
Article.is_display: request.is_display
})
db.commit()
return {'message': 'success'}@router.put()とputメソッドを用います。
@router.put('/{id}')
def update_article(id: int, request: ArticleCreate, db: Session = Depends(get_db)):
return db_article.update_article(db, id, request)削除 – Delete
削除も同様です。
def delete_article(db: Session, id: int):
article = db.query(Article).filter(Article.id == id)
article.delete()
db.commit()
return {'message': 'success'}@router.delete('/{id}')
def delete_article(id: int, db: Session = Depends(get_db)):
return db_article.delete_article(db, id)まとめ
これで一通りの基本的なCRUD操作が出来るようになりました。もちろん
- 認証
- エラーハンドリング
など付帯で処理しないといけない機能はありますが、基本はばっちりですね！
FastAPIの基礎についての記事まとめ
コメント