【FastAPI】SQLAlchemyのCRUD操作【入門】

FastAPISQLAlchemyを用いて、データベースへの

  • create
  • read
  • update
  • delete

操作を実行してみます。

データベースへの接続設定とモデル作成方法は下記記事に記載しています。

この記事のサンプルコード

目次

モデルの作成

簡単なブログを想定した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_article

create_article関数の引数には、データベースセッションと、リクエストを受け付けるようにし

  • title
  • content
  • is_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 = True

response_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_displayTrueになっているものだけを全件取得
  • 特定の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:3is_displayfalseに設定していますので、表示されませんでした。

1件取得

1件取得も同様の流れで作成します。違いとしては、パスパラメーターで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操作が出来るようになりました。もちろん

  • 認証
  • エラーハンドリング

など付帯で処理しないといけない機能はありますが、基本はばっちりですね!

この記事のサンプルコード

この記事が気に入ったら
フォローしてね!

よかったらシェアしてね!

コメント

コメントする

目次
閉じる