FastAPIのシンプルなサンプルコードの紹介

概要

FastAPIのシンプルなサンプルコードを紹介します。

コード一式は、下記にあります。すべてを確認するためには、Download ZIPからZIPをダウンロードしてください。

最初に、テーブルと機能を説明し、続いて、環境構築や実行方法を説明します。
最後に、ファイル構成と、抜粋したコードや補足の説明をします。

テーブルとカラム

著者テーブルと書籍テーブルを操作します。データベースは、SQLiteを使います。

テーブル	カラム
著者（Author）	ID（id）、名前（name）、書籍（books）
書籍（Book）	ID（id）、名前（name）、著者ID（author_id）、著者（author）

• Book.author_idは、Author.idの外部キーです。
• Author.booksとBook.authorは、リレーション用です。

機能

2つの表を操作する11の機能があります。

method	パスとパラメーター	関数	説明
POST	/authors?name=*	add_author()	著者の追加
GET	/authors	get_authors()	全著者の取得
GET	/authors/<author_id>	get_author()	指定著者の取得
PUT	/authors?author_id=*&name=*	update_author()	指定著者の更新
DELETE	/authors?author_id=*	delete_author()	指定著者の削除
POST	/books?name=*	add_book()	書籍の追加
GET	/books	get_books()	全書籍の取得
GET	/books/<book_id>	get_books()	指定書籍の取得
GET	/books/<book_id>/details	book_details()	指定書籍の情報
PUT	/books?book_id=*&name=*	update_book()	指定書籍の更新
DELETE	/books?book_id=*	delete_book()	指定書籍の削除

• 著者と書籍が親子構造になっています
• 書籍を追加するには、親となる著者が必要です
• 指定著者を削除すると、子供である書籍も削除されます

環境構築

Python 3.11で動作します。Poetryが必要です。
以下のようにしてFastAPIの仮想環境を作成します。

poetry install

データベース初期化

以下のようにしてデータベースを初期化します。
ダミーの著者と書籍を追加しています。

poetry run python create_table.py

FastAPIの起動

以下のようにしてFastAPIを起動します。

poetry run uvicorn src.main:app --host 0.0.0.0 --reload

対話的APIドキュメント

下記から対話的APIドキュメント（Swagger UI）が使えます。

• http://localhost:8000/docs

REST APIのファイル構成

APIはsrcディレクトリにあり、下記の5つのファイルからなります。

• main.py：FastAPIのインスタンス（app）を作成しています。
• database.py：SQLAlchemy ORMのクラスとセッションを返す関数（get_db）を定義しています。
• functions.py：データベースを操作する11機能を定義しています。
• schemas.py：APIで扱うpydanticのクラスを定義しています。
• routers.py：パスオペレーション関数を定義しています。

main.py（抜粋）

main.pyは、FastAPIのインスタンス（app）を作成しています。
下記はその抜粋です（一部のimport文は省略しています）。

src/main.py

from .routers import router

app = FastAPI()
app.include_router(router)

routers.pyで定義したパスオペレーション関数を取り込むことで、main.pyをシンプルにしています。

database.py（抜粋）

database.pyは、ORMのクラスを定義しています。SQLAlchemy2.0では、下記のようにDeclarativeBaseやMapped、mapped_columnを使います（参考）。MappedAsDataclassからの派生は省略できますが、派生するとdataclassのように使えて便利です。

src/database.py

class Base(DeclarativeBase):
    pass

class Author(MappedAsDataclass, Base):
    __tablename__ = "author"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    name: Mapped[str] = mapped_column(String(16))
    ...

また、下記のようにAsyncSessionを返すジェネレーターget_dbを定義しています。get_dbは、パスオペレーション関数で使います。

src/database.py

async def get_db() -> AsyncIterator[AsyncSession]:
    async with AsyncSession(engine) as session:
        yield session

functions.py（抜粋）

functions.pyは、データベースを操作する関数を定義しています。
下記は、authorテーブルから主キーでレコードを取得する関数です。

src/functions.py

async def get_author(author_id: int, db: AsyncSession) -> Author | None:
    return await db.get(Author, author_id)

schemas.py（抜粋）

schemas.pyは、パスオペレーション関数で扱う、pydanticのクラスを定義しています。

src/schemas.py

class Author(BaseModel):
    id: int
    name: str
    ...

database.Authorのオブジェクトからschemas.Authorのオブジェクトへの変換については、後述の「ORMクラスからpydanticクラスへの変換の補足」を参照してください。

routers.py（抜粋）

routers.pyでは、パスオペレーション関数を定義しています。Depends(get_db)とすることで、get_dbを差し替えられるようにしています。

src/routers.py

@router.get("/authors/{author_id}", tags=["/authors"])
async def get_author(author_id: int, db: AsyncSession = Depends(get_db)) -> Author:
    author = await functions.get_author(author_id, db)
    ...

pytestの実行

下記のようにして、11の機能をテストします。

poetry run pytest

テストでは、別のengineを使うように、get_dbをget_test_dbで差し替えています。

tests/conftest.py

    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    ...
    async def get_test_db():
        async with AsyncSession(engine) as session:
            yield session

    app.dependency_overrides[get_db] = get_test_db

リレーションのデータの取得について補足

SQLAlchemy ORMのBookクラスは、親のAuthorのリレーション（author）を持っています。

src/database.py

class Book(MappedAsDataclass, Base):
    __tablename__ = "book"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    name: Mapped[str] = mapped_column(String(32))
    author_id: Mapped[int] = mapped_column(ForeignKey("author.id"))
    author: Mapped[Author] = relationship(Author)

Book.authorの情報を取得するには、下記のようにoptions(selectinload(Book.author))を使います。

src/functions.py

async def book_details(book_id: int, db: AsyncSession) -> Book | None:
    return await db.scalar(
        select(Book).where(Book.id == book_id).options(selectinload(Book.author))
    )

ORMクラスからpydanticクラスへの変換の補足

下記は、指定した著者を取得するパスオペレーション関数です。

src/routers.py

@router.get("/authors/{author_id}", tags=["/authors"])
async def get_author(author_id: int, db: AsyncSession = Depends(get_db)) -> Author:
    author = await functions.get_author(author_id, db)
    if author is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, detail="Unknown author_id")
    return Author.model_validate(author)

上記のAuthor.model_validate(author)では、ORMクラス（database.Author）から、下記のpydanticのクラス（schemas.Author）に変換しています。下記のmodel_config = ConfigDict(from_attributes=True)を書くことで、この変換ができるようになります。

src/schemas.py

class Author(BaseModel):
    id: int
    name: str

    model_config = ConfigDict(from_attributes=True)

以上