Pythonライブラリ(SQL)：SQLModel（応用編-FastAPI)

１．概要

　前回記事ではSQLModelの基礎編を紹介しました。本記事ではアプリ化する時のファイル構成やFastAPIとの連動を紹介します。

【公式Docs】　

SQLModel

GitHub - tiangolo/sqlmodel: SQL databases in Python, designed for simplicity, compatibility, and robustness.

２．パッケージのファイル構成

　本章では公式で紹介しているパッケージの作成方法を紹介します。

【SQLModelのパッケージ】
Case１：モデル(DB用テーブルクラス）を一つのモジュールで作成
Case２：1つのモデルを１モジュールごとで管理
　ー＞Relationshipを使用する場合は「TYPE_CHECKING」などの工夫が必要
　ー＞テーブルモデルが増えるとこちらの方が管理しやすいと思う

Code Structure and Multiple Files - SQLModel

　なおパッケージをターミナルから実行する場合は"-m"の引数が必要です。

２－１．CASE1：モデルを１モジュールで管理

　DBテーブルのモデルを１つのモジュールのみで管理するパッケージのファイル構成および注意点は下記の通りです。

　２－１－１．パッケージ構成

【パッケージ構成】
●__init__.py：パッケージとして認識させるためのモジュール(空ファイル)
●app.py：CRUD操作を実行
●database.py：DB接続用のengine作成／テーブル作成
●models.py：DBテーブル(クラス）用 

　２－１－２．パッケージ作成時の注意点

　パッケージ作成時の注意点は下記の通りです。

【作成時の注意点】
●処理の順番には注意が必要：database.pyのSQLModel.metadata.create_all()を実行前にmodelsモジュールのimportが必要

　２－１－３．サンプルコード

[__init__.py] ※空ファイル

[app.py]
from sqlmodel import Session

from .database import create_db_and_tables, engine
from .models import Hero, Team

def create_heroes():
    with Session(engine) as session:
        team_z_force = Team(name="Z-Force", headquarters="Sister Margaret’s Bar")
        hero_deadpond = Hero(name="Deadpond", secret_name="Dive Wilson", team=team_z_force)
        session.add(hero_deadpond)
        session.commit()

        session.refresh(hero_deadpond)

        print("Created hero:", hero_deadpond)
        print("Hero's team:", hero_deadpond.team)

def main():
    create_db_and_tables() #事前にmodelsからDBテーブルクラスのimportが必要
    create_heroes()

if __name__ == "__main__":
    main()

[database.py]
from sqlmodel import SQLModel, create_engine

sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

engine = create_engine(sqlite_url) #DB接続情報

def create_db_and_tables():
    SQLModel.metadata.create_all(engine) #DBファイル・テーブル作成

[models.py]
from typing import List, Optional
from sqlmodel import Field, Relationship, SQLModel

class Team(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    headquarters: str

    heroes: List["Hero"] = Relationship(back_populates="team") #HeroはString


class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)

    team_id: Optional[int] = Field(default=None, foreign_key="team.id") #外部キー
    team: Optional[Team] = Relationship(back_populates="heroes") #Teamはclass

　ターミナルで下記を実行するとDBファイルが作成されます。

[Terminal]
python -m project.app

２－２．CASE２：モデルを複数モジュールで管理

　次にDBテーブルのモデルを複数のモジュールで管理するパッケージのファイル構成および注意点は下記の通りです。

　２－２－１．パッケージ構成

【パッケージ構成】
●__init__.py：パッケージとして認識させるためのモジュール(空ファイル)
●app.py：CRUD操作を実行
●database.py：DB接続用のengine作成／テーブル作成
●hero_model.py：DBテーブルのheroクラス専用モジュール
●team_model.py：DBテーブルのteamクラス専用モジュール

　２－２－２．パッケージ作成時の注意点

　パッケージ作成時の注意点は下記の通りです。

【作成時の注意点】
●Circular Importsによるエラー防止のために工夫が必要
　ー＞モジュールをimport時に「TYPE_CHECKING」を使用
　ー＞Relationshipの属性の型ヒントは List[String]で記載(※classではない)
　ー＞Circular importsが無ければ上記は不要(Relationshipなどで他テーブルを参照しない場合）

【Circular importsについて】
　今回のモジュール(heroとteam)ではそれぞれ属性としてお互いのクラスを参照しているため循環参照のような形となりPythonではエラーが発生する現象です。よってSQLModelのコード内には工夫が必要です。

　２－２－３．サンプルコード

　モジュールで実行すると２回目以降にエラーがでるためテーブルクラスに"__table_args__ = {'extend_existing': True}"を追加しました。

[__init__.py] ※空ファイル

[app.py]
from sqlmodel import Session

from .database import create_db_and_tables, engine
from .models import Hero, Team

def create_heroes():
    with Session(engine) as session:
        team_z_force = Team(name="Z-Force", headquarters="Sister Margaret’s Bar")
        hero_deadpond = Hero(name="Deadpond", secret_name="Dive Wilson", team=team_z_force)
        session.add(hero_deadpond)
        session.commit()

        session.refresh(hero_deadpond)

        print("Created hero:", hero_deadpond)
        print("Hero's team:", hero_deadpond.team)

def main():
    create_db_and_tables() #事前にmodelsからDBテーブルクラスのimportが必要
    create_heroes()

if __name__ == "__main__":
    main()

[database.py]
from sqlmodel import SQLModel, create_engine

sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

engine = create_engine(sqlite_url) #DB接続情報

def create_db_and_tables():
    SQLModel.metadata.create_all(engine) #DBファイル・テーブル作成

[team_model.py]
from typing import TYPE_CHECKING, List, Optional
from sqlmodel import Field, Relationship, SQLModel

if TYPE_CHECKING:
    from .hero_model import Hero #Circular Importsによるエラー防止

class Team(SQLModel, table=True):
    __table_args__ = {'extend_existing': True} #moduleで実行したあとに残るメタデータのエラー防止
    
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    headquarters: str

    heroes: List["Hero"] = Relationship(back_populates="team") #型ヒントのclassは文字列で指定

[hero_model.py]
from typing import TYPE_CHECKING, Optional
from sqlmodel import Field, Relationship, SQLModel

if TYPE_CHECKING:
    from .team_model import Team #Circular Importsによるエラー防止


class Hero(SQLModel, table=True):
    __table_args__ = {'extend_existing': True} #moduleで実行したあとに残るメタデータのエラー防止
    
    id: Optional[int] = Field(default=None, primary_key=True) #主キー制約
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True) #Nullable

    team_id: Optional[int] = Field(default=None, foreign_key="team.id") #外部キー制約
    team: Optional["Team"] = Relationship(back_populates="heroes") #型ヒントのclassは文字列で指定

　出力は前節と同じとなります。

３．SQLModel×FastAPI：基礎知識編

３－１．FastAPIのおさらい

　FastAPIを使用してSQLModelとの連動をしていきます。FastAPIの記事は作成済みですが備忘録として一部下記に抜粋しました。

【FastAPI備忘録】
● "response_model=型"で戻り値の型ヒントを設定 (FastAPIのdocsで確認可)
●作成したclassにつけた型ヒントをPydanticが読み取りFastAPIの入力値や戻り値の型(JSON形式)を明確にしてくれる
　ー＞DB操作ごとに使用する型が異なる（CREATE：primary_keyのid不要、READ：全列必要）ため（継承を使用して）複数クラスを作成
　ー＞SQLModel用のテーブルクラスは"table=True"として、DB操作用のクラスは"table=False"でPydanticモデルとして使用する。
●FastAPIはAPIドキュメントを自動作成ー＞CREATE(POSTメソッド)用とREAD(GETメソッド)用でクラスを分けることで型の確認が明確にできる。
●@app.patchを使用して部分的な更新が可能（UPDATE操作）
●「FastAPIのDependency機能：yeild session」を使用することでをwith構文を使用しなくてもシンプルに記載できる。
●Query(defalut, lte)でdefault値に上限も設定できる（lte=less than equal）。

３－２．テーブルモデルクラスの構成：Baseクラスの継承

　構成は下記の通りとなります。

【テーブルモデルクラスの構成】※サンプルはHeroテーブル
●HeroBase：
　ー＞継承専用のクラスでありSQLやPydanticに直接使用しない
　ー＞外部キー制約もこのBaseクラスに記載：継承されたHeroクラスがSQLテーブルを作成する時に外部キーとして働く(公式Docs)
　※下記記載の通りRelationshipはテーブルモデルのみに記載すること
●Hero（★テーブルモデル）
　ー＞SQLModel(DBに登録される)のテーブルのため”table=True”が必要
　ー＞Relationship(back_populates="テーブル名")はSQLModelの機能であるためBaseではなくテーブルモデル用クラスに記載が必要
●HeroCreate(API Docs)
　ー＞CREATE専用：id(primary key)の値はDB側で作成するため不要
●HeroRead(API Docs)
　ー＞READ専用クラス：idの型ヒント追加でAPIドキュメントが見やすい
　ー＞Relationshipの属性を取得する場合はこのクラスを継承した新しいクラスを作成する（公式Docs）
●HeroReadWithTeam（Relationshipの属性も追加したい場合）
　ー＞（Relationshipの）属性も含めたREAD操作（例：team.heroesを取得)
　ー＞テーブルモデルではないためRelationshipクラスそのものは不要（型ヒントをつけてresponse_modelに戻り値を理解させる）
　ー＞Noneでもエラーが出ないようにdefaultは空(Noneや[])にする
●HeroUpdate
　ー＞UPDATE専用：データの更新は１列のみ実施すると他の列はすべて空データ（NULL値に更新ではなく値を与えない）のためそのままだと型ヒントにかかる。よってすべてOptional[型]に変更する必要がある
　ー＞継承はHeroBaseではなくSQLModelから継承
　ー＞HTTPメソッドはGET/POSTではなくPUT(@app.patch)を使用

※DELETE操作は特別なクラスは不要でありパスパラメータで実行可能

３－３．FastAPIのDependency機能：with構文の省略

　今までsessionの起動/停止を実行するためにwith構文を使用したがFastAPIのDependency機能を使用することでシンプルな記載に変更できます(公式)。

Dependencies - First Steps - FastAPI

　通常とDependency機能のコード比較は下記の通りです。

【比較表】
●fastapiからDependsをimportする
●get_session()関数を用いてwith構文からyield sessionでオブジェクト作成
　ー＞FastAPI側でwith構文を調整してくれる。
●(HTTPメソッド)関数に"*"と"session"を渡す
　ー＞classの前にsession引数を渡すためPythonはデフォルト値が判断できない。よって*を渡して残りの引数が”keyword only”であることを示す
●
●
●

[IN ※通常コード]ｋ
from typing import List, Optional

from fastapi import FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, select

class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)

class Hero(HeroBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)

class HeroCreate(HeroBase):
    pass


engine = create_engine('sqlite:///:memory:', echo=True, connect_args={"check_same_thread": False})

def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroRead)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.from_orm(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero

[IN ※Dependency]
from typing import List, Optional

from fastapi import Depends, FastAPI, HTTPException, Query #Dependency使用=Depends
from sqlmodel import Field, Session, SQLModel, create_engine, select

class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)

class Hero(HeroBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)

class HeroCreate(HeroBase):
    pass

def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


def get_session(): #Dependency機能追加
    with Session(engine) as session:
        yield session


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroRead)
def create_hero(*, session: Session = Depends(get_session), hero: HeroCreate):
    db_hero = Hero.from_orm(hero)
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero

４．SQLModel×FastAPI：実践編-1テーブル

　本章から実際のコードを作成していきます。

４－１．DB構成

　DB構成は下記の通りです。

【本章のDB構成】
●テーブルは1つのみ作成する(heroテーブル：下表参照)。
●テーブルは1つのためRelationship()や外部キーなどは設定しない
●アプリファイルは今回は一つで作成(パッケージは作成しない）

４－２．全コード

　全コードおよび要点は下記の通りです。

【要点１：テーブルモデル】
　●Baseモデルを作成して継承することで重複を防止
　●UPDATEは更新しない属性がエラーにならないように(Baseではなく)SQLModelを継承して、かつすべてOptionalにする
【要点2：FastAPI】
　●FastAPI起動時にDB接続できるよう@app.on_event("startup")に設定
　●入力値の型ヒントや戻り値の型ヒント(response_model)を使用することでPydanticのチェック機能を充実させる（SQLModelの動作にも影響）。
　●データが空の時に注意機構をつけるためHTTPExceptionを追加
【要点3：POSTメソッド】
　●POSTしたデータからオブジェクトデータを作成できるよう.from_orm()メソッドを使用
【要点4：GETメソッド】
　●大量のデータを取得しないようにQueryを使用してlimit設定
　●List形式だけでなく1つのデータも取得できるようにルーティング
【要点5：UPDATEメソッド】
　●hero.dict(exclude_unset=True)とすることで初期値のNone(HeroUpdate)が残っている属性は除外してくれる
【要点6：DELETEメソッド】
　●現状だと比較的簡単にデータが削除できるため実際に機能を追加する場合は権限の設定などが必要。

[main1.py]
from typing import List, Optional

from fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, select

#DBテーブルモデルの定義
#継承用ベースモデル
class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)

#テーブルモデル用クラス
class Hero(HeroBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)

class HeroCreate(HeroBase): #idはDB側で自動生成するため不要(HeroBaseをそのまま継承)
    pass

class HeroRead(HeroBase):
    id: int #idに型ヒントをつけることでAPIドキュメントが明確になる

class HeroUpdate(SQLModel): #継承はHeroBaseではなくSQLModel
    name: Optional[str] = None #1つの属性作成時に他属性がNoneとなるためOptionalをつける
    secret_name: Optional[str] = None
    age: Optional[int] = None

#engine作成
sqlite_file_name = "220504_noteSQLModel.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)

#with構文省略のためのFastAPI Dependency機能
def get_session():
    with Session(engine) as session:
        yield session


#FastAPIのアプリ作成
app = FastAPI()

#FastAPI起動時の動作設定
@app.on_event("startup")
def on_startup():
    create_db_and_tables()

#POSTメソッド
@app.post("/heroes/", response_model=HeroRead)
def create_hero(*, session: Session = Depends(get_session), hero: HeroCreate):
    db_hero = Hero.from_orm(hero) #JSONデータ(hero)からオブジェクト作成するため.from_orm()メソッドを使用
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero) #DBからオブジェクトデータを再取得
    return db_hero

#GETメソッド
@app.get("/heroes/", response_model=List[HeroRead])
def read_heroes(*,session: Session = Depends(get_session),
                offset: int = 0, limit: int = Query(default=100, lte=100),): #limitは初期値=100かつ、制限：100以下
    heroes = session.exec(select(Hero).offset(offset).limit(limit)).all() #出力はList(response_modelに記載の通り)
    return heroes


@app.get("/heroes/{hero_id}", response_model=HeroRead)
def read_hero(*, session: Session = Depends(get_session), hero_id: int):
    hero = session.get(Hero, hero_id) #session.get()でprimary_keyを直接指定して取得
    if not hero: #heroが空の場合
        raise HTTPException(status_code=404, detail="Hero not found")
    return hero

#UPDATE(PUT)メソッド
@app.patch("/heroes/{hero_id}", response_model=HeroRead)
def update_hero(*, session: Session = Depends(get_session), hero_id: int, hero: HeroUpdate):
    #更新したいデータを取得
    db_hero = session.get(Hero, hero_id)
    if not db_hero: #db_heroが空の場合
        raise HTTPException(status_code=404, detail="Hero not found")
    
    hero_data = hero.dict(exclude_unset=True) #Pydanticの機能で{"name": None,"secret_name": None,"age": None,} から値があるkeyだけ残してNoneは削除
    
    for key, value in hero_data.items():
        setattr(db_hero, key, value) #db_heroのkey属性にvalueをセット(db_hero.key = valueと同義->keyは既に存在するため更新になる)
        
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero

#DELETEメソッド
@app.delete("/heroes/{hero_id}")
def delete_hero(*, session: Session = Depends(get_session), hero_id: int):

    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    session.delete(hero)
    session.commit()
    return {"ok": True}

　FastAPIの起動(+APIドキュメント移動）は下記で実行しました。

[Terminal ※開発中のため--reload記載]
uvicorn main1:app --reload

４－３．HTTPクライアントでAPIを叩く：requests

　FastAPIを起動したらHTTPクライアントでCRUD操作を実施してみます。

[IN　※新しくJUpyter Notebookなどで処理]
import requests 
import json

root_api = "http://127.0.0.1:8000"  #開発中のためlocalhostのIPアドレス指定

#登録用データ※for文で回したいためListにまとめて記載
heroes =[{'name':"Deadpond",
          'secret_name':"Dive Wilson"}, #id=1, age=None
         {'name':"Spider-Boy",
          'secret_name':"Pedro Parqueador"}, #id=2, age=None
        {'name':"Rusty-Man", 
         'secret_name':"Tommy Sharp",
         'age':48}] #id=3

#CREATE操作
for hero in heroes:
    res = requests.post(root_api + "/heroes/", data=json.dumps(hero))
    if res.status_code==200:
        print('データ登録：', res.json())

#READ操作
res = requests.get(root_api + "/heroes/")
if res.status_code == 200:
    print(res.json())


[OUT] ※CREATE側のprintは省略
[{'name': 'Deadpond', 'secret_name': 'Dive Wilson', 'age': None, 'id': 1},
 {'name': 'Spider-Boy', 'secret_name': 'Pedro Parqueador', 'age': None, 'id': 2},
 {'name': 'Rusty-Man', 'secret_name': 'Tommy Sharp', 'age': 48, 'id': 3}]

　後述のAPIドキュメント(GET)で確認すると下記の通りとなります。

４－４．APIドキュメント確認

　FastAPI起動後にブラウザで「http://127.0.0.1:8000/docs」に接続するとAPIドキュメントを確認することができます。

　４－３－１．API-Docs：HTTPメソッド

　APIドキュメントのdefaultより下記が確認できます。

【確認事項】
●作成した全てのHTTPメソッド(DB側ではCRUD操作)のAPIドキュメントが作成されておりルーティングや関数名も確認できる
●ブロックを開くと各メソッドでの渡す値(パスパラメータやPOST用データ)の詳細を確認できる(APIドキュメントから操作も可能)

　４－３－２．API-Docs：Schemas

　Schemas(データモデル）より下記が確認できます。

【確認事項】
●BaseやSQLテーブルモデルはSchemasに出ない(APIの関数で未使用のため)
●Optional[]を設定している属性には"*"がついており必須パラメータであることが分かる（公式Docs）
●CREATEのパラメータにidは無くUPDATE時は全パラメータがOptionalである

５．SQLModel×FastAPI：実践編-複数テーブル

５－１．DB構成

【本章のDB構成】
●テーブルは２つ作成(hero, teamテーブル：下表参照)。
●Relationship()を設定して、外部キーのidでテーブルを紐づける
●アプリファイルは今回は一つで作成(パッケージは作成しない）

５－２．全コード

　全コード、前章コードからの変更点および要点は下記の通りです。

【前コード(heroテーブルのみ)から変更点】
●Teamモデル(teamテーブル)を作成してSQLModelのテーブルにheroes属性(Relationshipクラス)を設定する
●Heroモデルには①team_idを追加して外部キーを設定、②SQLModelのテーブル(Heroクラス)にteam属性(Relationshipクラス)を追加する

【要点１：テーブルモデル】
　●基本的な要点は前章と同じ
　●Relationship()の他テーブル属性の読み取りは別クラス(型ヒント)を作成

【要点２：属性の型ヒント】
　●属性は指定が無いと無限ループになる(hero->team->heroes->team・・
)。これを防止するため属性の戻り値に型ヒントをつけて値を固定
　class TeamReadWithHeroes(TeamRead): heroes: List[HeroRead] = []

[main2.py]
from typing import List, Optional

from fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Relationship, Session, SQLModel, create_engine, select

#DBテーブルモデルの定義
#継承用ベースモデル
class TeamBase(SQLModel):
    name: str = Field(index=True)
    headquarters: str

class Team(TeamBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)

    heroes: List["Hero"] = Relationship(back_populates="team") #先に記載するためList[String]

class TeamCreate(TeamBase):
    pass

class TeamRead(TeamBase):
    id: int

class TeamUpdate(SQLModel):
    id: Optional[int] = None
    name: Optional[str] = None
    headquarters: Optional[str] = None


#継承用ベースモデル
class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)

    team_id: Optional[int] = Field(default=None, foreign_key="team.id") #外部キー


class Hero(HeroBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)

    team: Optional[Team] = Relationship(back_populates="heroes") #Teamの後にくるためOptional[class]


class HeroCreate(HeroBase): #idはDB側で自動生成するため不要(HeroBaseをそのまま継承)
    pass

class HeroRead(HeroBase):
    id: int #idに型ヒントをつけることでAPIドキュメントが明確になる

class HeroUpdate(SQLModel): #継承はHeroBaseではなくSQLModel
    name: Optional[str] = None #1つの属性作成時に他属性がNoneとなるためOptionalをつける
    secret_name: Optional[str] = None
    age: Optional[int] = None
    team_id: Optional[int] = None


#属性(Relationship())込みのAPI
class HeroReadWithTeam(HeroRead):
    team: Optional[TeamRead] = None #属性が無限ループしないため型ヒント設置

class TeamReadWithHeroes(TeamRead):
    heroes: List[HeroRead] = [] #属性が無限ループしないため型ヒント設置
    

#engine作成
sqlite_file_name = "220504_noteSQLModel2Tbl.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)

#with構文省略のためのFastAPI Dependency機能
def get_session():
    with Session(engine) as session:
        yield session

#FastAPIのアプリ作成
app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


#HeroテーブルのCRUD操作
#POSTメソッド
@app.post("/heroes/", response_model=HeroRead)
def create_hero(*, session: Session = Depends(get_session), hero: HeroCreate):
    db_hero = Hero.from_orm(hero)
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero

#GETメソッド
@app.get("/heroes/", response_model=List[HeroRead])
def read_heroes(*, session: Session = Depends(get_session),
                offset: int = 0, limit: int = Query(default=100, lte=100),):
    heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
    return heroes

@app.get("/heroes/{hero_id}", response_model=HeroReadWithTeam)
def read_hero(*, session: Session = Depends(get_session), hero_id: int):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    return hero

#UPDATE(PUT)メソッド
@app.patch("/heroes/{hero_id}", response_model=HeroRead)
def update_hero(*, session: Session = Depends(get_session), hero_id: int, hero: HeroUpdate):
    
    db_hero = session.get(Hero, hero_id) #hero_idを元にDBから取得
    if not db_hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    
    hero_data = hero.dict(exclude_unset=True) #不要なNoneデータを削除
    for key, value in hero_data.items():
        setattr(db_hero, key, value) #オブジェクト内のデータ置換
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero

#DELETEメソッド
@app.delete("/heroes/{hero_id}")
def delete_hero(*, session: Session = Depends(get_session), hero_id: int):

    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    session.delete(hero)
    session.commit()
    return {"ok": True}


#TeamテーブルのCRUD操作
@app.post("/teams/", response_model=TeamRead)
def create_team(*, session: Session = Depends(get_session), team: TeamCreate):
    db_team = Team.from_orm(team)
    session.add(db_team)
    session.commit()
    session.refresh(db_team)
    return db_team

#GETメソッド
@app.get("/teams/", response_model=List[TeamRead])
def read_teams(*, session: Session = Depends(get_session),
                offset: int = 0, limit: int = Query(default=100, lte=100),):
    teams = session.exec(select(Team).offset(offset).limit(limit)).all()
    return teams

@app.get("/teams/{team_id}", response_model=TeamReadWithHeroes)
def read_team(*, team_id: int, session: Session = Depends(get_session)):
    team = session.get(Team, team_id)
    if not team:
        raise HTTPException(status_code=404, detail="Team not found")
    return team

#UPDATE(PUT)メソッド
@app.patch("/teams/{team_id}", response_model=TeamRead)
def update_team(*, session: Session = Depends(get_session),
                team_id: int, team: TeamUpdate,):
    db_team = session.get(Team, team_id)
    if not db_team:
        raise HTTPException(status_code=404, detail="Team not found")
    team_data = team.dict(exclude_unset=True)
    for key, value in team_data.items():
        setattr(db_team, key, value)
    session.add(db_team)
    session.commit()
    session.refresh(db_team)
    return db_team

#DELETEメソッド
@app.delete("/teams/{team_id}")
def delete_team(*, session: Session = Depends(get_session), team_id: int):
    team = session.get(Team, team_id)
    if not team:
        raise HTTPException(status_code=404, detail="Team not found")
    session.delete(team)
    session.commit()
    return {"ok": True}

５－３．HTTPクライアントでAPIを叩く：requests

　FastAPIを起動したらHTTPクライアントでCRUD操作を実施してみます。

[IN]
import requests 
import json

root_api = "http://127.0.0.1:8000"  #開発中のためlocalhostのIPアドレス指定

#登録用データ※for文で回したいためListにまとめて記載
heroes =[{'name':"Deadpond",
          'secret_name':"Dive Wilson"}, #id=1, age=None
         {'name':"Spider-Boy",
          'secret_name':"Pedro Parqueador"}, #id=2, age=None
        {'name':"Rusty-Man", 
         'secret_name':"Tommy Sharp",   
         'age':48}] #id=3

teams= [{'name':"Preventers", 
        'headquarters':"Sharp Tower"}, #id=1
        {'name':"Z-Force", 
        'headquarters':"Sister Margaret’s Bar"}] #id=2

#CREATE操作
for hero in heroes:
    res = requests.post(root_api + "/heroes/", data=json.dumps(hero))
    if res.status_code==200:
        print('データ登録(Hero)：', res.json())

for teams in teams:
    res = requests.post(root_api + "/teams/", data=json.dumps(teams))
    if res.status_code==200:
        print('データ登録(Team)：', res.json())

#READ操作
res_hero = requests.get(root_api + "/heroes/")
res_team = requests.get(root_api + "/teams/")
if res_hero.status_code == 200:
    print(res_hero.json())
if res_team.status_code == 200:
    print(res_team.json())

[OUT]
データ登録(Hero)： {'name': 'Deadpond', 'secret_name': 'Dive Wilson', 'age': None, 'team_id': None, 'id': 1}
データ登録(Hero)： {'name': 'Spider-Boy', 'secret_name': 'Pedro Parqueador', 'age': None, 'team_id': None, 'id': 2}
データ登録(Hero)： {'name': 'Rusty-Man', 'secret_name': 'Tommy Sharp', 'age': 48, 'team_id': None, 'id': 3}
データ登録(Team)： {'name': 'Preventers', 'headquarters': 'Sharp Tower', 'id': 1}
データ登録(Team)： {'name': 'Z-Force', 'headquarters': 'Sister Margaret’s Bar', 'id': 2}

[{'name': 'Deadpond', 'secret_name': 'Dive Wilson', 'age': None, 'team_id': None, 'id': 1}, {'name': 'Spider-Boy', 'secret_name': 'Pedro Parqueador', 'age': None, 'team_id': None, 'id': 2}, {'name': 'Rusty-Man', 'secret_name': 'Tommy Sharp', 'age': 48, 'team_id': None, 'id': 3}]
[{'name': 'Preventers', 'headquarters': 'Sharp Tower', 'id': 1}, {'name': 'Z-Force', 'headquarters': 'Sister Margaret’s Bar', 'id': 2}]

５－４．APIドキュメント確認

　５－４－１．API-Docs：HTTPメソッド

　APIドキュメントのdefaultより下記が確認できます。

【確認事項】
●作成した全てのHTTPメソッド(DB側ではCRUD操作)のAPIドキュメントが作成されておりルーティングや関数名も確認できる
●ブロックを開くと各メソッドでの渡す値(パスパラメータやPOST用データ)の詳細を確認できる(APIドキュメントから操作も可能)

　５－４－２．Schemas

Schemas(データモデル）より下記が確認できます。

【確認事項】
●BaseやSQLテーブルモデルはSchemasに出ない(APIの関数で未使用のため)
●Optional[]を設定している属性には"*"がついており必須パラメータであることが分かる（公式Docs）
●CREATEのパラメータにidは無くUPDATE時は全パラメータがOptionalである
●外部キーや他テーブル属性も（参照値のため）必須パラメータではない

---

参考記事

FastAPI with Async SQLAlchemy, SQLModel, and Alembic

あとがき

　SQLModelを実際に使っていきたいけどどのタイミングで取り入れるかは要検討かな。