FastAPI の現場から
- Event:
DjangoCongress JP 2025
- Presented:
2025/02/22🐈 nikkie
お前、誰よ
nikkie（にっきー） ※本発表は個人の見解です
機械学習 エンジニア
プロダクトとして価値を届けるために Web APIの開発も します（今回FastAPIの知見を共有）
お前、誰よ
ブログ 連続 820 日突破
@ftnext このスライドで使ってる ftnext/sphinx-new-tab-link など
2023年に悪いことしました Djangoアプリに作り込んで学ぶ脆弱性
FastAPI、ご存知ですか？
聞いたことがある🙋♂️
使ったことがある🙋♀️
トーク14本中FastAPIが登場しそうなのは他に1本
PythonコミュニティにおけるFastAPI
Webフレームワーク 第3位
データサイエンス で使われる
FastAPI
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def root():
return {"message": "Hello World"}
Djangoの urls.py のところの話
path / に GET operation が来たら
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def root():
return {"message": "Hello World"}
root 関数を実行してレスポンスを返す
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def root():
return {"message": "Hello World"}
この延長に機械学習モデルをサーブするAPI
他方 Django REST Framework
bash: curl -u admin -H 'Accept: application/json; indent=4' http://127.0.0.1:8000/users/
Enter host password for user 'admin':
{
"count": 1,
"next": null,
"previous": null,
"results": [
{
"url": "http://127.0.0.1:8000/users/1/",
"username": "admin",
"email": "admin@example.com",
"groups": []
}
]
}記事 Django vs. FastAPI, An Honest Comparison
Batteries includedか、 自分で組み合わせる 必要があるか
非同期対応の度合い（部分的か、fullyか）
IMO：それぞれ得意分野が異なる
FastAPIの現場から
社内向けの 小さなWeb API をチームで開発（認証機能はなし）
FastAPIのチュートリアルを皆で参照しながら
見聞きしていたPyConのトークも手がかりに（👉今回の知見共有。熟知はしてないです）
手がかり（個々に取り上げます）🏃♂️
外部のLLMの API を使うアプリケーション
class GPTView(View):
async def post(request):
res = await openai.ChatCompletion.acreate({...})非同期IO が有効。皆やってみたさがあり、FastAPIを選択
小さい単位で 都度設計 しながら進めています
最初に全機能設計したわけではありません
path 1つ、operation 1つ に絞って（既存を拡張するよう）設計し、実装
これを繰り返す。 その時点の最適解 を更新していく
XP（eXtreme Programming）🏃♂️
アジャイル開発の1手法
小さい価値でも届け、そこからの学びを活かす サイクル を何度も何度も回す（今回のAPIは3ヶ月経過）
対象の ドメイン や使っている技術の理解が少しずつ増えていく
開発の流れ
開発単位：ユーザストーリー
完了条件となる受け入れテストを書く（ATDD）
既存実装を拡張する設計を考え、テスト駆動開発（&ペアプログラミング）で実装
同僚による🏃♂️
デプロイ先は Kubernetes
マイクロサービスなAPI群
今回のFastAPIアプリもコンテナ化
GKEにデプロイ
私が暗黙の前提にしてるかも
サンプルアプリケーション
お品書き：FastAPIの現場から
非同期IO
クリーンなアーキテクチャを志向する
Twelve-Factor App
非同期IO
FastAPI
SQLModel (SQLAlchemy)
FastAPI
Starlette + Pydantic + 作者tiangolo氏の工夫
encode/starlette
lightweight ASGI framework（非同期IOサポート）
高いパフォーマンス （Web Framework Benchmarks 0.18.0時点。FastAPIにも恩恵）
Pydantic
型ヒントに沿って パース するライブラリ（『ロバストPython』）
エディタでの補完や、実行時バリデーション
FastAPIの async def
@app.get("/sync") # path operation decorator
def path_operation_function_def():
return {"message": "Hello World"}
@app.get("/async")
async def path_operation_function_async_def():
return {"message": "どんなときにasync defにする？"}雑な回答
📣「FastAPIなら async def でしょ！」
処理は2種類（『Python実践入門』）
IOバウンド
外部との通信（APIやDB）
ファイルの読み書き
CPUバウンド
CPUで計算
そもそも非同期IOが有効なのは
IOバウンド な処理（CPUバウンドではない）
CPUをIO待ちにしないで他の処理を進める
FastAPIのドキュメント曰く
非同期IOをサポートするライブラリを使うなら、パスオペレーション関数は
async defPythonの文法として
awaitが使えるのはasync defの中だけ
client = AsyncOpenAI()
@app.get("/async")
async def use_async_support_library():
chat_completion = await client.chat.completions.create(...)FastAPIのドキュメント曰く
非同期IOをサポートしないライブラリなら、パスオペレーション関数は
def＝ブロッキングIOでは
def
@app.get("/sync")
def use_blocking_io_library():
response = requests.get(...)質問：CPUバウンドな処理は、どっち？
FastAPIでは async def ‼️
it's better to use
async defunless your path operation functions use code that performs blocking I/O.
「パスオペレーション関数がブロッキングIOするコードを使わない場合、 async def がよい」
https://fastapi.tiangolo.com/async/#path-operation-functions
FastAPIでの非同期IO🥟
非同期IOをサポートしたライブラリを使って、パスオペレーション関数を
async defで書こう！CPUバウンドな処理もFastAPIでは
async defで書くとよい
SQLModel
SQLAlchemy + Pydantic + 作者tiangolo氏の工夫
速習SQLModel🏃♂️ (PyCon JP 2024)
チュートリアル の内容を30分でカバーしてます
ここでもPydantic
SQLModelを継承したクラスは、PydanticのBaseModelでもあるので、 FastAPIで使えるORMを扱う際にもエディタで補完が効く
SQLAlchemy
代表的なORM
DBの非同期IO をサポート！（ありがとう！！）
DBの非同期IO
SQLAlchemy にSQLModelは乗っかる（SQLAlchemyの話になります）
PostgreSQLの例
SQLModelとして
from sqlalchemy.ext.asyncio import create_async_engine
from sqlmodel.ext.asyncio.session import AsyncSession
database_url = "postgresql+asyncpg://developer:mysecretpassword@127.0.0.1:5432/practice"
engine = create_async_engine(database_url)
async with AsyncSession(engine) as session:
...IMO： async_sessionmaker といいとこ取り
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlmodel.ext.asyncio.session import AsyncSession as SQLModelAsyncSession
database_url = "postgresql+asyncpg://developer:mysecretpassword@127.0.0.1:5432/practice"
engine = create_async_engine(database_url)
AsyncSession = async_sessionmaker(engine, class_=SQLModelAsyncSession)
async with AsyncSession() as session: # AsyncSession.begin() もできる
...SQLModelが提供するDBセッション
from sqlmodel import Session
from sqlmodel.ext.asyncio.session import AsyncSessionDBとのやり取り はセッションを通して
SQLAlchemyの
SessionやAsyncSessionを継承したクラス
DBセッション比較
ライブラリ |
|
|
SQLAlchemy |
ある |
ない |
SQLModel |
ある |
ある （推奨） |
SQLModelのDBセッションは 型が当たる
Sessionインスタンスにstatementを渡して、exec()メソッドを呼ぶ
statement = select(Hero)
async with AsyncSession(async_engine) as session:
results = await session.exec(statement)
for hero in results:
print(repr(hero))SQLAlchemyのSessionの
execute().scalars()メソッドを呼んで実現🏃♂️
sessionmaker の提供
- SQLAlchemy:
ある。セッションの ファクトリ クラスを返すため、常に
engineを渡して初期化しなくてよくなる- SQLModel:
ない。tiangolo氏 「常に
with Session(engine)すればよい」（IMO：せやろか？）
提案：sessionmakerでSQLModelのセッションを返させる
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlmodel.ext.asyncio.session import AsyncSession as SQLModelAsyncSession
AsyncSession = async_sessionmaker(engine, class_=SQLModelAsyncSession)SQLModelのセッションのファクトリなので
exec()が使える！現状、世にはSQLAlchemyの例が多いので、DB接続部分は参考にできる
提案の参考🏃♂️
IMO：SQLModel所感
SQLModelでFastAPIアプリは
async defで書けるが、まだまだ発展途上（0.0.22）SQLAlchemyの経験 がある方にだけ、オススメします
経験ない我が身には、SQLModel + SQLAlchemyと 学びが2倍 だった🙌
IMO：SQLModelの現状🏃♂️
チュートリアルはtoy exampleで実務レベルの例がない（プルリクチャンス！）
先の比較記事 より、シングルメンテナゆえに判断を誤ったのでは
FastAPIユーザが 分断 されてしまっている（SQLModel採用のアプリケーション例が見つかりにくい）
お品書き：FastAPIの現場から
非同期IO
クリーンなアーキテクチャを志向する
Twelve-Factor App
The Clean Architecture
Robert C. Martin (Uncle Bob) によるブログ
https://blog.cleancoder.com/uncle-bob/2012/08/13/the-clean-architecture.html
氏による書籍『Clean Architecture』と区別してます
XXXアーキテクチャに共通するもの
Hexagonal (=Ports and Adapters) / Onion / e.t.c.
レイヤ分け による 関心の分離
「共通するもの」の概要（schematic）
注：この4層を導入すれば、クリーンアーキテクチャ ではありません
The Dependency Rule
source code dependencies can only point inwards.
各レイヤは、自分より 内側のレイヤだけに依存 する
なぜ提唱されたのか（私の理解）
例えば、DjangoのModel-View-Template
アプリケーションの中心にあるのは、フレームワーク（Django・FastAPI）や具体のDBになってませんか？
ユーザ価値を提供する ビジネスルール をアプリケーションの中心に置きたい！
The Clean Architecture は
中心にはビジネスルール
フレームワークやデータベースから 独立 （差し替え可能）
詳細の決定を遅らせる
Clean Architectureを学んできた道🏃♂️
レイヤ分け
src/api
├── domain/
├── driver/
├── gateway/
├── port/
├── rest/
└── use_case/このレイヤ分けで実現したいこと
use_case を注入@app.get("/books", response_model=list[BookReadModel])
async def get_books(
use_case: Annotated[ListBooksUseCase, Depends(inject_list_books_use_case)],
):
books = await use_case.execute()
return [BookReadModel.from_book(book) for book in books]同心円の内側 -> 外側の向きに見ていきます
domain
domain
同心円の一番内側（図ではEntities）
システムがなくても存在 するビジネスルール（例：銀行）
事業領域をコードで表現する（ドメインモデリング 。ここに時間を使えています）
ドメインモデリング 参考資料🏃♂️
Pydanticの BaseModel を使って実装
frozen=True変更できなくする
class Book(BaseModel, frozen=True):
id: str
isbn: ISBN
title: str
page: intPydanticの BaseModel を使って実装
同値性 （
__eq__()）特殊メソッドを追加 （コレクションの例：
__iter__()でイテラブルに）
class Books(BaseModel, frozen=True):
values: list[Book]
def __iter__(self) -> Generator[Book, None, None]: # type: ignore[override]
yield from self.valuesusecase
usecase
同心円の内側から2番目
システムで 事業領域の課題を解決するロジック （＝アプリケーション）
メモリ上では完璧に構築（IOなどはないので使えない）
usecaseが知っているもの（内側のみ）
usecaseは内側のdomainを知っている
usecaseは port も知っている（メソッドは呼び出せる）
await self.fetch_books_port.fetch()port
port
インターフェース
class FetchBooksPort(ABC):
@abstractmethod
async def fetch(self) -> Books:
raise NotImplementedErrorusecaseは
portの 使い方を知っている （例：
fetch()メソッドはある！）中身がどんな実装をされているか（例：データ取得がファイルからかデータベースからか）は知らない
usecaseにportが渡る
class ListBooksUseCase:
def __init__(self, fetch_books_port: FetchBooksPort) -> None:
self.fetch_books_port = fetch_books_port作ると使うを一緒にせずに 分ける
usecaseは 使うだけ （依存性の逆転）
gateway
gateway
portを実装する
usecaseには（portを実装した）gatewayを渡す
class FetchBooksFromDatabase(FetchBooksPort):usecaseはgatewayに依存する
class ListBooksUseCase:
def execute(self) -> Books:
fetch_books = FetchBooksFromDatabase()
return await fetch_books.fetch()作ると使うを分ける
class ListBooksUseCase:
def __init__(self, fetch_books: FetchBooksFromDatabase) -> None:
self.fetch_books = fetch_books
def execute(self) -> Books:
return await self.fetch_books.fetch()依存性の逆転（DIP）
具体のgatewayではなく、port＝インターフェース（使い方） に依存 する
usecaseは詳細は知らなくても使える
class ListBooksUseCase:
def __init__(self, fetch_books_port: FetchBooksPort) -> None:
self.fetch_books_port = fetch_books_port
async def execute(self) -> Books:
return await self.fetch_books_port.fetch()gatewayでデータを 変換
ORM（SQLModel）をgatewayに置く
外界（プログラミング言語プリミティブな型で表現）から、ドメインの型に変換する
class BookRecord(SQLModel, table=True):class Book(BaseModel, frozen=True):さらにgatewayでもインターフェースを定義する
gateway自身はこのインターフェースを知っている
driverが実装する（再び登場、 依存性の逆転）
class DatabaseDriver(ABC):
@abstractmethod
async def select_books(self) -> list[BookRecord]:
raise NotImplementedErrordriver
driver
一番外側
外界に接する：HTTPリクエスト、DB
int,str,list,dictなど、プログラミング言語プリミティブな型で扱う（ドメインの型ではない）
gatewayのインターフェースを実装
class PostgresqlDatabaseDriver(DatabaseDriver):
async def select_books(self) -> list[BookRecord]:
# 初期化で渡されるセッションを使って、DBとやり取りrest
rest
これもまた外界：Web APIにして外界に接する
FastAPIのフレームワークの機能を活用 （HTTPリクエストを処理するcontrollerも兼ねる）
実装では、Web APIとして動かすための依存を渡した
usecaseを組み立てる
def inject_list_books_use_case(
session: Annotated[type[SQLModelAsyncSession], Depends(get_session)],
) -> ListBooksUseCase:
return ListBooksUseCase(FetchBooksFromDatabase(PostgresqlDatabaseDriver(session)))これをpath operation関数で実行
APIのレスポンスの型を定義
ドメインの型を直接は返さない
API利用者に見せるレスポンスの型を作る（たまたまドメインの型と同じ属性になることもある）
まとめ🥟 クリーンなアーキテクチャを志向したレイヤ分け
src/api
├── domain/ # システムによらないビジネスルールを表現
├── use_case/ # domainを操作して課題解決
├── port/ # インターフェース。use_caseが依存し、gatewayで実装
├── gateway/ # portを実装。データの変換。driver向けにインターフェースを定義
├── driver/ # gatewayのインターフェースを実装。DBやWeb APIと接する
└── rest/ # Webアプリ。今回はFastAPIの機能を使った上ほど同心円の内側
知識をレイヤに閉じ込めるために
工夫2つを共有（議論の種として）
1️⃣FastAPIのDepends
rest層でこれがやりたかった（再掲）
use_case を注入def inject_list_books_use_case(
session: Annotated[type[SQLModelAsyncSession], Depends(get_session)],
) -> ListBooksUseCase:
return ListBooksUseCase(FetchBooksFromDatabase(PostgresqlDatabaseDriver(session)))
@app.get("/books", response_model=list[BookReadModel])
async def get_books(
use_case: Annotated[ListBooksUseCase, Depends(inject_list_books_use_case)],
):
books = await use_case.execute()
return [BookReadModel.from_book(book) for book in books]FastAPIの Depends 🌟
https://fastapi.tiangolo.com/tutorial/dependencies/
async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
return {"q": q, "skip": skip, "limit": limit}
@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
return commonsAnnotated[..., Depends(...)]
typing.Annotated[<type>, <metadata>]メタデータを付与メタデータの
Depends(関数)は、 関数を実行して返り値を型ヒントしてる変数に代入IMO：動きが pytestのフィクスチャ っぽい
Dependsは 連鎖 する⛓️
https://fastapi.tiangolo.com/tutorial/dependencies/sub-dependencies/
def query_extractor(q: str | None = None):
# 省略
def query_or_cookie_extractor(
q: Annotated[str, Depends(query_extractor)],
last_query: Annotated[str | None, Cookie()] = None,
):
# 省略
@app.get("/items/")
async def read_query(
query_or_default: Annotated[str, Depends(query_or_cookie_extractor)],
):
# 省略Depends の連鎖を使ってDBセッションを注入
def inject_list_books_use_case(
session: Annotated[type[SQLModelAsyncSession], Depends(get_session)],
) -> ListBooksUseCase:async def get_session():
yield AsyncSession # モジュールスコープでasync_sessionmakerしている出典は rhoboro/async-fastapi-sqlalchemy
Sub-dependenciesに加えて Classes as Dependencies
@router.get("")
async def read_all(
use_case: ReadAllNote = Depends(ReadAllNote),
) -> ReadAllNoteResponse:class ReadAllNote:
def __init__(self, session: AsyncSession) -> None:
self.async_session = sessionasync def get_session() -> AsyncIterator[async_sessionmaker]:
...
AsyncSession = Annotated[async_sessionmaker, Depends(get_session)]FastAPIの Depends は積極的に活用できない
レイヤ分け を優先すると、
Dependsは domain や usecase には型ヒントできない（FastAPIはrest層の技術詳細）FastAPIは
Dependsを使い倒した方が開発者は楽ができそう（差し替え可能より優先する判断もあり得る）
2️⃣拡張関数を志向する
拡張関数（例：Kotlin）
継承しなくてもクラスにメソッドを追加できる
拡張する例
MutableList<Int> に swap メソッドを追加fun MutableList<Int>.swap(index1: Int, index2: Int) {
val tmp = this[index1] // 'this' corresponds to the list
this[index1] = this[index2]
this[index2] = tmp
}MutableList はcollectionの1つ
レイヤ分けにおける拡張関数の出番
gateway層：ドメインの型と、外界の型（言語プリミティブな型）の 変換
ドメインは、DBのレコードの型（同心円の外側）を知り得ない
gatewayでドメインを拡張する案
def books_from_records(book_records: list[BookRecord]) -> Books:
return Books(...)
Books.from_ = staticmethod(books_from_records) # type: ignore[attr-defined]gatewayにおけるBooksドメインに from_() メソッドを実行時に生やした
Python版 拡張関数
レイヤ間の関心の分離は達成！
実質はメタプログラミング 。型チェックで怒られ、コードジャンプに影響
他の方法は今は浮かんでいません...
お品書き：FastAPIの現場から
非同期IO
クリーンなアーキテクチャを志向する
Twelve-Factor App
Twelve-Factor App
このドキュメントは、多種多様なSaaSアプリケーション開発現場での私たちの経験と観察をすべてまとめたものである。
環境変数とロギングの2つについて取り上げます
1️⃣ III. 設定
設定を環境変数に格納する
設定を 環境変数 に
設定とは例えばデータベース接続
設定をコードから厳密に分離したい
環境変数を変えることで、 コードは変更しない が実現できる
pydantic-settings
https://docs.pydantic.dev/latest/concepts/pydantic_settings/
環境変数を読み込む + Pydanticによるパース
環境変数だけでなく
.envもサポート
pydantic-settingsでの設定例
from pydantic import PostgresDsn
from pydantic_settings import BaseSettings
class Config(BaseSettings):
pg_dsn: PostgresDsn
config = Config()環境変数 PG_DSN で設定できる
ネストも可能
DBの設定、LLM APIの設定と分けたい
class DB(BaseModel):
pg_dsn: PostgresDsn
class LlmApi(BaseModel):
api_key: str
class Config(BaseSettings):
model_config = SettingsConfigDict(env_nested_delimiter="__")
db: DB
llm: LlmApi
config = Config()環境変数 DB__PG_DSN や LLM__API_KEY
設定の使い方
from books_api.config import config
async_engine = create_async_engine(str(config.pg_dsn))
AsyncSession = async_sessionmaker(async_engine, class_=SQLModelAsyncSession)🏃♂️脱線：pydantic-settingsでCLIも作れます！
個人的イチオシポイント（さよならargparse）
2️⃣ XI. ログ
ログをイベントストリームとして扱う
ファイルではなく ストリーム
それぞれの実行中のプロセスはイベントストリームをstdout（標準出力）にバッファリングせずに書きだす。
k8s(GKE)のPodや、Google CloudのCloud Loggingで見ています
uvicornによるロギング
uvicorn app:app
--log-levelや --log-configfastapiコマンドより高機能なのでこちらを選択
Pythonのロギングの要素（抜粋）
ロガー
ハンドラ
フォーマッタ
2週前のPyCon mini Shizuoka 2024 continueで ロギングの話 をしました
デフォルトのlog config
uvicornロガー（標準エラー出力）
uvicorn.errorロガーからpropagate
uvicorn.accessロガー（標準出力）
色が付くのはuvicorn自前のフォーマッタによる
uvicorn.logging.DefaultFormatter
ログレコードの
levelprefixに色を付ける
uvicorn.logging.AccessFormatter
設定の仕方は標準ライブラリ logging.config より
version: 1
formatters:
default:
class: uvicorn.logging.DefaultFormatter
format: '%(levelprefix)s %(message)s'
use_colors: null
handlers:
default:
formatter: default
class: logging.StreamHandler
stream: ext://sys.stderr
loggers:
uvicorn:
handlers:
- default
level: INFO
propagate: false
uvicorn.error:
level: INFO
disable_existing_loggers: false設定例：発行されるSQLをログ出力
formatters:
default:
class: uvicorn.logging.DefaultFormatter
format: '%(asctime)s - %(name)s - %(levelprefix)s - %(message)s'
use_colors: null
handlers:
default_stdout:
formatter: default
class: logging.StreamHandler
stream: ext://sys.stdout
loggers:
sqlalchemy.engine:
handlers:
- default_stdout
level: INFO小まとめ🥟 Twelve-Factor App
pydantic-settings による、環境変数での設定
ロギングは logging.config に沿って uvicorn を設定
まとめ🌯 FastAPIの現場から
SQLModel（SQLAlchemy）で 全部 async def で書ける FastAPIアプリ！
レイヤ分け してビジネスロジックとフレームワークやDBを切り離したアーキテクチャ
環境変数から設定できる pydantic-settings。uvicornでのロギング