python + asgi + graphqlまわりの試行錯誤用の最初のコードの作成。あるいは何かを学びたくなったときに最初に何をするかについて

最近はpython + asgi + grpahqlあたりで試行錯誤をすることが少しずつ増えてきた。その辺りのためのコードのメモ(gist)。

ついでに何か新しい領域を学ぼうと思ったとき、あるいは以前触れていたものの陳腐化してしまったとかんじている領域に対する知識のアップデートをしようと思ったときに何をするのかということをメモしようと思った。

graphql?

graphQLはクライアント側で返ってくるレスポンスの形式をクエリーを内に含めてリクエストできるようなエンドポイントを作成できるような何か。詳しくは色々な解説を読み漁ると良い。

• (暇だったらこの辺りに良さそう資料のURLを貼る)

仕様はこの辺にある。

• https://graphql.github.io/graphql-spec/

今回気にするのはpythonでのgraphQLについて。

情報収集 (依存関係の確認)

最初にすることはもちろん情報収集。ただそれぞれの詳しいやり方を調べる前にどのような依存関係があるかを調べることが多い気がする。あとエコシステム。

pythonでのgraphql用のライブラリについて

pythonでのgraphql用のライブラリについては大まかに以下の３つがある。

• graphene -- 老舗, graphql-coreと同じところが作っている, graphql-coreのまま
• ariadne -- 新興, asgiのinterface, schema first, graphql-core-nextが使われている
• strawberry -- こちらも新し目。あまり詳しくない。dataclassesにinspireされているらしい, graphql-core-nextが使われている

もう少しまともな比較はこの辺りを覗くと良い。

• https://blog.graphqleditor.com/top-3-python-libraries-for-graphql/

依存関係は以下の様な形

graphql-core-next -> ariadne, starwberry
graphql-core -> graphene

graphql-core, graphql-core-next?

pythonでのgraphqlの基礎部分のライブラリには、同じorganizationが作っているが、graphql-core-nextとgraphql-coreがあった。

現在はgraphql-core-nextはgraphql-coreに統合され。graphql-core自体はlegacyになった。

$ python -m pip install "graphql-core>=3a"

しかしエコシステム（特にgraphene）が完全にアップデートできているかと言うとそうではない模様。

なぜgraphql-core-nextを作ることになったのかの経緯はこの辺りでも書かれている。python3.6以降の機能を使って書きたいよねとか実装が陳腐化してしまっていて元のGraphQL.jsに追いつけてないとかそのあたりが理由。

graphql-core-nextのリポジトリのreadmeにそれぞれの対応状況についての記述がある。

• https://github.com/graphql-python/graphql-core-next#integration-with-other-libraries-and-roadmap

今回は新しいものを試したいのでgraphql-core-next上のものをみていくことにする。ariadneのあたりを試してみようと思った。

色々な粒度で実行を確認できるようにする

誰しもがある程度の雰囲気を掴んだらhello world的なものを進めていくとは思う。その後の動きは深さ優先的な人と幅優先的な人がいるように思う(雰囲気で話している)。個人的にはそのまま学ぼうとしている何かを使おうとしたり次に直接進む前に、色々な粒度での実行確認ができるような実験場を作ろうすることが多い。

理由は試行錯誤の手間が学習の妨げになるので。何かを確認したいと思った時にはちょうど良い粒度でのコードでの実行確認ができると良い。フットプリントが大きすぎては確認する手間を惜しむ様になっていく。そのような状態では細かな部分への理解が後々おぼつかなくなっていく。あるいは単純に不要な部分を切り分けることができなくなっていくような気がしている。そうなってくると何を学んで何を学ばなかったのかなど記録としての価値が喪われていってしまうような感覚がある。

この辺りについて色々と語りたいこともある気がするけれど。省略。

情報収集で分かったことを雑にまとめる

自分のための備忘録としてはここからが本題。大雑把に分かっていることをまとめる。今回はariadneを対象にした例。

• pythonのgraphqlライブラリにはgraphene,strawbery,ariadneがある
• より基礎的なライブラリとしてgraphql-core,graphql-core-nextがある
• ariadneはgraphql-core-nextの上に構築されている
• ariadneはasgi用のインターフェイスが用意されている
• asgiは非同期用のインターフェイスのこと。wsgiの非同期版(wsgiを知らなければwsgiも知っておく)
• asgiを試すのにはuvicorn辺りが手軽
• そうは言っても常にasgi込みで実行するのはめんどくさい

1ファイルで実行したい

色々な粒度のものを1ファイルで実行したい。なぜ1ファイルかというとこれから数十回くらいは試すことになるので。そのたびにプロジェクトやらを作っているのはバカバカしい。cookiecutterなどのプロジェクトテンプレートで生成を手軽にしたとしてファイルが不用意に増えていくのは把握がし辛い。

基本は以下のどちらか

$ python <target file>.py
$ pytest <target file> # もしくは単にpytest

引数が欲しくなるときのためにタスクランナーやMakefileなどで指定できるようにしておく。

Makefile

# hello world
00:
  # python 00hello.py とか書くのが面倒なので
    python $@*.py

# aggressive
01:
    python $@*.py --aggressive

# pytest
02:
    pytest $@* -vv

コマンドラインの引数や実行方法を意識せずに色々な例を試したいという気持ちでの作業なのでMakefileである必要はない。他の手に馴染んだものを使えば良い。実際その時の気分で利用するものは変える。

どのような粒度で実行したいか考える

例えば以下の様な粒度がありそう。先程の情報収集の結果を元に洗い出してみる。あるいはフルの状態のコードを元にコレを外せないか？アレが面倒なのだけど？？とやっていくのも良い。ちょうどよく手軽だと感じた辺りで辞める。

• 基礎的な部分だけで実行 (graphql-core-next)
• ラッパーライブラリも含めて実行 (ariadne)
• 実際に利用するときの形式だけで実行 (asgi appのhello world)
• 実際に利用するときに形式に繋げた形で実行 (graphqlをasgi appとして実行)
• E2E的に実行 (動作チェックはシェルスクリプトになる（今回は範囲外）)

ariadneに関してはこの辺りを一度揃えておくと便利そう（最初から綺麗に考えられる事は少ないかもしれない。実際のところは試行錯誤の後の後付だったりする）。

requirements.txt

graphql-core>=3a
graphene
ariadne
uvicorn
async-asgi-testclient
pytest
pytest-asyncio

基礎的な部分だけで実行 (graphql-core-next)

同期版 00graphql-core-next.py

from graphql import GraphQLSchema, GraphQLObjectType, GraphQLField, GraphQLString

schema = GraphQLSchema(
    query=GraphQLObjectType(
        name="RootQueryType",
        fields={
            "hello": GraphQLField(GraphQLString, resolve=lambda obj, info: "world")
        },
    )
)


def main() -> None:
    from graphql import graphql_sync

    query = "{ hello }"

    print(graphql_sync(schema, query))


if __name__ == "__main__":
    main()

output

$ python 00graphql-core-next.py
ExecutionResult(data={'hello': 'world'}, errors=None)

非同期版

import asyncio
from graphql import GraphQLSchema, GraphQLObjectType, GraphQLField, GraphQLString


async def resolve_hello(obj, info):
    print("start resolve hello")
    await asyncio.sleep(1)
    print("finish resolve hello")
    return "world"


schema = GraphQLSchema(
    query=GraphQLObjectType(
        name="RootQueryType",
        fields={"hello": GraphQLField(GraphQLString, resolve=resolve_hello)},
    )
)


def main() -> None:
    from graphql import graphql

    async def run():
        query = "{ hello }"
        print(await graphql(schema, query))

    asyncio.run(run(), debug=True)


if __name__ == "__main__":
    main()

output

$ python 01graphql-core-next.py
start resolve hello
finish resolve hello
ExecutionResult(data={'hello': 'world'}, errors=None)

ラッパーライブラリも含めて実行 (ariadne)

02run-ariadne.py

import asyncio
import logging
from ariadne import ObjectType, gql, make_executable_schema, graphql


# Ariadne uses dedicated objects
query = ObjectType("Query")

# Map resolvers to fields in Query type using decorator syntax...
@query.field("hello")
def resolve_hello(_, info):
    request = info.context["request"]
    user_agent = request.get("HTTP_USER_AGENT", "guest")
    return "Hello, %s!" % user_agent


type_defs = gql(
    """
    type Query {
        hello: String!
    }
"""
)


logging.basicConfig(level=logging.DEBUG)
schema = make_executable_schema(type_defs, query)

query = {"query": "{ hello }"}
atask = graphql(
    schema, query, debug=True, context_value={"request": {"HTTP_USER_AGENT": "World"}}
)
print(asyncio.run(atask, debug=True))

output

$ python 02run-ariadne.py
DEBUG:asyncio:Using selector: EpollSelector
DEBUG:asyncio:Close <_UnixSelectorEventLoop running=False closed=False debug=True>
(True, {'data': {'hello': 'Hello, World!'}})

実際に利用するときの形式だけで実行 (asgi appのhello world)

少し寄り道。asgi自体のテストの仕方も整理しておかないとめんどくさい。starletteなどのasgi用のフレームワークを使っている場合にはそのフレームワークが提供しているテストクライアントを使えば良いと言えば良いのだけれど。そうなると1ファイルの部分で説明したように依存が複雑になる。それは避けたいので別のライブラリを使う

• https://github.com/vinissimus/async-asgi-testclient

The motivation behind this project is building a common testing library that doesn't depend on the web framework (Quart, Startlette, ...).

03test_asgi.py

import pytest
import async_asgi_testclient as aat


async def app(scope, receive, send):
    # assert scope["type"] == "http"
    await send(
        {
            "type": "http.response.start",
            "status": 200,
            "headers": [[b"content-type", b"text/plain"]],
        }
    )
    await send({"type": "http.response.body", "body": b"Hello, world!"})


@pytest.mark.asyncio
async def test_app():
    async with aat.TestClient(app) as client:
        resp = await client.get("/")
        assert resp.status_code == 200
        assert resp.text == "Hello, world!"

output

$ pytest -vv 03test_asgi.py
===================================== test session starts =====================================
platform linux -- Python 3.7.4, pytest-5.2.1, py-1.7.0, pluggy-0.12.0 -- $HOME/venvs/my/bin/python
cachedir: .pytest_cache
rootdir: $HOME/venvs/my/individual-sandbox/daily/20191021/example_ariadne
plugins: cov-2.6.1, asyncio-0.10.0
collected 1 item                                                                              

03test_asgi.py::test_app PASSED                                                         [100%]

====================================== 1 passed in 0.05s ======================================

実際に利用するときに形式に繋げた形で実行 (graphqlをasgi appとして実行)

繋げた形もテストコードの形式になっていると便利。

04test_ariadne.py

import pytest
from async_asgi_testclient import TestClient

from ariadne import ObjectType, gql, make_executable_schema
from ariadne.asgi import GraphQL

TestClient.__test__ = False  # prevent PytestCollectionWarning

# Ariadne uses dedicated objects
query = ObjectType("Query")

# Map resolvers to fields in Query type using decorator syntax...
@query.field("hello")
def resolve_hello(_, info):
    request = info.context["request"]
    user_agent = request.get("HTTP_USER_AGENT", "guest")
    return "Hello, %s!" % user_agent


type_defs = gql(
    """
    type Query {
        hello: String!
    }
"""
)

schema = make_executable_schema(type_defs, query)
# As standalone ASGI or WSGI app...
app = GraphQL(schema, debug=True)


@pytest.mark.asyncio
async def test_app():
    client = TestClient(app)

    resp = await client.post("/", json={"query": "{ hello }"})
    assert resp.status_code == 200
    assert resp.json() == {"data": {"hello": "Hello, guest!"}}

output

$ pytest -vv 04test_ariadne.py
===================================== test session starts =====================================
platform linux -- Python 3.7.4, pytest-5.2.1, py-1.7.0, pluggy-0.12.0 -- /home/nao/venvs/my/bin/python
cachedir: .pytest_cache
rootdir: /home/nao/venvs/my/individual-sandbox/daily/20191021/example_ariadne
plugins: cov-2.6.1, asyncio-0.10.0
collected 1 item                                                                              

04test_ariadne.py::test_app PASSED                                                      [100%]

====================================== 1 passed in 0.16s ======================================

おしまい。

手元で実行したいときのためのgist

参考

pythonとは直接関係ないけれどschema-firstのアプローチは嬉しいの嬉しくないのに対する1つの意見。

• https://www.prisma.io/blog/the-problems-of-schema-first-graphql-development-x1mn4cb0tyl3

strawberry, ariadne, grapheneの比較。

• https://blog.graphqleditor.com/top-3-python-libraries-for-graphql/

API作成系のことについてのとても頑張っているメモ

• https://gdevops.gitlab.io/tuto_http/api/api.html