FastAPIレスポンスモデルとSQLModel¶

ここでは、FastAPIのresponse_modelを**SQLModel**と共に使用する方法を示します。

インタラクティブなAPIドキュメント¶

これまで使用してきたコードでは、APIドキュメントはクライアントが送信する必要があるデータを知っていました。

Interactive API docs UI

このインタラクティブなドキュメントUIはSwagger UIによって提供されており、Swagger UIが行うことは、標準のOpenAPIを使用して、すべてのデータスキーマ（データ形状）を含むAPIを定義する大きなJSONコンテンツを読み取り、それをその優れたUIに表示することです。

FastAPIは、Swagger UIが読み取るために、自動的にその**OpenAPIを生成**します。

そして、それは**あなたが書いたコードに基づいて**、Pydanticモデル（この場合は**SQLModel**モデル）と型アノテーションを使用して、APIが処理するデータのスキーマを認識して生成します。

レスポンスデータ¶

しかし、これまでAPIドキュメントUIは、アプリケーションが送り返すレスポンスのスキーマを知りませんでした。

コード200を持つ「正常なレスポンス」の可能性があることがわかりますが、レスポンスデータがどのように見えるかはわかりません。

API docs UI without response data schemas

現時点では、FastAPIに受信したいデータだけを伝えていますが、送り返したいデータはまだ伝えていません。

それを今行いましょう。🤓

`response_model`の使用¶

response_modelを使用して、送り返したいデータのスキーマをFastAPIに伝えることができます。

たとえば、同じHero **SQLModel**クラスを渡すことができます（これもPydanticモデルであるため）。

Python 3.10以上Python 3.9以上Python 3.7以上

# Code above omitted 👆

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

# Code below omitted 👇

# Code above omitted 👆

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

# Code below omitted 👇

# Code above omitted 👆

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

# Code below omitted 👇

👀 ファイルプレビュー全文

Python 3.10以上Python 3.9以上Python 3.7以上

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


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


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()


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


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


@app.get("/heroes/", response_model=list[Hero])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

from typing import Optional

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


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)


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()


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


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


@app.get("/heroes/", response_model=list[Hero])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

from typing import List, Optional

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


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)


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()


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


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


@app.get("/heroes/", response_model=List[Hero])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

`response_model`におけるヒーローのリスト¶

Pydanticフィールドと同様に、他の型アノテーションも使用できます。たとえば、Heroのリストを渡すことができます。

まず、typingからListをインポートし、次にList[Hero]を使用してresponse_modelを宣言します。

Python 3.10以上Python 3.9以上Python 3.7以上

# Code here omitted 👈

@app.get("/heroes/", response_model=list[Hero])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

# Code below omitted 👇

# Code here omitted 👈

@app.get("/heroes/", response_model=list[Hero])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

# Code below omitted 👇

from typing import List, Optional

# Code here omitted 👈

@app.get("/heroes/", response_model=List[Hero])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

# Code below omitted 👇

👀 ファイルプレビュー全文

Python 3.10以上Python 3.9以上Python 3.7以上

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


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


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()


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


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


@app.get("/heroes/", response_model=list[Hero])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

from typing import Optional

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


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)


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()


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


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


@app.get("/heroes/", response_model=list[Hero])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

from typing import List, Optional

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


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)


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()


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


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


@app.get("/heroes/", response_model=List[Hero])
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes

FastAPIとレスポンスモデル¶

FastAPIはこのresponse_modelを使用して、レスポンスのデータ検証とフィルタリングを行います。

そのため、これはアプリケーションとクライアント間の契約として機能します。

詳細については、FastAPIのresponse_modelに関するドキュメントを参照してください。

新しいAPIドキュメントUI¶

これで、ドキュメントUIに戻り、受信するレスポンスのスキーマが表示されるようになりました。

API docs UI without response data schemas

クライアントは、どのようなデータが期待できるかを知ることができます。

自動クライアント¶

response_modelを使用する最も顕著な利点は、APIドキュメントUIに表示されることです。

しかし、FastAPIがこのモデルを使用してレスポンスデータの自動データ検証とフィルタリングを行うなど、他の利点もあります。

さらに、スキーマは標準を使用して定義されているため、多くのツールがこれを活用できます。

たとえば、多くの言語でAPIと通信するために必要なコードを自動的に作成できるクライアントジェネレーターがあります。

情報

標準について興味がある場合は、FastAPIは内部的にJSON Schemaを使用するOpenAPIを生成します。

これらについては、FastAPIドキュメント - 最初のステップを参照してください。

要約¶

response_modelを使用して、送り返したいデータのスキーマをFastAPIに伝え、素晴らしいデータAPIを作成しましょう。😎

FastAPIレスポンスモデルとSQLModel¶

インタラクティブなAPIドキュメント¶

レスポンスデータ¶

response_modelの使用¶

response_modelにおけるヒーローのリスト¶

FastAPIとレスポンスモデル¶

新しいAPIドキュメントUI¶

自動クライアント¶

要約¶

`response_model`の使用¶

`response_model`におけるヒーローのリスト¶