FastAPI SQLModel 入門

**SQLModel** はPythonコードから SQL databases と会話するためのライブラリです。ここでは **FastAPI** での使われ方を見ていきますが、FastAPIとは独立したものとして設計されています。

SQLModelは**FastAPI** の作者が自ら作成しており、**SQLAlchemy** と **Pydantic** の両方との互換性を保っています。**Pydantic** はデータ検証のためのPythonライブラリです。Pythonのtype hintが使われます。ですからPydanticクラスはAPIの入り口であるパスオペレーション関数で使われることが一般的です。

過去記事「[FastAPI と SQL Databases(SQLAlchemy)](https://qiita.com/sand/items/461b3f68060952fc7415)」でFastAPIで **SQLAlchemy** を使ってSQL Databaseを扱う方法を述べましたが、**SQLAlchemy** と **Pydantic** の2重のモデルを使う必要があり冗長性は否めませんでした。**SQLModel** はこの冗長性を解消してくれます。

本記事はSQLModelの公式ドキュメントをまとめたものです。公式ドキュメントは説明が丁寧すぎて何度も読み返すには長すぎる懸念がありますので、短くまとめてみました
[公式サイト Tutorial - User Guide](https://sqlmodel.tiangolo.com/tutorial/)


まず以下のコマンドで環境の作成、sqlmodelのインストールを行います。

```
python -m venv env
.\env\Scripts\activate        # Windows
python -m pip install --upgrade pip
python -m pip install sqlmodel
python -m pip install fastapi "uvicorn[standard]"
```

FastAPIの起動コマンドです。
```
uvicorn main:app --reload
```

# 1. ソースコード

まず全ソースコード(main.py)を掲載して、続いて部分的な解説を行っていきたいと思います。

```Python:main.py
from typing import List, Optional

from fastapi import Depends, 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

class HeroRead(HeroBase):
    id: int

class HeroUpdate(SQLModel):
    name: Optional[str] = None
    secret_name: Optional[str] = None
    age: Optional[int] = None


sqlite_file_name = "database.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)


app = FastAPI()


def get_session():
    with Session(engine) as session:
        yield session


@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


@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=HeroRead)
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


@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:
        raise HTTPException(status_code=404, detail="Hero not found")
    hero_data = hero.dict(exclude_unset=True)
    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


@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}
```




## 1-1. Multiple Models

以下のようにMultiple Modelsを使い分けるのが、**SQLModel** の最も大きな特徴となっています。それぞれのModelは単に**data model**に過ぎないものであったり、または**data model** と **table model**の両方であったりします。2重にfieldを定義しないように、Baseのクラスを継承する形となっています。

```Python:
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

class HeroRead(HeroBase):
    id: int

class HeroUpdate(SQLModel):
    name: Optional[str] = None
    secret_name: Optional[str] = None
    age: Optional[int] = None
```

以下が各Modelの概要です。APIごとにModelを使い分けることでより厳密なValidationが可能となります。

* **HeroBase** : 共通のfieldを集めたmodel
* **Hero** : Heroの**table model**かつdata model。idは、tabile modelとしては必須だが、data modelとしては最初はNoneなのでOptionalとなっている。
* **HeroCreate** : Create APIで使われるmodel。HeroBaseと同じだが可読性向上のため。
* **HeroRead** : Read API で使われるmodel。tableから読み込まれたmodelなのでidはオプションではない。
* **HeroUpdate** : Update APIのためのmodel。全fieldがデフォルト値Noneを持つ。クライアントから明示的に送信された値がある場合にのみ、Noneでない値をもつ。明示的に送信された値のみtableのfieldが更新される。



## 1-2.  SQLAlchemy Engine

databaseをハンドルするための**SQLAlchemy Engine**は以下のように作られます。SQLModelがSQLAlchemyを継承していることに注意してください。

```Python:
sqlite_file_name = "database.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)
```


**Engine Echo** を Trueに設定することで実行される全SQL文をプリントアウトできるようになります。学習時やデバッグ時に便利なものです。


## 1-3. SQLModel MetaData
table = Trueで定義されたtable modelは、SQLModelの**metadata** 属性に記録されます。つまりこの時点でSQLModel.metadataは既にHeroを知っています。

```Python:
def create_db_and_tables():
    SQLModel.metadata.create_all(engine)
---
@app.on_event("startup")
def on_startup():
    create_db_and_tables()
```

create_all(engine)はengineのdatabaseと、この時点でmetadataに記憶されているtableの初期化を行います。
create_db_and_tables()関数は、FastAPIのスタートアップイベント時に1回呼ばれるようにします。


## 1-4. FastAPI Dependency

**session** はdatabaseを扱うためのハンドラーです。ここではsessionを得るための関数 get_session() は **FastAPI Dependency** として実装されてます。ちなみにyieldを含むget_session()関数は、Pythonの**ジェネレータ関数**です。

```Python:
def get_session():
    with Session(engine) as session:
        yield session
```

**path operation function** の引数の中でこの Dependencyが呼ばれると、yield sessionでsessionが返されます。リクエストの全ての処理が終わると、yield session の後ろの処理が行われます。この場合withブロックを抜けsessionがクローズされることになります。

例えばpath operation functionは以下のようになります。

```Python:
@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
```

ここでちょっと注意すべきなのが、path operation functionのcreate_hero()の引数です。sessionはデフォルト値付きで、その後ろにheroというデフォルト値無しの引数が続きます。通常、Pythonではデフォルト値無の引数が後ろに来る必要があります。但しこの場合、先頭に'*'を置くことで、全引数を**キーワード引数**とすることでこの問題を避けています。



## 1-5. path operation

以下にCreate, Read , Update, Deleteのpath operationを見ていきます。

### 1-5-1. Create

```Python:
@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
```

まずcreate_hero()はDependencyを使ってsessionを取り出します。
次の引数のheroは **HeroCreate** のインスタンスとしてクライアントから送り込まれます。これを**Hero.from_orm()**を使って **table model** である **Hero** のインスタンスdb_heroに変換しています。メソッド.from_orm()は他のオブジェクトから attributes を読み込んで新しいクラス（この場合はHero）のインスタンスを作成するものです。

session.commit()で実際にtableに追加されます。しかしこの時idが自動付加され、まだidが割り付けられていないdb_hero と実際のテーブル行が一致しません。これをsession.refresh(db_hero)で一致させます。db_heroが更新されidを持つようになります。そして最後に db_hero をreturnしていますが、response_model=**HeroRead** に注意してください。HeroReadのidはオプションではなく必須項目です。Heroの場合はidがなくともOKですが、HeroReadの場合はidは必須となりより正確なValidationが行われます。


### 1-5-2. Read (Limit と Offset - Query Parameters)

Query ParametersとしてLimit と Offsetが渡され、response_model=List[HeroRead]が返されます。

```Python:
@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
```

session.exec(select...) は **iterable**オブジェクトを返しますが。**.all()**メソッドを追加することで全要素のリストに変換してくれます。つまりHeroReadインスタンスのリストが返されます。



### 1-5-3. Read


```Python:
@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)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    return hero
```

session.get(Hero, hero_id)で、Heroテーブルから、hero_idの要素をしゅとくします。




### 1-5-4. Update

```Python:
@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:
        raise HTTPException(status_code=404, detail="Hero not found")
    hero_data = hero.dict(exclude_unset=True)
    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
```

まず db_hero = session.get(Hero, hero_id)でtableのdb_heroを取り出します。
heroは**HeroUpdate**のインスタンスであり、hero_data = hero.dict(exclude_unset=True)とすることでクライアントが意図的に送ってきた値だけ dict である hero_data に取り出すことができます。つまりデフォルト値のNoneをもつものは排除されます。この値で、setattr(db_hero, key, value)によりdb_heroを更新します。次にdb_heroの更新をsession.add,commit,refreshでテーブルに反映します。



### 1-5-5. Delete


```Python:
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}
```

session.delete(hero)でテーブルから削除するだけです。


### （おまけ）対話的APIドキュメント
FastAPIの大きな調書ですが、以下のようなドキュメントが自動生成されます。**Swagger UI** で実装されており、RESTful APIの仕様書であり、実際にRequestを発行することができます。開発効率に大きく寄与するものです。SQLiteの**DB Browser**と一緒に使うことでテストが大変楽になります。
http://localhost:8000/docs

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/61915/fddf10c6-a1c5-5f67-f14f-db5522f9c1f1.png)



今回は以上です。