LoginSignup
16
18

More than 1 year has passed since last update.

@sand(sand mypress)

FastAPI SQLModel 入門

Posted at

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

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

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

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

まず以下のコマンドで環境の作成、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)を掲載して、続いて部分的な解説を行っていきたいと思います。

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のクラスを継承する形となっています。

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を継承していることに注意してください。

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を知っています。

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のジェネレータ関数です。

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

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

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

@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

@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]が返されます。

@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

@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

@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

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

今回は以上です。

16
18
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
16
18