コンテンツへスキップ

FastAPIを使ったシンプルなヒーローAPI

FastAPIを使ってシンプルなヒーローWeb APIを構築することから始めましょう。✨

FastAPIのインストール

最初のステップはFastAPIをインストールすることです。

FastAPIはWeb APIを作成するためのフレームワークです。

しかし、それを実行するための別の種類のプログラム、「サーバー」も必要です。ここではUvicornを使用します。そして、その標準依存関係と共にUvicornをインストールします。

仮想環境がアクティブになっていることを確認してください。

次に、FastAPIとUvicornをインストールします

$ python -m pip install fastapi "uvicorn[standard]"

---> 100%

SQLModel コード - モデル、エンジン

それでは、SQLModelコードから始めましょう。

ヒーローのみ（まだチームなし）の最もシンプルなバージョンから始めます。

これは、これまでの例で見てきたコードとほぼ同じです。

# One line of FastAPI imports here later 👈
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)

# Code below omitted 👇

from typing import Optional

# One line of FastAPI imports here later 👈
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)

# Code below omitted 👇
👀 ファイル全体のプレビュー 
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/")
def create_hero(hero: Hero):
    with Session(engine) as session:
        session.add(hero)
        session.commit()
        session.refresh(hero)
        return hero


@app.get("/heroes/")
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/")
def create_hero(hero: Hero):
    with Session(engine) as session:
        session.add(hero)
        session.commit()
        session.refresh(hero)
        return hero


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

以前使用していたコードからの変更点は、connect_args内のcheck_same_threadだけです。

これは、SQLAlchemyがデータベースとの通信を担当する低レベルライブラリに渡す設定です。

check_same_threadはデフォルトでTrueに設定されており、単純なケースでの誤用を防ぎます。

しかしここでは、複数のリクエストで同じセッションを共有せず、その設定が存在する問題をすべて防ぐための最も安全な方法であることを確認します。

また、FastAPIでは各リクエストが複数の相互作用するスレッドによって処理される可能性があるため、無効にする必要があります。

情報

今のところはこれで十分な情報です。詳細はFastAPIのasyncとawaitに関するドキュメントを参照してください。

重要なのは、複数のリクエストで同じセッションを共有しないようにすることで、コードが安全になるということです。

FastAPI アプリ

次のステップは、FastAPIアプリを作成することです。

fastapiからFastAPIクラスをインポートします。

そして、そのFastAPIクラスのインスタンスであるappオブジェクトを作成します。

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

# SQLModel code here omitted 👈

app = FastAPI()

# Code below omitted 👇

from typing import Optional

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

# SQLModel code here omitted 👈

app = FastAPI()

# Code below omitted 👇
👀 ファイル全体のプレビュー 
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/")
def create_hero(hero: Hero):
    with Session(engine) as session:
        session.add(hero)
        session.commit()
        session.refresh(hero)
        return hero


@app.get("/heroes/")
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/")
def create_hero(hero: Hero):
    with Session(engine) as session:
        session.add(hero)
        session.commit()
        session.refresh(hero)
        return hero


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

startup時のデータベースとテーブルの作成

アプリが実行を開始したら、データベースとテーブルを作成するために、関数create_tablesが確実に呼び出されるようにする必要があります。

これは、すべてのリクエストの前にではなく、起動時に一度だけ呼び出される必要があるため、"startup"イベントを処理する関数に配置します。

# Code above omitted 👆

app = FastAPI()


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

# Code below omitted 👇

# Code above omitted 👆

app = FastAPI()


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

# Code below omitted 👇
👀 ファイル全体のプレビュー 
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/")
def create_hero(hero: Hero):
    with Session(engine) as session:
        session.add(hero)
        session.commit()
        session.refresh(hero)
        return hero


@app.get("/heroes/")
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/")
def create_hero(hero: Hero):
    with Session(engine) as session:
        session.add(hero)
        session.commit()
        session.refresh(hero)
        return hero


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

ヒーローの作成 パスオペレーション

情報

パスオペレーション（特定のHTTP操作を持つエンドポイント）とそのFastAPIでの使用方法について復習する必要がある場合は、FastAPIファーストステップのドキュメントを確認してください。

新しいヒーローを作成するためのパスオペレーションコードを作成しましょう。

これは、ユーザーが/heroes/パスにPOST操作を含むリクエストを送信したときに呼び出されます。

# Code above omitted 👆

app = FastAPI()


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


@app.post("/heroes/")
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 = FastAPI()


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


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

# Code below omitted 👇
👀 ファイル全体のプレビュー 
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/")
def create_hero(hero: Hero):
    with Session(engine) as session:
        session.add(hero)
        session.commit()
        session.refresh(hero)
        return hero


@app.get("/heroes/")
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/")
def create_hero(hero: Hero):
    with Session(engine) as session:
        session.add(hero)
        session.commit()
        session.refresh(hero)
        return hero


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

情報

これらの概念の一部を復習する必要がある場合は、FastAPIのドキュメントを確認してください。

SQLModel の利点

ここで、SQLModelクラスモデルがSQLAlchemyモデルとPydanticモデルの両方であることが同時に輝く場所です。✨

ここでは、APIによって受信されるリクエストボディを定義するために、同じクラスモデルを使用します。

FastAPIはPydanticに基づいているため、同じモデル（Pydantic部分）を使用して、JSONリクエストからの自動データ検証と変換をHeroクラスの実際のインスタンスであるオブジェクトに実行します。

そして、この同じSQLModelオブジェクトはPydanticモデルインスタンスだけでなくSQLAlchemyモデルインスタンスでもあるため、データベース内の行を作成するためにセッションで直接使用できます。

そのため、直感的な標準的なPython型アノテーションを使用でき、データベースモデルとAPIデータモデルのコードを大量に複製する必要がありません。🎉

ヒント

後でさらに改善しますが、今のところは、SQLModelクラスがSQLAlchemyモデルとPydanticモデルの両方であることの威力を示しています。

ヒーローの読み取り パスオペレーション

次に、すべてのヒーローを読み取る別のパスオペレーションを追加しましょう。

# Code above omitted 👆

app = FastAPI()


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


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


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

# Code above omitted 👆

app = FastAPI()


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


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


@app.get("/heroes/")
def read_heroes():
    with Session(engine) as session:
        heroes = session.exec(select(Hero)).all()
        return heroes
👀 ファイル全体のプレビュー 
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/")
def create_hero(hero: Hero):
    with Session(engine) as session:
        session.add(hero)
        session.commit()
        session.refresh(hero)
        return hero


@app.get("/heroes/")
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/")
def create_hero(hero: Hero):
    with Session(engine) as session:
        session.add(hero)
        session.commit()
        session.refresh(hero)
        return hero


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

これは非常に簡単です。

クライアントがGETHTTP操作を使用してパス/heroes/にリクエストを送信すると、この関数が実行され、データベースからヒーローを取得して返します。

リクエストごとの1つのセッション

操作の各グループごとにSQLModelセッションを使用する必要があり、他の関連のない操作が必要な場合は別のセッションを使用する必要があることを覚えていますか？

ここではさらに明白です。

ほとんどの場合、リクエストごとに1つのセッションを持つのが普通です。

いくつかの孤立したケースでは、内部で新しいセッションが必要になるため、リクエストごとに複数のセッションが必要になります。

しかし、異なるリクエスト間で同じセッションを共有することは決して望ましくありません。

この簡単な例では、パスオペレーション関数で手動で新しいセッションを作成するだけです。

後の例では、FastAPI依存関係を使用してセッションを取得し、他の依存関係と共有し、テスト中に置き換えることができます。🤓

FastAPI アプリケーションの実行

これで、FastAPIアプリケーションを実行する準備ができました。

そのコードをすべてmain.pyというファイルに入れます。

次に、Uvicornで実行します。

$ uvicorn main:app

<span style="color: green;">INFO</span>:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
<span style="color: green;">INFO</span>:     Started reloader process [28720]
<span style="color: green;">INFO</span>:     Started server process [28722]
<span style="color: green;">INFO</span>:     Waiting for application startup.
<span style="color: green;">INFO</span>:     Application startup complete.

情報

コマンドuvicorn main:appは次のことを意味します。

  • main: ファイルmain.py（Python「モジュール」）。
  • app: app = FastAPI()という行でmain.py内に作成されたオブジェクト。

Uvicorn --reload

開発中（開発中のみ）は、Uvicornに--reloadオプションを追加することもできます。

コードに変更を加えるたびにサーバーが再起動されるため、より迅速に開発できます。🤓

$ uvicorn main:app --reload

<span style="color: green;">INFO</span>:     Will watch for changes in these directories: ['/home/user/code/sqlmodel-tutorial']
<span style="color: green;">INFO</span>:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
<span style="color: green;">INFO</span>:     Started reloader process [28720]
<span style="color: green;">INFO</span>:     Started server process [28722]
<span style="color: green;">INFO</span>:     Waiting for application startup.
<span style="color: green;">INFO</span>:     Application startup complete.

本番環境では--reloadを使用しないようにしてください。必要以上に多くのリソースを消費し、エラーが発生しやすくなるなどです。

APIドキュメントUIの確認

ブラウザでそのURLhttp://127.0.0.1:8000にアクセスできます。ルートパス/のパスオペレーションは作成していないため、そのURLだけでは「Not Found」エラーが表示されます…その「Not Found」エラーはFastAPIアプリケーションによって生成されます。

しかし、パス/docsにある自動生成されたインタラクティブなAPIドキュメントにアクセスできます。http://127.0.0.1:8000/docs。✨

この自動APIドキュメントUIには、上記で定義したパスとその操作があり、パスオペレーションが受信するデータの形式を既に認識しています。

Interactive API docs UI

APIの使用

実際には、Try it outボタンをクリックして、ヒーローの作成パスオペレーションでいくつかのヒーローを作成するリクエストを送信できます。

そして、ヒーローの読み取りパスオペレーションでそれらを取り戻すことができます。

Interactive API docs UI reading heroes

データベースの確認

これで、ターミナルに戻ってCtrl+Cを押して、Uvicornサーバーを終了できます。

そして、DB Browser for SQLite を開いてデータベースを確認し、データを探してヒーローが確かに保存されていることを確認できます。🎉

DB Browser for SQLite showing the heroes

まとめ

素晴らしい！これは既に、ヒーローデータベースとやり取りするためのFastAPI ウェブAPI アプリケーションです。🎉

改善および拡張できる点がいくつかあります。例えば、各新しいヒーローのIDをデータベースで決定したいので、ユーザーがIDを送信することを許可したくありません。

これらの改善は次の章で行います。🚀