概要
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つの表を操作する12の機能があります。
| method | パスとパラメーター | 関数 | 説明 |
|---|---|---|---|
| POST | /authors?name=... |
add_author() |
著者の追加 |
| GET | /authors |
get_authors() |
全著者の取得 |
| GET | /authors/<author_id> |
get_author() |
指定著者の取得 |
| GET | /authors/<author_id>/details |
author_details() |
指定著者の詳細 |
| PATCH | /authors?id=... |
update_author() |
指定著者の更新 |
| DELETE | /authors?author_id=... |
delete_author() |
指定著者の削除 |
| POST | /books?book=... |
add_book() |
書籍の追加 |
| GET | /books |
get_books() |
全書籍の取得 |
| GET | /books/<book_id> |
get_book() |
指定書籍の取得 |
| GET | /books/<book_id>/details |
book_details() |
指定書籍の詳細 |
| PATCH | /books?id=... |
update_book() |
指定書籍の更新 |
| DELETE | /books?book_id=... |
delete_book() |
指定書籍の削除 |
- 著者と書籍が親子構造になっている
- 書籍を追加するには、親となる著者が必要
- 指定著者を削除すると、子供である書籍も削除される
環境構築
Python 3.12で動作します。uvが必要です。
以下のようにしてFastAPIの仮想環境を作成します。
uv venv
FastAPIの起動
以下のようにしてFastAPIを起動します。
uv run fastapi dev src/fastapi_book_sample/main.py
※ 本番環境では、devをrunにします。
著者が空の時にダミーの著者と書籍を追加しています。
対話的APIドキュメント
下記から対話的APIドキュメント(Swagger UI)が使えます。
REST APIのファイル構成
APIはsrc/fastapi_book_sampleディレクトリにあり、下記の6つのファイルからなります。
-
__init__.py:パッケージ化するための空のファイル -
main.py:FastAPIのインスタンス(app)を作成 -
database.py:SQLAlchemy ORMのクラスとセッションを返す関数(get_db)を定義 -
functions.py:データベースを操作する12機能を定義 -
schemas.py:APIで扱うpydanticのクラスを定義 -
routers.py:パスオペレーション関数を定義
main.py(抜粋)
main.pyは、FastAPIのインスタンス(app)を作成しています。
下記はその抜粋です(一部のimport文は省略しています)。
from .routers import router
app = FastAPI(lifespan=lifespan)
app.include_router(router)
routers.pyで定義したパスオペレーション関数を取り込むことで、main.pyをシンプルにしています。
database.py(抜粋)
database.pyは、ORMのクラスを定義しています。SQLAlchemy2.0では、下記のようにDeclarativeBaseやMapped、mapped_columnを使います(参考)。MappedAsDataclassからの派生は省略できますが、派生するとdataclassのように使えて便利です。
class Base(sqlalchemy.orm.DeclarativeBase):
pass
class Author(sqlalchemy.orm.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(db: AsyncSession, *, author_id: int) -> Author | None:
return await db.get(Author, author_id)
schemas.py(抜粋)
schemas.pyは、パスオペレーション関数で扱う、pydanticのクラスを定義しています。
class AuthorBase(BaseModel):
name: str
class Author(AuthorBase):
id: int | None = None
今回は、pydanticのクラスを下記のように定義しています。
| クラス名 | 基底クラス | 目的 |
|---|---|---|
| AuthorBase | BaseModel | id以外のデータ |
| Author | AuthorBase | idを含むデータ |
| AuthorAdd | AuthorBase | 追加時の引数用 |
| AuthorGet | AuthorAdd | 取得時の戻り値用 |
| AuthorGetWithBooks | AuthorGet | 詳細時の戻り値用 |
| AuthorUpdate | BaseModel | 更新時の引数用 |
| BookBase | BaseModel | id以外のデータ |
| Book | BookBase | idを含むデータ |
| BookAdd | BookBase | 追加時の引数用 |
| BookGet | BookAdd | 取得時の戻り値用 |
| BookGetWithAuthor | BookGet | 詳細時の戻り値用 |
| BookUpdate | BaseModel | 更新時の引数用 |
このように目的に応じたクラスを作ることで、シンプルな記述で安全に動作するようになっています。
routers.py(抜粋)
routers.pyでは、パスオペレーション関数を定義しています。Depends(get_db)とすることで、get_dbを差し替えられるようにしています。
@router.get("/authors/{author_id}", tags=["/authors"])
async def get_author(author_id: int, db: Annotated[AsyncSession, Depends(get_db)]) -> AuthorGet:
author = await functions.get_author(db, author_id=author_id)
...
return author
戻り値の型にAuthorGetを指定することで、database.Author型のauthorが、AuthorGetに変換されます。
pytestの実行
下記のようにして、12の機能をテストします。
uv run pytest
テストでは、別のengineを使うように、get_dbをget_test_dbで差し替えています。
@pytest.fixture(autouse=True)
def override_get_db(db):
def get_test_db():
yield db
app.dependency_overrides[get_db] = get_test_db
リレーションのデータの取得について補足
SQLAlchemy ORMのBookクラスは、親のAuthorのリレーション(author)を持っています。
class Book(sqlalchemy.orm.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(db: AsyncSession, *, book_id: int) -> Book | None:
return await db.scalar(
select(Book).where(Book.id == book_id).options(selectinload(Book.author)),
)
Qiitaの記事
以上