【AI_7日目】FastAPI_1冊目

こんにちは投資ロウトです。

背景

AI開発ができるためにwebAPIについて理解を深めていきます。
※Tailwind CSSとkeycloakは現場でピンチになれば再開します。

CRUD処理

コントローラーはモデルとビューを接続するためにファイルが拡大しがちなので、CRUD処理だけこちらに記載すると良いとのことでした。

api/cruds/stock.py

from sqlalchemy.orm import Session

import api.models.stock as stock_model
import api.schemas.stock as stock_schema

# DBセッションとStockCreateスキーマを引数に取り、Stockモデルのインスタンスを返す関数
def create_stock(db: Session, stock: stock_schema.StockCreate) -> stock_model.Stock:
    stock = stock_model.Stock(**stock.dict())
    db.add(stock)
    db.commit()
    db.refresh(stock)
    return stock

ルーターも変更していく。

api/routers/stock.py

from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session

import api.cruds.stock as stock_crud
from api.db import get_db

from typing import List

import api.schemas.stock as stock_schema

router = APIRouter()

@router.post("/stocks", response_model=stock_schema.StockCreateResponse)
async def create_stock(stock_body: stock_schema.StockCreate, db: Session = Depends(get_db)):
    # CRUD操作を行うcreate_stock関数を呼び出し、DBに新しいstockを作成して結果を返す
    return stock_crud.create_stock(db, stock_body)

「orm_mode = True」の意味は、レスポンススキーマが暗黙的にORMからDBモデルを受け取って変換することを意味するとのこと。

api/schemas/stock.py

from pydantic import BaseModel, Field

class StockBase(BaseModel):
    # 共通するフィールドは親クラスに定義
    stock_name: str | None = Field(None, example="投資ロウト株式会社")

class StockCreate(StockBase):
    pass

class StockCreateResponse(StockCreate):
    id: int

    class Config:
        # orm_modeはDB操作の為に必要とのこと。
        orm_mode = True

class Stock(StockBase):
    id: int

    class Config:
        orm_mode = True

動作確認をする。まずはMySQLのレコードを確認

executeの実施

レコードの確認。きちんと登録されています。(2件登録しました)

read処理

read処理の追加

from sqlalchemy.orm import Session
from sqlalchemy import Select
from sqlalchemy.engine import Result

import api.models.stock as stock_model
import api.schemas.stock as stock_schema

# DBセッションを引数に取り、(id, stock_name)のタプルのリストを返す関数
def get_stocks(db: Session) -> list[tuple[int, str]]:
    result: Result = db.execute(Select(stock_model.Stock.id, stock_model.Stock.stock_name).select_from(stock_model.Stock))  
    return result.all()

ルーターにも処理を追加

from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session

import api.cruds.stock as stock_crud
from api.db import get_db

from typing import List

import api.schemas.stock as stock_schema

router = APIRouter()

@router.get("/stocks", response_model=List[stock_schema.Stock])
async def list_stocks(db: Session = Depends(get_db)):
    return stock_crud.get_stocks(db)

問題なく値も取得できていました。

Update

idから取得する処理と、stock_nameを書き換える処理を記載する。

api/cruds/stock.py

# 引数で渡されたstockIdに該当するレコードがないか確認し、その結果を返す
def get_stock(db: Session, stock_id: int) -> stock_model.Stock | None:
    result: Result = db.execute(Select(stock_model.Stock).filter(stock_model.Stock.id == stock_id))
    return result.scalars().first()

# 引数で渡されたstock_nameに書き換えて更新する。
def update_stock(db: Session, stock_create: stock_schema.StockCreate, original: stock_model.Stock) -> stock_model.Stock:
    original.stock_name = stock_create.stock_name
    db.add(original)
    db.commit()
    db.refresh(original)
    return original

対象のidがなければ404エラーを返し、あれば更新を実施する。

api/routers/stock.py

# PUTリクエストを受け付け、指定されたstock_idに基づいてstockを更新するエンドポイント。レスポンスにはStockCreateResponseスキーマを返す
@router.put("/stocks/{stock_id}", response_model=stock_schema.StockCreateResponse)
async def update_stock(stock_id: int, stock_body: stock_schema.StockCreate, db: Session = Depends(get_db)):
    stock = stock_crud.get_stock(db, stock_id)
    if stock is None:
        raise HTTPException(status_code=404, detail="Stock not found")
    return stock_crud.update_stock(db, stock_body, stock)

テストを実施する。まずはDBの再確認から

idが1から3までしかないので、4を指定すると

404のNot Foundが出る

次に2のレコードで更新をしてみる

getメソッド側でデータを確認してみると

きちんと更新されていました。

Delete

最後に削除処理、同じようにoriginalの取得処理は別関数で定義されているので、そちらを利用

api/cruds/stock.py

# 削除処理
def delete_stock(db: Session, original: stock_model.Stock) -> None:
    db.delete(original)
    db.commit()

ルーター側の処理も先ほどとほぼ同じ処理

@router.delete("/stocks/{stock_id}", response_model=None)
async def delete_stock(stock_id: int, db: Session = Depends(get_db)):
    stock = stock_crud.get_stock(db, stock_id)
    if stock is None:
        raise HTTPException(status_code=404, detail="Stock not found")
    stock_crud.delete_stock(db, stock)
    return None

では削除していきます。

消えましたね。

と一旦以上で学習を区切りたいと思います。焦らずコツコツ自分のペースでできればと思います。ご精読ありがとうございました。一歩ずついきたいと思います。