fastapi-book-sample: A Python repository from SaitoTsutomu

概要

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

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

https://github.com/SaitoTsutomu/fastapi-book-sample

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

テーブルとカラム

著者テーブルと書籍テーブルを操作します。データベースは、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文は省略しています）。

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のように使えて便利です。

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は、パスオペレーション関数で使います。

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

`functions.py`（抜粋）

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

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

`schemas.py`（抜粋）

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

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

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

`routers.py`（抜粋）

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

@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で差し替えています。

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

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))を使います。

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クラスへの変換の補足

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

@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)を書くことで、この変換ができるようになります。

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

    model_config = ConfigDict(from_attributes=True)

参考

https://qiita.com/SaitoTsutomu/items/6fd5cd835a4b904a5a3e

以上

SaitoTsutomu/fastapi-book-sample

概要