関連記事

FastAPI + postgressのアプリ構築メモ part.1 環境構築
FastAPI + postgressのアプリ構築メモ part.2 CREATE (イマココ)
FastAPI + postgressのアプリ構築メモ part.3 READ

今回やること

前回DBの接続からマイグレーションまでやった

CRUD処理のCをいい感じに作っていく

API部分はこんな感じで肥大化しないようにファイル分けて適宜routeを作る

FastAPIでmain.pyが肥大化しそうだからAPIRouterを使ってファイル分割する

CREATE

lib/basemodels.py というファイルを作成してFastAPIのリクエスト・レスポンスの型を全部ここにまとめる（自分はこの方法が好き）

from pydantic import BaseModel

class NewsRegister(BaseModel):
    title: str
    link: str

class NewsRegisterResult(BaseModel):
    message: str

routers/news.py

from fastapi import APIRouter

# my modules
from lib.base_models import *

router = APIRouter()

@router.post("/news/register_news", response_model=NewsRegisterResult, tags=["news"])
async def register_news(form_data: NewsRegister):
    return {"message": f"title: {form_data.title}, link: {form_data.link}"}

main.py

import uvicorn
from fastapi import FastAPI
from starlette.middleware.cors import CORSMiddleware

# routers
from routers import news

app = FastAPI()
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"]
)
app.include_router(news.router)

@app.get("/")
async def root():
    return {"message": "Hello World"}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8889)

http://localhost:8889/docs

にいけばルートができてることが確認できる

http://localhost:8889/news/register_news

に

{
    "title": "ポルシェ 986 ボクスター ダイレクトイグニッションコイル交換",
    "link": "https://monaledge.com/article/499"
}

このbodyでpostすると

{
    "message": "title: ポルシェ 986 ボクスター ダイレクトイグニッションコイル交換, link: https://monaledge.com/article/499"
}

をreturnしてきているのでpostしてAPI側にちゃんとbodyは届いていますね

あとは、これをDBに格納する処理を追加すればOK

database.py

にsessionに関するものを追記

from sqlalchemy.orm import sessionmaker, declarative_base

session = sessionmaker(
    autocommit=False,
    autoflush=False,
    bind=engine
)

lib/news_handler.py

from fastapi import HTTPException
from sqlalchemy.exc import SQLAlchemyError

# my modules
from models import NEWS
from database import Session
from lib.base_models import *


def add_news_to_db(form_data):
    # check format
    if not form_data.title or not form_data.link:
        raise HTTPException(status_code=400, detail="Invalid request")

    # check news already exists
    try:
        ses = Session()
        title_exists = ses.query(NEWS).filter_by(title=form_data.title).scalar()
        link_exists = ses.query(NEWS).filter_by(link=form_data.link).scalar()
        if title_exists or link_exists:
            raise HTTPException(status_code=400, detail="news already exists")
    except SQLAlchemyError:
        raise HTTPException(status_code=500, detail="db check failed")
    finally:
        ses.close()

    # insert into db
    try:
        ses = Session()
        news = NEWS(
            title = form_data.title,
            link  = form_data.link
        )
        ses.add(news)
        ses.commit()
    except SQLAlchemyError:
        ses.rollback()
        raise HTTPException(status_code=500, detail="db insertaion failed")
    finally:
        ses.close()

    return {"message": "registered successed!"}

news.pyの編集

from fastapi import APIRouter

# my modules
from lib.news_handlers import add_news_to_db
from lib.base_models import *

router = APIRouter()

@router.post("/news/register_news", response_model=NewsRegisterResult, tags=["news"])
async def register_news(form_data: NewsRegister):
    return add_news_to_db(form_data)

これでDBにレコード登録はヨシ！