StrawberryでFastAPIにGraphQLを追加する—型ヒントで書くスキーマ

StrawberryでFastAPIにGraphQLを追加する—型ヒントで書くスキーマ | mohablog

GraphQLの利点は、クライアントが欲しいフィールドだけを1回で取れること。Strawberryはそのスキーマを、Pythonの型ヒントだけで組み立てます。エンドポイントがリソースごとに増えていくRESTの構成なら、置き換え候補になります。

目次

型ヒントがそのままGraphQLスキーマになる

Strawberryは@strawberry.typeを付けたクラスの型注釈を読み、GraphQLの型定義へ変換します。スキーマ定義言語(SDL)のファイルを別に書きません。現行は0.320.2系。動作要件はPython 3.10以上です(requires_python = "<4.0,>=3.10")。

RESTのover-fetchingを型で解く

RESTは1エンドポイントが固定のJSONを返します。一覧画面はタイトルだけ欲しいのに、本文まで返る。GraphQLはクエリ側でフィールドを選ぶので、over-fetchingが減ります。Strawberryはその選択肢を、サーバ側の型定義から自動で組み立てます。

スキーマファースト(gqlgen)との違い

GraphQLサーバの作り方は2系統。SDLを先に書くスキーマファーストと、コードから型を起こすコードファーストです。Goのgqlgenは前者。Strawberryは後者で、.graphqlファイルを持たず、Pythonの型がスキーマの正になります。

観点スキーマファーストコードファースト(Strawberry)
スキーマの正.graphqlファイルPythonの型定義
型の同期生成コマンドが要る常に一致する
覚えることSDLの文法型ヒントの知識で足りる

grapheneとの違い

Python GraphQLの老舗はgraphene。属性をgraphene.String()のように宣言する独自の書き方です。Strawberryは標準の型ヒントとdataclassに寄せています。既存のデータクラス定義を流用しやすいのはStrawberry。新規でも、型ヒントの知識だけで書き始められます。

最小構成で動かす

動くものから確かめます。FastAPI連携込みで入れるならpip install 'strawberry-graphql[fastapi]'

最小スキーマとGraphQLRouter

import strawberry
from fastapi import FastAPI
from strawberry.fastapi import GraphQLRouter


@strawberry.type
class Book:
    title: str
    author: str


@strawberry.type
class Query:
    @strawberry.field
    def books(self) -> list[Book]:
        return [Book(title="型ヒント入門", author="moha")]


schema = strawberry.Schema(query=Query)
graphql_app = GraphQLRouter(schema)

app = FastAPI()
app.include_router(graphql_app, prefix="/graphql")

GraphQLRouterstrawberry.fastapiのクラス。FastAPIのAPIRouterとして振る舞うので、include_routerでマウントするだけです。prefix="/graphql"を付けたぶん、エンドポイントは/graphqlになります。起動して確認。

uvicorn main:app --reload

実行結果。http://127.0.0.1:8000/graphqlにGraphiQL(ブラウザ用のクエリIDE)が立ち上がります。

GraphiQLで叩いて確かめる

{
  books {
    title
    author
  }
}

返ってくるJSON。

{
  "data": {
    "books": [
      { "title": "型ヒント入門", "author": "moha" }
    ]
  }
}

リレーションを持つ型を組む

実用ではオブジェクト同士がつながります。著者が複数の本を持つ、という関係。フィールドresolverで解決します。

@strawberry.type
class Author:
    id: int
    name: str

    @strawberry.field
    async def books(self) -> list[Book]:
        # 著者ごとに本を取りに行く
        return await fetch_books(self.id)

このbooksフィールドは、クエリでbooksを要求されたときだけ呼ばれます。要求されなければDBに触りません。使うフィールドの分だけクエリが走る仕組みです。ただしこの書き方、著者を100件並べると問題が出ます。

resolverは非同期を基本にする

公式ドキュメントは、全フィールドをasync defで書くことを勧めています。理由は、1ワーカーで同時リクエストをさばくため。DBアクセスを含むresolverはasyncにしておきます。

N+1問題をDataLoaderで潰す

さきほどのリレーション。authorsと、そのなかのbooksを一緒に要求すると何が起きるか。

{
  authors {
    name
    books { title }
  }
}

Strawberryはまず著者一覧を1回で取得。次に著者1件ごとにbooks resolverを呼びます。著者が100件なら、著者の1クエリ + 本の100クエリ。手元のSQLログで数えたら、まさに101回のクエリが飛んでいました。これがN+1です。

DataLoaderでキーをまとめて引く

DataLoaderは、同一イベントループ内で発生したload()呼び出しを1回のバッチにまとめます。100件のload(author_id)が、1回のWHERE author_id IN (...)に化けます。

from strawberry.dataloader import DataLoader


async def load_books(author_ids: list[int]) -> list[list[Book]]:
    rows = await db.fetch_books_in(author_ids)  # WHERE author_id IN (...)
    grouped: dict[int, list[Book]] = {aid: [] for aid in author_ids}
    for row in rows:
        grouped[row.author_id].append(Book(title=row.title, author=row.author))
    # 受け取ったキーと同じ順・同じ件数で返す
    return [grouped[aid] for aid in author_ids]


book_loader = DataLoader(load_fn=load_books)

resolver側はload()を呼ぶだけ。

@strawberry.field
async def books(self) -> list[Book]:
    return await book_loader.load(self.id)

実行結果。SQLログのクエリ数が101回から2回に落ちます。内訳は著者の取得1回と、本のバッチ1回。

load_fnがキー順を守らないと壊れる

DataLoaderの要は戻り値の並びです。公式のDataLoadersガイドは、load関数が渡されたキーと同じ順序・同じ件数で結果を返すことを求めています。DBはIN句に並べた順を保証しません。だから上のコードは[grouped[aid] for aid in author_ids]で、キー順に組み直しています。ここをrowsのDB順のまま返すと、著者Aの本が著者Bにぶら下がります。件数がずれれば例外に。

DataLoaderはリクエスト単位で作る

いまbook_loaderをモジュール直下のグローバルに置きました。動作確認にはこれで十分です。ただ本番だと、キャッシュがリクエストをまたいで残ります。あるユーザ向けの結果が、別ユーザのレスポンスに混ざる事故のもと。DataLoaderはリクエストごとに新しく作るのが基本です。その置き場がcontext_getterです。

context_getterで依存を渡す

DataLoader、DBセッション、認証済みユーザ。resolverから使いたい値はcontext_getterに集めます。FastAPIの依存注入がそのまま挿さります。

async def get_context():
    return {"book_loader": DataLoader(load_fn=load_books)}


graphql_app = GraphQLRouter(schema, context_getter=get_context)

resolverはstrawberry.Infoを受け取り、info.contextから読みます。

@strawberry.field
async def books(self, info: strawberry.Info) -> list[Book]:
    loader = info.context["book_loader"]
    return await loader.load(self.id)

get_contextはリクエストごとに1回呼ばれます。ここでDataLoaderを作れば、キャッシュはそのリクエストの内側で完結。バッチ化は効いたまま、キャッシュがリクエストをまたぐ問題だけが消えます。

Mutationと入力型

更新系は@strawberry.mutation。引数が増えるなら@strawberry.inputで束ねます。

@strawberry.input
class BookInput:
    title: str
    author: str


@strawberry.type
class Mutation:
    @strawberry.mutation
    def add_book(self, data: BookInput) -> Book:
        return Book(title=data.title, author=data.author)


schema = strawberry.Schema(query=Query, mutation=Mutation)

呼び出し側。

mutation {
  addBook(data: { title: "GraphQL実践", author: "moha" }) {
    title
    author
  }
}

実行結果。

{ "data": { "addBook": { "title": "GraphQL実践", "author": "moha" } } }

本番に出す前の設定

開発の初期値のまま公開すると、GraphiQLとGETクエリが開いたまま。GraphQLRouterの引数で締めます。

引数初期値本番での指定効果
graphql_ide"graphiql"Noneブラウザ用IDEを出さない
allow_queries_via_getTrueFalseGETでのクエリ実行を止める
context_getterなしリクエスト単位でDataLoader生成キャッシュのリクエスト間漏れを防ぐ

3つともGraphQLRouter(schema, ...)に渡すだけ。スキーマ全体を外から探索されたくないなら、introspectionも止めます。Strawberryはgraphql-coreの検証ルールで無効化できますが、GraphiQLもintrospectionに依存するのでgraphql_ide=Noneとセットで考えます。

まとめ

StrawberryはPythonの型ヒントをそのままGraphQLスキーマに使う、コードファーストのライブラリです。要点を並べます。

  • スキーマは型ヒントから生成。SDLファイルとの二重管理が要らない
  • FastAPI連携はGraphQLRouterinclude_routerでマウントするだけ
  • リレーションのresolverはasync defにし、N+1はDataLoaderでバッチ化する
  • load_fnは、渡されたキーと同じ順序・同じ件数で返す
  • DataLoaderはリクエスト単位で作り、context_getterに置く
  • 本番ではgraphql_ide=Noneallow_queries_via_get=Falseで口を締める
よかったらシェアしてね!
  • URLをコピーしました!
  • URLをコピーしました!
目次